Vrf_lite module initial draft

plugins/module_utils/common/data.py

@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, annotations, division, print_function
+
+import ast
+import json
+from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
+
+Parser = Callable[[str], Any]
+DEFAULT_VALUE_PARSERS: tuple[Parser, ...] = (json.loads, ast.literal_eval)
+
+
+def get_params(source: Any) -> dict[str, Any]:
+    """Return a mutable params mapping from either module.params or a raw params dict."""
+    if isinstance(source, dict):
+        return source
+
+    params = getattr(source, "params", None)
+    if isinstance(params, dict):
+        return params
+
+    return {}
+
+
+def parse_value(value: Any, parsers: Sequence[Parser] = DEFAULT_VALUE_PARSERS, default: Any = None) -> Any:
+    """Parse a serialized value with the first parser that accepts it.
+
+    Dicts and lists are returned unchanged because callers often pass values
+    that were already decoded by the controller client. Empty, invalid, or
+    unprintable values return ``default``.
+    """
+    if isinstance(value, (dict, list)):
+        return value
+    if value is None:
+        return default
+
+    try:
+        text = str(value).strip()
+    except Exception:
+        return default
+
+    if not text:
+        return default
+
+    for parser in parsers:
+        try:
+            return parser(text)
+        except Exception:
+            continue
+
+    return default
+
+
+def coerce_dict_list(data: Any, list_keys: Sequence[str] = ("DATA", "data", "items")) -> list[dict[str, Any]]:
+    """Return dict items from common shallow controller response shapes.
+
+    This intentionally checks only the top-level value or one configured
+    top-level wrapper key. Deeper response shapes should be handled by the
+    caller because those paths are resource-specific.
+    """
+    if isinstance(data, list):
+        return [item for item in data if isinstance(item, dict)]
+
+    if isinstance(data, dict):
+        for key in list_keys:
+            value = data.get(key)
+            if isinstance(value, list):
+                return [item for item in value if isinstance(item, dict)]
+
+    return []
+
+
+def copy_dict_items(items: Optional[Iterable[Any]]) -> list[dict[str, Any]]:
+    """Copy a list of dict-like or pydantic-like objects into plain dicts."""
+    copied = []
+    for item in items or []:
+        if isinstance(item, Mapping):
+            copied.append(dict(item))
+        elif hasattr(item, "model_dump"):
+            copied.append(item.model_dump(by_alias=False, exclude_none=True))
+    return copied
+
+
+def try_int(value: Any) -> Optional[int]:
+    """Best-effort integer conversion."""
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None

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

