-
Notifications
You must be signed in to change notification settings - Fork 25
Add RestSend framework, enums, and shared unit test infrastructure #185
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: nd42_integration
Are you sure you want to change the base?
Changes from all commits
c579008
5ef139a
eadf98a
d8d6fa9
c34cd0b
146b369
b413b10
6647191
2002e08
f2b6711
3c3a9bf
0ff98fb
b2ce688
9ee6b7a
9b91c2c
e513963
96fa949
4be47fe
a879f03
f21a6b8
9499c13
d7c9f36
51beb08
8f6b51f
cb8a9f8
ba44bfa
b0032d7
d22f1e7
b08e34f
6837fb1
8bbcc2e
ec8d798
e43db59
e7e14a3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| # -*- coding: utf-8 -*- | ||
|
|
||
| # Copyright: (c) 2026, Allen Robel (@arobel) <arobel@cisco.com> | ||
|
|
||
| # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) | ||
| """ | ||
| # exceptions.py | ||
|
|
||
| Exception classes for the cisco.nd Ansible collection. | ||
| """ | ||
|
|
||
| # isort: off | ||
| # fmt: off | ||
| from __future__ import (absolute_import, division, print_function) | ||
| from __future__ import annotations | ||
| # fmt: on | ||
| # isort: on | ||
|
|
||
| # pylint: disable=invalid-name | ||
| __metaclass__ = type | ||
| # pylint: enable=invalid-name | ||
|
|
||
| from typing import Any, Optional | ||
|
|
||
| from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import ( | ||
| BaseModel, | ||
| ConfigDict, | ||
| ) | ||
|
|
||
|
|
||
| class NDErrorData(BaseModel): | ||
| """ | ||
| # Summary | ||
|
|
||
| Pydantic model for structured error data from NDModule requests. | ||
|
|
||
| This model provides type-safe error information that can be serialized | ||
| to a dict for use with Ansible's fail_json. | ||
|
|
||
| ## Attributes | ||
|
|
||
| - msg: Human-readable error message (required) | ||
| - status: HTTP status code as integer (optional) | ||
| - request_payload: Request payload that was sent (optional) | ||
| - response_payload: Response payload from controller (optional) | ||
| - raw: Raw response content for non-JSON responses (optional) | ||
|
|
||
| ## Raises | ||
|
|
||
| - None | ||
| """ | ||
|
|
||
| model_config = ConfigDict(extra="forbid") | ||
|
|
||
| msg: str | ||
| status: Optional[int] = None | ||
| request_payload: Optional[dict[str, Any]] = None | ||
| response_payload: Optional[dict[str, Any]] = None | ||
| raw: Optional[Any] = None | ||
|
|
||
|
|
||
| class NDModuleError(Exception): | ||
| """ | ||
| # Summary | ||
|
|
||
| Exception raised by NDModule when a request fails. | ||
|
|
||
| This exception wraps an NDErrorData Pydantic model, providing structured | ||
| error information that can be used by callers to build appropriate error | ||
| responses (e.g., Ansible fail_json). | ||
|
|
||
| ## Usage Example | ||
|
|
||
| ```python | ||
| try: | ||
| data = nd.request("/api/v1/endpoint", HttpVerbEnum.POST, payload) | ||
| except NDModuleError as e: | ||
| print(f"Error: {e.msg}") | ||
| print(f"Status: {e.status}") | ||
| if e.response_payload: | ||
| print(f"Response: {e.response_payload}") | ||
| # Use to_dict() for fail_json | ||
| module.fail_json(**e.to_dict()) | ||
| ``` | ||
|
|
||
| ## Raises | ||
|
|
||
| - None | ||
| """ | ||
|
|
||
| # pylint: disable=too-many-arguments | ||
| def __init__( | ||
| self, | ||
| msg: str, | ||
| status: Optional[int] = None, | ||
| request_payload: Optional[dict[str, Any]] = None, | ||
| response_payload: Optional[dict[str, Any]] = None, | ||
| raw: Optional[Any] = None, | ||
| ) -> None: | ||
| self.error_data = NDErrorData( | ||
| msg=msg, | ||
| status=status, | ||
| request_payload=request_payload, | ||
| response_payload=response_payload, | ||
| raw=raw, | ||
| ) | ||
| super().__init__(msg) | ||
|
|
||
| @property | ||
| def msg(self) -> str: | ||
| """Human-readable error message.""" | ||
| return self.error_data.msg | ||
|
|
||
| @property | ||
| def status(self) -> Optional[int]: | ||
| """HTTP status code.""" | ||
| return self.error_data.status | ||
|
|
||
| @property | ||
| def request_payload(self) -> Optional[dict[str, Any]]: | ||
| """Request payload that was sent.""" | ||
| return self.error_data.request_payload | ||
|
|
||
| @property | ||
| def response_payload(self) -> Optional[dict[str, Any]]: | ||
| """Response payload from controller.""" | ||
| return self.error_data.response_payload | ||
|
|
||
| @property | ||
| def raw(self) -> Optional[Any]: | ||
| """Raw response content for non-JSON responses.""" | ||
| return self.error_data.raw | ||
|
|
||
| def to_dict(self) -> dict[str, Any]: | ||
| """ | ||
| # Summary | ||
|
|
||
| Convert exception attributes to a dict for use with fail_json. | ||
|
|
||
| Returns a dict containing only non-None attributes. | ||
|
|
||
| ## Raises | ||
|
|
||
| - None | ||
| """ | ||
| return self.error_data.model_dump(exclude_none=True) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,243 @@ | ||
| # -*- coding: utf-8 -*- | ||
|
|
||
| # Copyright: (c) 2026, Allen Robel (@arobel) <arobel@cisco.com> | ||
|
|
||
| # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) | ||
|
|
||
| # pylint: disable=too-few-public-methods | ||
| """ | ||
| # Summary | ||
| Pydantic compatibility layer. | ||
| This module provides a single location for Pydantic imports with fallback | ||
| implementations when Pydantic is not available. This ensures consistent | ||
| behavior across all modules and follows the DRY principle. | ||
| ## Usage | ||
| ### Importing | ||
| Rather than importing directly from pydantic, import from this module: | ||
| ```python | ||
| from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import BaseModel | ||
| ``` | ||
| This ensure that Ansible sanity tests will not fail due to missing Pydantic dependencies. | ||
| """ | ||
|
|
||
| # isort: off | ||
| # fmt: off | ||
| from __future__ import (absolute_import, division, print_function) | ||
| from __future__ import annotations | ||
| # fmt: on | ||
| # isort: on | ||
|
|
||
| # pylint: disable=invalid-name | ||
| __metaclass__ = type | ||
| # pylint: enable=invalid-name | ||
|
|
||
| import traceback | ||
| from typing import TYPE_CHECKING, Any, Callable, Union | ||
|
|
||
allenrobel marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| if TYPE_CHECKING: | ||
| # Type checkers always see the real Pydantic types | ||
| from pydantic import ( | ||
| AfterValidator, | ||
| BaseModel, | ||
| BeforeValidator, | ||
| ConfigDict, | ||
| Field, | ||
| PydanticExperimentalWarning, | ||
| StrictBool, | ||
| ValidationError, | ||
| field_serializer, | ||
| field_validator, | ||
| model_validator, | ||
| validator, | ||
| ) | ||
|
|
||
| HAS_PYDANTIC = True # pylint: disable=invalid-name | ||
| PYDANTIC_IMPORT_ERROR = None # pylint: disable=invalid-name | ||
| else: | ||
| # Runtime: try to import, with fallback | ||
| try: | ||
| from pydantic import ( | ||
| AfterValidator, | ||
| BaseModel, | ||
| BeforeValidator, | ||
| ConfigDict, | ||
| Field, | ||
| PydanticExperimentalWarning, | ||
| StrictBool, | ||
| ValidationError, | ||
| field_serializer, | ||
| field_validator, | ||
| model_validator, | ||
| validator, | ||
| ) | ||
| except ImportError: | ||
| HAS_PYDANTIC = False # pylint: disable=invalid-name | ||
| PYDANTIC_IMPORT_ERROR: Union[str, None] = traceback.format_exc() # pylint: disable=invalid-name | ||
|
|
||
| # Fallback: Minimal BaseModel replacement | ||
| class BaseModel: | ||
| """Fallback BaseModel when pydantic is not available.""" | ||
|
|
||
| model_config = {"validate_assignment": False, "use_enum_values": False} | ||
|
|
||
| def __init__(self, **kwargs): | ||
| """Accept keyword arguments and set them as attributes.""" | ||
| for key, value in kwargs.items(): | ||
| setattr(self, key, value) | ||
|
|
||
| def model_dump(self, exclude_none: bool = False, exclude_defaults: bool = False) -> dict: # pylint: disable=unused-argument | ||
| """Return a dictionary of field names and values. | ||
| Args: | ||
| exclude_none: If True, exclude fields with None values | ||
| exclude_defaults: Accepted for API compatibility but not implemented in fallback | ||
| """ | ||
| result = {} | ||
| for key, value in self.__dict__.items(): | ||
| if exclude_none and value is None: | ||
| continue | ||
| result[key] = value | ||
| return result | ||
|
|
||
| # Fallback: ConfigDict that does nothing | ||
| def ConfigDict(**kwargs) -> dict: # pylint: disable=unused-argument,invalid-name | ||
| """Pydantic ConfigDict fallback when pydantic is not available.""" | ||
| return kwargs | ||
|
|
||
| # Fallback: Field that does nothing | ||
| def Field(**kwargs) -> Any: # pylint: disable=unused-argument,invalid-name | ||
| """Pydantic Field fallback when pydantic is not available.""" | ||
| if "default_factory" in kwargs: | ||
| return kwargs["default_factory"]() | ||
| return kwargs.get("default") | ||
|
|
||
| # Fallback: field_serializer decorator that does nothing | ||
| def field_serializer(*args, **kwargs): # pylint: disable=unused-argument | ||
| """Pydantic field_serializer fallback when pydantic is not available.""" | ||
|
|
||
| def decorator(func): | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| # Fallback: field_validator decorator that does nothing | ||
| def field_validator(*args, **kwargs) -> Callable[..., Any]: # pylint: disable=unused-argument,invalid-name | ||
| """Pydantic field_validator fallback when pydantic is not available.""" | ||
|
|
||
| def decorator(func): | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| # Fallback: AfterValidator that returns the function unchanged | ||
| def AfterValidator(func): # pylint: disable=invalid-name | ||
| """Pydantic AfterValidator fallback when pydantic is not available.""" | ||
| return func | ||
|
|
||
| # Fallback: BeforeValidator that returns the function unchanged | ||
| def BeforeValidator(func): # pylint: disable=invalid-name | ||
| """Pydantic BeforeValidator fallback when pydantic is not available.""" | ||
| return func | ||
|
|
||
| # Fallback: PydanticExperimentalWarning | ||
| PydanticExperimentalWarning = Warning | ||
|
|
||
| # Fallback: StrictBool | ||
| StrictBool = bool | ||
|
|
||
| # Fallback: ValidationError | ||
| class ValidationError(Exception): | ||
| """ | ||
| Pydantic ValidationError fallback when pydantic is not available. | ||
| """ | ||
|
|
||
| def __init__(self, message="A custom error occurred."): | ||
| self.message = message | ||
| super().__init__(self.message) | ||
|
|
||
| def __str__(self): | ||
| return f"ValidationError: {self.message}" | ||
|
|
||
| # Fallback: model_validator decorator that does nothing | ||
| def model_validator(*args, **kwargs): # pylint: disable=unused-argument | ||
| """Pydantic model_validator fallback when pydantic is not available.""" | ||
|
|
||
| def decorator(func): | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| # Fallback: validator decorator that does nothing | ||
| def validator(*args, **kwargs): # pylint: disable=unused-argument | ||
| """Pydantic validator fallback when pydantic is not available.""" | ||
|
|
||
| def decorator(func): | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| else: | ||
| HAS_PYDANTIC = True # pylint: disable=invalid-name | ||
| PYDANTIC_IMPORT_ERROR = None # pylint: disable=invalid-name | ||
|
|
||
|
|
||
| def require_pydantic(module) -> None: | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. As we discussed yesterday I would adapt this to raise an exception and not use module / fail_json in here specifically so we keep it all in the module itself. Is there a clever way for us to know this is executed from main() and just raised the exception without forcing it to call this method in every module main()?
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Hi @lhercot , there's no clever solution here that reduces total boilerplate, rather just one that distributes it differently. The current approach is pragmatic for Ansible. The module parameter is a mild coupling but it gives us There may be a way to use a @requires_pydantic decorator on main(), which would cleaner than an explicit call, but would require main() to accept an instantiated ansible_module as an argument, which isn't really standard Ansible practice (I'm not sure this would even work). Can we explore this in a separate PR (in the interest of time)? import functools
def requires_pydantic(func):
"""Decorator for module main() functions that require pydantic."""
@functools.wraps(func)
def wrapper(module, *args, **kwargs):
if not HAS_PYDANTIC:
from ansible.module_utils.basic import missing_required_lib
module.fail_json(msg=missing_required_lib("pydantic"), exception=PYDANTIC_IMPORT_ERROR)
return func(module, *args, **kwargs)
return wrapper
Then in each module:
from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import requires_pydantic
@requires_pydantic
def main(module):
# pydantic is guaranteed available here
... |
||
| """ | ||
| # Summary | ||
| Call `module.fail_json` if pydantic is not installed. | ||
| Intended to be called once at the top of a module's `main()` function, | ||
| immediately after `AnsibleModule` is instantiated, to provide a clear | ||
| error message when pydantic is a required dependency. | ||
| ## Example | ||
| ```python | ||
| from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import require_pydantic | ||
| def main(): | ||
| module = AnsibleModule(argument_spec=...) | ||
| require_pydantic(module) | ||
| ``` | ||
| ## Raises | ||
| None | ||
| ## Notes | ||
| - Does nothing if pydantic is installed. | ||
| - Uses Ansible's `missing_required_lib` to produce a standardized error | ||
| message that includes installation instructions. | ||
| """ | ||
| if not HAS_PYDANTIC: | ||
| from ansible.module_utils.basic import missing_required_lib # pylint: disable=import-outside-toplevel | ||
|
|
||
| module.fail_json(msg=missing_required_lib("pydantic"), exception=PYDANTIC_IMPORT_ERROR) | ||
|
|
||
|
|
||
| __all__ = [ | ||
| "AfterValidator", | ||
| "BaseModel", | ||
| "BeforeValidator", | ||
| "ConfigDict", | ||
| "Field", | ||
| "HAS_PYDANTIC", | ||
| "PYDANTIC_IMPORT_ERROR", | ||
| "PydanticExperimentalWarning", | ||
| "StrictBool", | ||
| "ValidationError", | ||
| "field_serializer", | ||
| "field_validator", | ||
| "model_validator", | ||
| "require_pydantic", | ||
| "validator", | ||
| ] | ||
Uh oh!
There was an error while loading. Please reload this page.