rename_duplicate_dals.RelationshipCache Class Reference

Collaboration diagram for rename_duplicate_dals.RelationshipCache:

Public Member Functions
	__init__ (self, Configuration|str db)
List[DalBase]	get_parents (self, DalBase obj)

Public Attributes
	db = db
Dict[DalBase, List[DalBase]]	graph = self._build_graph()

Protected Member Functions
	_relations (self, str class_name)
Dict[DalBase, List[DalBase]]	_build_graph (self)

Protected Attributes
dict	_relations_cache = {}
dict	_parents_cache = {}

Detailed Description

Caches parent-child relationships between DAL objects to avoid repeated DB calls.

Definition at line 31 of file rename_duplicate_dals.py.

Constructor & Destructor Documentation

__init__()

rename_duplicate_dals.RelationshipCache.__init__	(		self,
		Configuration | str	db )

Definition at line 34 of file rename_duplicate_dals.py.

    def __init__(self, db: Configuration | str):
        if isinstance(db, str):
            db = Configuration("oksconflibs:" + db)
        self.db = db
        self._relations_cache: Dict[str, Any] = {}
        self._parents_cache: Dict[Any, list] = {}
        self.graph = self._build_graph()
 

Member Function Documentation

_build_graph()

Dict[DalBase, List[DalBase]] rename_duplicate_dals.RelationshipCache._build_graph ( self )

protected

Definition at line 47 of file rename_duplicate_dals.py.

    def _build_graph(self) -> Dict[DalBase, List[DalBase]]:
        graph = {}
        for obj in self.db.get_all_dals().values():
            children = []
            for rel in self._relations(obj.className()):
                related = getattr(obj, rel, None)
                if related is None:
                    continue
                children.extend(related if isinstance(related, list) else [related])
            if children:
                graph[obj] = children
        return graph
 

_relations()

rename_duplicate_dals.RelationshipCache._relations ( self, str class_name )

protected

Definition at line 42 of file rename_duplicate_dals.py.

    def _relations(self, class_name: str):
        if class_name not in self._relations_cache:
            self._relations_cache[class_name] = self.db.relations(class_name, all=True)
        return self._relations_cache[class_name]
 

get_parents()

List[DalBase] rename_duplicate_dals.RelationshipCache.get_parents	(		self,
		DalBase	obj )

Definition at line 60 of file rename_duplicate_dals.py.

    def get_parents(self, obj: DalBase) -> List[DalBase]:
        if obj not in self._parents_cache:
            self._parents_cache[obj] = [
                parent for parent, children in self.graph.items() if obj in children
            ]
        return self._parents_cache[obj]
 
 
# ----- Extended DAL (wraps a DAL with its config + tree context) -----
 

Member Data Documentation

_parents_cache

dict rename_duplicate_dals.RelationshipCache._parents_cache = {}

protected

Definition at line 39 of file rename_duplicate_dals.py.

_relations_cache

dict rename_duplicate_dals.RelationshipCache._relations_cache = {}

protected

Definition at line 38 of file rename_duplicate_dals.py.

db

rename_duplicate_dals.RelationshipCache.db = db

Definition at line 37 of file rename_duplicate_dals.py.

graph

Dict[DalBase, List[DalBase]] rename_duplicate_dals.RelationshipCache.graph = self._build_graph()

Definition at line 40 of file rename_duplicate_dals.py.

---

The documentation for this class was generated from the following file:

• /github/workspace/dunedaq/sourcecode/daqconf/python/daqconf/rename_duplicate_dals.py