@@ -0,0 +1,37 @@
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Centralized base paths for top-down and resource-manager NDFC endpoints.
+"""
+
+from __future__ import absolute_import, annotations, division, print_function
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Final
+
+
+class TopDownBasePath:
+    """Base path helper for top-down lan-fabric APIs."""
+
+    API: "Final" = "/appcenter/cisco/ndfc/api/v1/lan-fabric/rest/top-down"
+
+    @classmethod
+    def path(cls, *segments: str) -> str:
+        if not segments:
+            return cls.API
+        return f"{cls.API}/{'/'.join(segments)}"
+
+
+class ResourceManagerBasePath:
+    """Base path helper for resource-manager APIs."""
+
+    API: "Final" = "/appcenter/cisco/ndfc/api/v1/lan-fabric/rest/resource-manager"
+
+    @classmethod
+    def path(cls, *segments: str) -> str:
+        if not segments:
+            return cls.API
+        return f"{cls.API}/{'/'.join(segments)}"

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

@@ -0,0 +1,50 @@
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Endpoint model for /api/v1/manage/fabrics/{fabric_name}/actions/configSave.
+"""
+
+from __future__ import absolute_import, annotations, division, print_function
+
+from typing import Literal
+from urllib.parse import quote
+
+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 (
+    FabricNameMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.base_path import (
+    BasePath,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+from ansible_collections.cisco.nd.plugins.module_utils.types import IdentifierKey
+
+
+class EpFabricConfigSavePost(FabricNameMixin, NDEndpointBaseModel):
+    class_name: Literal["EpFabricConfigSavePost"] = Field(
+        default="EpFabricConfigSavePost",
+        frozen=True,
+        description="Class name for backward compatibility",
+    )
+
+    def set_identifiers(self, identifier: IdentifierKey = None):
+        self.fabric_name = identifier
+
+    @property
+    def path(self) -> str:
+        if self.fabric_name is None:
+            raise ValueError(f"{type(self).__name__}.path: fabric_name must be set before accessing path.")
+        return BasePath.path("fabrics", quote(self.fabric_name, safe=""), "actions", "configSave")
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST
+
+
+EpManageFabricsActionsConfigSavePost = EpFabricConfigSavePost

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

@@ -0,0 +1,53 @@
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Endpoint model for /api/v1/manage/fabrics/{fabric_name}/actions/deploy.
+"""
+
+from __future__ import absolute_import, annotations, division, print_function
+
+from typing import Literal
+from urllib.parse import quote
+
+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 (
+    FabricNameMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.base_path import (
+    BasePath,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+from ansible_collections.cisco.nd.plugins.module_utils.types import IdentifierKey
+
+
+class EpFabricDeployPost(FabricNameMixin, NDEndpointBaseModel):
+    class_name: Literal["EpFabricDeployPost"] = Field(
+        default="EpFabricDeployPost",
+        frozen=True,
+        description="Class name for backward compatibility",
+    )
+
+    force_show_run: bool = Field(default=True, description="forceShowRun query parameter value.")
+
+    def set_identifiers(self, identifier: IdentifierKey = None):
+        self.fabric_name = identifier
+
+    @property
+    def path(self) -> str:
+        if self.fabric_name is None:
+            raise ValueError(f"{type(self).__name__}.path: fabric_name must be set before accessing path.")
+        base_path = BasePath.path("fabrics", quote(self.fabric_name, safe=""), "actions", "deploy")
+        return "{0}?forceShowRun={1}".format(base_path, str(self.force_show_run).lower())
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST
+
+
+EpManageFabricsActionsDeployPost = EpFabricDeployPost

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

@@ -0,0 +1,50 @@
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Endpoint model for /api/v1/manage/fabrics/{fabric_name}/switches.
+"""
+
+from __future__ import absolute_import, annotations, division, print_function
+
+from typing import Literal
+from urllib.parse import quote
+
+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 (
+    FabricNameMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.base_path import (
+    BasePath,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+from ansible_collections.cisco.nd.plugins.module_utils.types import IdentifierKey
+
+
+class EpFabricSwitchesGet(FabricNameMixin, NDEndpointBaseModel):
+    class_name: Literal["EpFabricSwitchesGet"] = Field(
+        default="EpFabricSwitchesGet",
+        frozen=True,
+        description="Class name for backward compatibility",
+    )
+
+    def set_identifiers(self, identifier: IdentifierKey = None):
+        self.fabric_name = identifier
+
+    @property
+    def path(self) -> str:
+        if self.fabric_name is None:
+            raise ValueError(f"{type(self).__name__}.path: fabric_name must be set before accessing path.")
+        return BasePath.path("fabrics", quote(self.fabric_name, safe=""), "switches")
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.GET
+
+
+EpManageFabricsSwitchesGet = EpFabricSwitchesGet

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

@@ -0,0 +1,66 @@
+# Copyright: (c) 2026, Sivakami Sivaraman <sivakasi@cisco.com>
+
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Endpoint models for /top-down/fabrics/{fabric_name}/vrfs.
+"""
+
+from __future__ import absolute_import, annotations, division, print_function
+
+from typing import Literal
+from urllib.parse import quote
+
+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 (
+    FabricNameMixin,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.endpoints.v1.manage.lan_fabric_base_path import (
+    TopDownBasePath,
+)
+from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum
+from ansible_collections.cisco.nd.plugins.module_utils.types import IdentifierKey
+
+
+class _EpFabricVrfsBase(FabricNameMixin, NDEndpointBaseModel):
+    def set_identifiers(self, identifier: IdentifierKey = None):
+        self.fabric_name = identifier
+
+    @property
+    def path(self) -> str:
+        if self.fabric_name is None:
+            raise ValueError(f"{type(self).__name__}.path: fabric_name must be set before accessing path.")
+
+        return TopDownBasePath.path("fabrics", quote(self.fabric_name, safe=""), "vrfs")
+
+
+class EpFabricVrfsGet(_EpFabricVrfsBase):
+    class_name: Literal["EpFabricVrfsGet"] = Field(
+        default="EpFabricVrfsGet",
+        frozen=True,
+        description="Class name for backward compatibility",
+    )
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.GET
+
+
+class EpFabricVrfsPost(_EpFabricVrfsBase):
+    class_name: Literal["EpFabricVrfsPost"] = Field(
+        default="EpFabricVrfsPost",
+        frozen=True,
+        description="Class name for backward compatibility",
+    )
+
+    @property
+    def verb(self) -> HttpVerbEnum:
+        return HttpVerbEnum.POST
+
+
+EpTopDownFabricsVrfsGet = EpFabricVrfsGet
+EpTopDownFabricsVrfsPost = EpFabricVrfsPost