Add tag pool for links + fix potential consistency and dead lock issues

CHANGELOG.rst

@@ -7,6 +7,14 @@ All notable changes to the ``topology`` project will be documented in this file.
 [UNRELEASED] - Under development
 ********************************
 
+Changed
+=======
+- Added tag pool for ``Links``, separate from ``Interfaces``.
+
+Fixed
+=====
+- Fixed potential race conditions in multithreaded code.
+
 [2025.2.0] - 2026-02-02
 ***********************
 

controllers/__init__.py

@@ -35,7 +35,7 @@
 class TopoController:
     """TopoController."""
 
-    def __init__(self, get_mongo=lambda: Mongo()) -> None:
+    def __init__(self, get_mongo=Mongo) -> None:
         """Constructor of TopoController."""
         self.mongo = get_mongo()
         self.db_client = self.mongo.client
@@ -156,6 +156,19 @@ def disable_interface(self, interface_id: str) -> Optional[dict]:
             interface_id, {"$set": {"enabled": False}}
         )
 
+    def delete_interface(self, interface_id: str) -> Optional[dict]:
+        """Try to delete an interface embedded in a switch."""
+        switch_id, _, port_num = interface_id.rpartition(":")
+        port_num = int(port_num)
+        return self.db.switches.find_one_and_update(
+            {"_id": switch_id},
+            {"$unset": {"interfaces.$[iface]": 1}},
+            array_filters=[{
+                "iface.port_number": {"$eq": port_num}
+            }],
+            return_document=ReturnDocument.AFTER
+        )
+
     def add_interface_metadata(
         self, interface_id: str, metadata: dict
     ) -> Optional[dict]:
@@ -189,6 +202,69 @@ def _update_interface(
             return_document=ReturnDocument.AFTER,
         )
 
+    def enable_interfaces(
+        self,
+        switch_id: str,
+        ports: list[int]
+    ):
+        """Try to enable several interfaces."""
+        return self._update_interfaces_of_switch(
+            switch_id, ports, {"$set": {"enabled": True}}
+        )
+
+    def disable_interfaces(
+        self,
+        switch_id: str,
+        ports: list[int]
+    ):
+        """Try to disable several interfaces."""
+        return self._update_interfaces_of_switch(
+            switch_id, ports, {"$set": {"enabled": False}}
+        )
+
+    def enable_interfaces_lldp(
+        self,
+        switch_id: str,
+        ports: list[int]
+    ):
+        """Try to enable lldp on several interfaces."""
+        return self._update_interfaces_of_switch(
+            switch_id, ports, {"$set": {"lldp": True}}
+        )
+
+    def disable_interfaces_lldp(
+        self,
+        switch_id: str,
+        ports: list[int]
+    ):
+        """Try to disable lldp on several interfaces."""
+        return self._update_interfaces_of_switch(
+            switch_id, ports, {"$set": {"lldp": False}}
+        )
+
+    def _update_interfaces_of_switch(
+        self,
+        switch_id: str,
+        ports: list[int],
+        update_expr: dict
+    ):
+        self._set_updated_at(update_expr)
+        interfaces_expression = {}
+        for operator, values in update_expr.items():
+            interfaces_expression[operator] = {
+                f"interfaces.$[iface].{key}": value
+                for key, value in values.items()
+            }
+
+        return self.db.switches.find_one_and_update(
+            {"_id": switch_id},
+            interfaces_expression,
+            array_filters=[{
+                "iface.port_number": {"$in": ports}
+            }],
+            return_document=ReturnDocument.AFTER
+        )
+
     def upsert_link(self, link_id: str, link_dict: dict) -> dict:
         """Update or insert a Link."""
         utc_now = datetime.utcnow()
@@ -299,17 +375,23 @@ def upsert_interface_details(
         id_: str,
         available_tags: dict[str, list[list[int]]],
         tag_ranges: dict[str, list[list[int]]],
+        default_tag_ranges: dict[str, list[list[int]]],
         special_available_tags: dict[str, list[str]],
-        special_tags: dict[str, list[str]]
+        special_tags: dict[str, list[str]],
+        default_special_tags: dict[str, list[str]],
+        supported_tag_types: list[str],
     ) -> Optional[dict]:
         """Update or insert interfaces details."""
         utc_now = datetime.utcnow()
         model = InterfaceDetailDoc(**{
                 "_id": id_,
                 "available_tags": available_tags,
                 "tag_ranges": tag_ranges,
+                "default_tag_ranges": default_tag_ranges,
                 "special_available_tags": special_available_tags,
                 "special_tags": special_tags,
+                "default_special_tags": default_special_tags,
+                "supported_tag_types": supported_tag_types,
                 "updated_at": utc_now
         }).model_dump(exclude={"inserted_at"})
         updated = self.db.interface_details.find_one_and_update(
@@ -358,3 +440,55 @@ def delete_interface_from_details(self, intf_id: str) -> Optional[dict]:
         return self.db.interface_details.find_one_and_delete(
             {"_id": intf_id}
         )
+
+    # pylint: disable=too-many-arguments
+    def upsert_link_details(
+        self,
+        id_: str,
+        available_tags: dict[str, list[list[int]]],
+        tag_ranges: dict[str, list[list[int]]],
+        default_tag_ranges: dict[str, list[list[int]]],
+        special_available_tags: dict[str, list[str]],
+        special_tags: dict[str, list[str]],
+        default_special_tags: dict[str, list[str]],
+        supported_tag_types: list[str],
+    ) -> Optional[dict]:
+        """Update or insert link details."""
+        utc_now = datetime.utcnow()
+        model = InterfaceDetailDoc(**{
+                "_id": id_,
+                "available_tags": available_tags,
+                "tag_ranges": tag_ranges,
+                "default_tag_ranges": default_tag_ranges,
+                "special_available_tags": special_available_tags,
+                "special_tags": special_tags,
+                "default_special_tags": default_special_tags,
+                "supported_tag_types": supported_tag_types,
+                "updated_at": utc_now
+        }).model_dump(exclude={"inserted_at"})
+        updated = self.db.link_details.find_one_and_update(
+            {"_id": id_},
+            {
+                "$set": model,
+                "$setOnInsert": {"inserted_at": utc_now},
+            },
+            return_document=ReturnDocument.AFTER,
+            upsert=True,
+        )
+        return updated
+
+    def get_links_details(
+        self, link_ids: List[str]
+    ) -> Optional[dict]:
+        """Try to get link details given a list of link ids."""
+        return self.db.link_details.aggregate(
+            [
+                {"$match": {"_id": {"$in": link_ids}}},
+            ]
+        )
+
+    def delete_link_from_details(self, link_id: str) -> Optional[dict]:
+        """Delete link from link_details."""
+        return self.db.link_details.find_one_and_delete(
+            {"_id": link_id}
+        )

db/models/__init__.py

@@ -128,5 +128,8 @@ class InterfaceDetailDoc(DocumentBaseModel):
 
     available_tags: Dict[str, list[list[int]]]
     tag_ranges: Dict[str, list[list[int]]]
+    default_tag_ranges: dict[str, list[list[int]]]
     special_available_tags: Dict[str, list[str]]
     special_tags: Dict[str, list[str]]
+    default_special_tags: dict[str, list[str]]
+    supported_tag_types: list[str]