Add ND 4.2 Links module (nd_links)

plugins/module_utils/endpoints/mixins.py

@@ -191,3 +191,24 @@ class PolicyGroupIdMixin(BaseModel):
         min_length=1,
         description="Policy Group ID (e.g., POLICY-GROUP-143310)",
     )
+
+
+class SrcClusterNameMixin(BaseModel):
+    """Mixin for endpoints that require src_cluster_name parameter."""
+
+    src_cluster_name: str | None = Field(default=None, min_length=1, description="Source cluster name")
+
+
+class DstClusterNameMixin(BaseModel):
+    """Mixin for endpoints that require dst_cluster_name parameter."""
+
+    dst_cluster_name: str | None = Field(default=None, min_length=1, description="Destination cluster name")
+
+
+class IsLogicalLinkMixin(BaseModel):
+    """Mixin for endpoints that accept isLogicalLink parameter."""
+
+    is_logical_link: BooleanStringEnum = Field(
+        default=BooleanStringEnum.FALSE,
+        description="Indicates if the link is a logical link",
+    )

plugins/module_utils/endpoints/query_params.py

@@ -140,6 +140,30 @@ def is_empty(self) -> bool:
         return len(self.model_dump(exclude_none=True, exclude_defaults=True)) == 0
 
 
+class UrlEncodedEndpointQueryParams(EndpointQueryParams):
+    """EndpointQueryParams whose ``to_query_string`` percent-encodes values.
+
+    The base class emits raw values; this variant URL-encodes each value so an
+    identifier containing reserved characters (for example a ticket id, fabric,
+    cluster or switch name with ``&``, ``=``, ``+``, ``#`` or a space) cannot
+    alter how the controller parses the query string. Field names are still
+    converted snake_case -> camelCase.
+    """
+
+    def to_query_string(self) -> str:
+        params = []
+        for field_name, field_value in self.model_dump(exclude_none=True).items():
+            api_key = self._to_camel_case(field_name)
+            if isinstance(field_value, bool):
+                api_value = str(field_value).lower()
+            elif isinstance(field_value, Enum):
+                api_value = field_value.value
+            else:
+                api_value = str(field_value)
+            params.append("{0}={1}".format(api_key, quote(api_value, safe="")))
+        return "&".join(params)
+
+
 class LuceneQueryParams(BaseModel):
     """
     # Summary

plugins/module_utils/endpoints/v1/manage/base_path.py

@@ -76,3 +76,7 @@ def path(cls, *segments: str) -> str:
         if not segments:
             return cls.API
         return f"{cls.API}/{'/'.join(segments)}"
+
+
+# Alias retained for links endpoints that import ``BasePathLinks`` directly.
+BasePathLinks = BasePath

plugins/module_utils/endpoints/v1/manage/link_actions.py

@@ -0,0 +1,29 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+from typing import Literal
+
+from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import Field
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.base import NDEndpointBaseModel
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.base_path import BasePathLinks
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.links import LinkMutateParams
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+
+
+class LinkActionsRemovePost(NDEndpointBaseModel):
+    """POST /api/v1/manage/linkActions/remove for bulk delete by linkId."""
+
+    class_name: Literal["LinkActionsRemovePost"] = Field(default="LinkActionsRemovePost", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePathLinks.path("linkActions", "remove")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST

plugins/module_utils/endpoints/v1/manage/links.py

@@ -0,0 +1,79 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+from typing import Literal
+
+from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import Field
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.base import NDEndpointBaseModel
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.mixins import (
+    ClusterNameMixin,
+    FabricNameMixin,
+    LinkUuidMixin,
+    SwitchIdMixin,
+    TicketIdMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.query_params import UrlEncodedEndpointQueryParams
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+
+from .base_path import BasePathLinks
+
+
+class LinksGetParams(FabricNameMixin, ClusterNameMixin, TicketIdMixin, SwitchIdMixin, UrlEncodedEndpointQueryParams):
+    """URL-encoded query params for GET /links (single cluster): fabricName, clusterName, ticketId, switchId."""
+
+
+class LinkMutateParams(ClusterNameMixin, TicketIdMixin, UrlEncodedEndpointQueryParams):
+    """URL-encoded query params for link create/update/delete (single cluster): clusterName, ticketId."""
+
+
+class LinksGet(NDEndpointBaseModel):
+    """GET /api/v1/manage/links for single cluster scope."""
+
+    class_name: Literal["LinksGet"] = Field(default="LinksGet", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinksGetParams = Field(default_factory=LinksGetParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePathLinks.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.GET
+
+
+class LinksPost(NDEndpointBaseModel):
+    """POST /api/v1/manage/links for bulk create."""
+
+    class_name: Literal["LinksPost"] = Field(default="LinksPost", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePathLinks.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST
+
+
+class LinkPut(LinkUuidMixin, NDEndpointBaseModel):
+    """PUT /api/v1/manage/links/{linkId} for single update (linkId set via link_uuid)."""
+
+    class_name: Literal["LinkPut"] = Field(default="LinkPut", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePathLinks.path("links", self.link_uuid) if self.link_uuid else BasePathLinks.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.PUT

plugins/module_utils/endpoints/v1/one_manage/base_path.py

@@ -0,0 +1,22 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Final
+
+
+class BasePath:
+    """Builds absolute paths anchored at /api/v1/manage for multi cluster link endpoints."""
+
+    API: Final = "/api/v1/manage"
+
+    @classmethod
+    def path(cls, *segments: str) -> str:
+        """Join ``segments`` onto the base API path; returns the bare base if none given."""
+        if not segments:
+            return cls.API
+        return "{0}/{1}".format(cls.API, "/".join(segments))

plugins/module_utils/endpoints/v1/one_manage/link_actions.py

@@ -0,0 +1,29 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+from typing import Literal
+
+from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import Field
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.base import NDEndpointBaseModel
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.one_manage.base_path import BasePath
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.one_manage.links import LinkMutateParams
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+
+
+class LinkActionsRemovePost(NDEndpointBaseModel):
+    """POST /api/v1/manage/linkActions/remove for bulk delete (multi cluster)."""
+
+    class_name: Literal["LinkActionsRemovePost"] = Field(default="LinkActionsRemovePost", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePath.path("linkActions", "remove")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST

plugins/module_utils/endpoints/v1/one_manage/links.py

@@ -0,0 +1,79 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+from typing import Literal
+
+from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import Field
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.base import NDEndpointBaseModel
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.mixins import (
+    DstClusterNameMixin,
+    FabricNameMixin,
+    LinkUuidMixin,
+    SrcClusterNameMixin,
+    TicketIdMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.query_params import UrlEncodedEndpointQueryParams
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+
+from .base_path import BasePath
+
+
+class LinksGetParams(FabricNameMixin, SrcClusterNameMixin, DstClusterNameMixin, UrlEncodedEndpointQueryParams):
+    """URL-encoded query params for GET /links (multi cluster): fabricName, srcClusterName, dstClusterName."""
+
+
+class LinkMutateParams(TicketIdMixin, UrlEncodedEndpointQueryParams):
+    """URL-encoded query params for link create/update/delete (multi cluster): ticketId."""
+
+
+class LinksGet(NDEndpointBaseModel):
+    """GET /api/v1/manage/links for multi cluster scope."""
+
+    class_name: Literal["LinksGet"] = Field(default="LinksGet", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinksGetParams = Field(default_factory=LinksGetParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePath.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.GET
+
+
+class LinksPost(NDEndpointBaseModel):
+    """POST /api/v1/manage/links for bulk create (multi cluster)."""
+
+    class_name: Literal["LinksPost"] = Field(default="LinksPost", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePath.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST
+
+
+class LinkPut(LinkUuidMixin, NDEndpointBaseModel):
+    """PUT /api/v1/manage/links/{linkId} for single update (multi cluster; linkId set via link_uuid)."""
+
+    class_name: Literal["LinkPut"] = Field(default="LinkPut", frozen=True, description="Class name for backward compatibility")
+    endpoint_params: LinkMutateParams = Field(default_factory=LinkMutateParams, description="Endpoint-specific query parameters")
+
+    @property
+    def path(self) -> str:
+        base = BasePath.path("links", self.link_uuid) if self.link_uuid else BasePath.path("links")
+        query_string = self.endpoint_params.to_query_string()
+        return "{0}?{1}".format(base, query_string) if query_string else base
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.PUT

plugins/module_utils/fabric_inventory_helpers.py

@@ -0,0 +1,91 @@
+# Copyright: (c) 2026, Shreyas Srish (@shrsr) <ssrish@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+"""
+Helpers that extend FabricSwitchInventory with the by-name lookup and
+priority-based resolve() semantics that the upstream class does not
+provide. Kept here (not in fabric_inventory.py) so the vendored upstream
+file stays byte-identical to the source PR.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from ansible_collections.cisco.nd.plugins.module_utils.fabric_inventory import FabricSwitchInventory
+from ansible_collections.cisco.nd.plugins.module_utils.models.manage_switches.switch_data_models import SwitchDataModel
+
+
+class _FabricInventorySenderAdapter:
+    """Bridge RestSend to the ``nd.request(verb=...)`` / ``nd.module.fail_json`` /
+    ``nd.rest_send.response_current`` shape that FabricSwitchInventory expects."""
+
+    def __init__(self, rest_send: Any) -> None:
+        self.rest_send = rest_send
+        self.module = rest_send.sender.ansible_module
+
+    def request(self, path: str, verb: str | None = None, **_kwargs: Any) -> Any:
+        self.rest_send.path = path
+        self.rest_send.verb = verb
+        self.rest_send.commit()
+        if not self.rest_send.success:
+            raise Exception("Request failed {0}".format(self.rest_send.error_summary))
+        return self.rest_send.response_current.get("DATA", {})
+
+
+def inventory_for_fabric(rest_send: Any, fabric_name: str, log: logging.Logger) -> FabricSwitchInventory:
+    """Build a FabricSwitchInventory for a fabric using RestSend + SwitchDataModel."""
+    adapter = _FabricInventorySenderAdapter(rest_send)
+    return FabricSwitchInventory.from_fabric(adapter, fabric_name, log, SwitchDataModel)
+
+
+def by_name(inventory: FabricSwitchInventory) -> dict[str, list[Any]]:
+    """Return switches keyed by hostname. Value is a list to expose collisions."""
+    result: dict[str, list[Any]] = {}
+    for sw in inventory.switches:
+        name = getattr(sw, "hostname", None)
+        if name:
+            result.setdefault(name, []).append(sw)
+    return result
+
+
+def resolve(
+    inventory: FabricSwitchInventory,
+    switch_id: str | None = None,
+    switch_ip: str | None = None,
+    switch_name: str | None = None,
+    fabric_name: str | None = None,
+    side: str = "",
+) -> str | None:
+    """Priority: switch_id > switch_ip > switch_name. Raises on miss or ambiguous name."""
+    if switch_id:
+        return switch_id
+
+    side_prefix = "{0}_".format(side) if side else ""
+    fabric_suffix = " in fabric '{0}'".format(fabric_name) if fabric_name else ""
+
+    if switch_ip:
+        sw = inventory.by_ip().get(switch_ip)
+        if not sw:
+            raise Exception(
+                "Could not resolve {0}switch_ip='{1}'{2}. " "No switch with that management IP was found.".format(side_prefix, switch_ip, fabric_suffix)
+            )
+        return sw.switch_id
+
+    if switch_name:
+        matches = by_name(inventory).get(switch_name, [])
+        if len(matches) == 1:
+            return matches[0].switch_id
+        if len(matches) > 1:
+            ids = [m.switch_id for m in matches]
+            raise Exception(
+                "{0}switch_name='{1}' is ambiguous{2} (matches {3} switches: {4}). "
+                "Use {0}switch_ip or {0}switch_id to disambiguate.".format(side_prefix, switch_name, fabric_suffix, len(matches), ", ".join(ids))
+            )
+        raise Exception(
+            "Could not resolve {0}switch_name='{1}'{2}. " "No switch with that hostname was found.".format(side_prefix, switch_name, fabric_suffix)
+        )
+
+    return None