From 62567a7995c0a8167938da80969ca9d1cc829348 Mon Sep 17 00:00:00 2001 From: Andrey Tikhonov Date: Tue, 25 Nov 2025 23:19:41 +0100 Subject: [PATCH 1/2] Base descanso client --- example.py | 6 +- example_async.py | 6 +- src/annetbox/base/client_async.py | 46 +---- src/annetbox/base/client_sync.py | 88 ++++----- src/annetbox/base/status.py | 19 ++ src/annetbox/v42/client_async.py | 267 +++------------------------ src/annetbox/v42/client_base.py | 296 ++++++++++++++++++++++++++++++ src/annetbox/v42/client_sync.py | 287 +++-------------------------- 8 files changed, 403 insertions(+), 612 deletions(-) create mode 100644 src/annetbox/base/status.py create mode 100644 src/annetbox/v42/client_base.py diff --git a/example.py b/example.py index 859e905..fd64a56 100644 --- a/example.py +++ b/example.py @@ -1,7 +1,7 @@ import os from annetbox.base.client_sync import NetboxStatusClient -from annetbox.v37.client_sync import NetboxV37 +from annetbox.v42.client_sync import NetboxV42 def main(): @@ -14,8 +14,8 @@ def main(): print(status) # basic netbox methods - netbox = NetboxV37(url=url, token=token) - res = netbox.dcim_devices(limit=1) + netbox = NetboxV42(url=url, token=token) + res = netbox.dcim_all_devices(limit=1) print(res) print() diff --git a/example_async.py b/example_async.py index 4dc6215..0b40066 100644 --- a/example_async.py +++ b/example_async.py @@ -2,7 +2,7 @@ import os from annetbox.base.client_async import NetboxStatusClient -from annetbox.v37.client_async import NetboxV37 +from annetbox.v42.client_async import NetboxV42 async def main(): @@ -17,8 +17,8 @@ async def main(): await status_client.close() # basic netbox methods - netbox = NetboxV37(url=url, token=token) - res = await netbox.dcim_devices(limit=1) + netbox = NetboxV42(url=url, token=token) + res = await netbox.dcim_all_devices(limit=1) print(res) print() diff --git a/src/annetbox/base/client_async.py b/src/annetbox/base/client_async.py index 63bff9e..29f014b 100644 --- a/src/annetbox/base/client_async.py +++ b/src/annetbox/base/client_async.py @@ -1,18 +1,13 @@ -import http from collections.abc import Awaitable, Callable from functools import wraps from ssl import SSLContext -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Concatenate, ParamSpec, TypeVar -from adaptix import NameStyle, Retort, name_mapping -from aiohttp import ClientResponse, ClientSession, TCPConnector -from dataclass_rest import get -from dataclass_rest.client_protocol import FactoryProtocol -from dataclass_rest.http.aiohttp import AiohttpClient, AiohttpMethod -from dataclass_rest.http_request import HttpRequest +from aiohttp import ClientSession, TCPConnector +from descanso.http.aiohttp import AiohttpClient -from .exceptions import ClientWithBodyError, ServerWithBodyError -from .models import Model, PagingResponse, Status +from .models import Model, PagingResponse +from .status import BaseNetboxStatusClient Class = TypeVar("Class") ArgsSpec = ParamSpec("ArgsSpec") @@ -115,29 +110,7 @@ async def wrapper( return wrapper -class NoneAwareAiohttpMethod(AiohttpMethod): - async def _on_error_default(self, response: ClientResponse) -> Any: - body = await self._response_body(response) - if http.HTTPStatus.BAD_REQUEST <= response.status \ - < http.HTTPStatus.INTERNAL_SERVER_ERROR: - raise ClientWithBodyError(response.status, body=body) - raise ServerWithBodyError(response.status, body=body) - - async def _pre_process_request(self, request: HttpRequest) -> HttpRequest: - request.query_params = { - k: v for k, v in request.query_params.items() if v is not None - } - return request - - async def _response_body(self, response: ClientResponse) -> Any: - if response.status == http.HTTPStatus.NO_CONTENT: - return None - return await super()._response_body(response) - - class BaseNetboxClient(AiohttpClient): - method_class = NoneAwareAiohttpMethod - def __init__( self, url: str, @@ -156,12 +129,9 @@ def __init__( super().__init__(url, session) async def close(self): - await self.session.close() + await self._session.close() -class NetboxStatusClient(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[name_mapping(name_style=NameStyle.LOWER_KEBAB)]) - @get("status") - async def status(self) -> Status: ... +class NetboxStatusClient(BaseNetboxClient, BaseNetboxStatusClient): + ... diff --git a/src/annetbox/base/client_sync.py b/src/annetbox/base/client_sync.py index 0115d35..3977c3c 100644 --- a/src/annetbox/base/client_sync.py +++ b/src/annetbox/base/client_sync.py @@ -1,28 +1,23 @@ -import http import logging from abc import abstractmethod from collections.abc import Callable, Iterable from functools import wraps from multiprocessing.pool import ThreadPool from ssl import SSLContext -from typing import Any, Concatenate, ParamSpec, Protocol, TypeVar +from typing import Concatenate, ParamSpec, Protocol, TypeVar from urllib.parse import parse_qs, urlparse -from adaptix import NameStyle, Retort, name_mapping -from dataclass_rest import get -from dataclass_rest.client_protocol import FactoryProtocol -from dataclass_rest.http.requests import RequestsClient, RequestsMethod -from requests import Response, Session +from descanso.http.requests import RequestsClient +from requests import Session from requests.adapters import HTTPAdapter -from .exceptions import ClientWithBodyError, ServerWithBodyError -from .models import Model, PagingResponse, Status +from .models import Model, PagingResponse +from .status import BaseNetboxStatusClient Class = TypeVar("Class") ArgsSpec = ParamSpec("ArgsSpec") SessionFactoryT = Callable[[Session], Session] - logger = logging.getLogger(__name__) T = TypeVar("T") @@ -52,9 +47,9 @@ def _collect_by_pages( @wraps(func) def wrapper( - self: Class, - *args: ArgsSpec.args, - **kwargs: ArgsSpec.kwargs, + self: Class, + *args: ArgsSpec.args, + **kwargs: ArgsSpec.kwargs, ) -> PagingResponse[Model]: kwargs.setdefault("offset", 0) page_size = kwargs.pop("page_size", 100) @@ -75,7 +70,7 @@ def wrapper( # approach here we copy 'limit' and 'offset' from next page parsed_url = urlparse(page.next) query_parameters = parse_qs(parsed_url.query) - if "offset" in query_parameters: + if "offset" in query_parameters: kwargs["offset"] = int(query_parameters["offset"][0]) if "limit" in query_parameters: kwargs["limit"] = int(query_parameters["limit"][0]) @@ -95,9 +90,9 @@ def wrapper( # default batch size 100 is calculated to fit list of UUIDs in 4k URL length def collect( - func: Callable[Concatenate[Class, ArgsSpec], PagingResponse[Model]], - field: str = "", - batch_size: int = 100, + func: Callable[Concatenate[Class, ArgsSpec], PagingResponse[Model]], + field: str = "", + batch_size: int = 100, ) -> Callable[Concatenate[Class, ArgsSpec], PagingResponse[Model]]: """ Collect data from method iterating over pages and filter batches. @@ -112,9 +107,9 @@ def collect( @wraps(func) def wrapper( - self: Class, - *args: ArgsSpec.args, - **kwargs: ArgsSpec.kwargs, + self: Class, + *args: ArgsSpec.args, + **kwargs: ArgsSpec.kwargs, ) -> PagingResponse[Model]: method = func.__get__(self, self.__class__) @@ -153,27 +148,13 @@ def apply(batch): return wrapper -class NoneAwareRequestsMethod(RequestsMethod): - def _on_error_default(self, response: Response) -> Any: - body = self._response_body(response) - if http.HTTPStatus.BAD_REQUEST <= response.status_code \ - < http.HTTPStatus.INTERNAL_SERVER_ERROR: - raise ClientWithBodyError(response.status_code, body=body) - raise ServerWithBodyError(response.status_code, body=body) - - def _response_body(self, response: Response) -> Any: - if response.status_code == http.HTTPStatus.NO_CONTENT: - return None - return super()._response_body(response) - - class CustomHTTPAdapter(HTTPAdapter): def __init__( - self, - ssl_context: SSLContext | None = None, - timeout: int = 30, - pool_connections: int = 10, - pool_maxsize: int = 10, + self, + ssl_context: SSLContext | None = None, + timeout: int = 30, + pool_connections: int = 10, + pool_maxsize: int = 10, ) -> None: self.ssl_context = ssl_context self.timeout = timeout @@ -193,15 +174,13 @@ def init_poolmanager(self, *args, **kwargs): class BaseNetboxClient(RequestsClient): - method_class = NoneAwareRequestsMethod - def __init__( - self, - url: str, - token: str = "", - ssl_context: SSLContext | None = None, - threads: int = 1, - session_factory: SessionFactoryT | None = None, + self, + url: str, + token: str = "", + ssl_context: SSLContext | None = None, + threads: int = 1, + session_factory: SessionFactoryT | None = None, ): url = url.rstrip("/") + "/api/" session = self._init_session(ssl_context, threads) @@ -211,12 +190,13 @@ def __init__( session.headers["Authorization"] = f"Token {token}" if session_factory: session = session_factory(session) + session.verify = False super().__init__(url, session) def _init_session( - self, - ssl_context: SSLContext | None = None, - pool_connections: int = 1, + self, + ssl_context: SSLContext | None = None, + pool_connections: int = 1, ) -> Session: adapter = CustomHTTPAdapter( ssl_context=ssl_context, @@ -237,9 +217,5 @@ def _init_pool(self, threads: int) -> _BasePool: return FakePool() -class NetboxStatusClient(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[name_mapping(name_style=NameStyle.LOWER_KEBAB)]) - - @get("status") - def status(self) -> Status: ... +class NetboxStatusClient(BaseNetboxClient, BaseNetboxStatusClient): + ... diff --git a/src/annetbox/base/status.py b/src/annetbox/base/status.py new file mode 100644 index 0000000..38e7024 --- /dev/null +++ b/src/annetbox/base/status.py @@ -0,0 +1,19 @@ +from adaptix import Retort, name_mapping, NameStyle +from descanso import RestBuilder +from descanso.response_transformers import ErrorRaiser + +from annetbox.base.models import Status + + +status_retort = Retort(recipe=[name_mapping(name_style=NameStyle.LOWER_KEBAB)]) +status_rest = RestBuilder( + request_body_dumper=status_retort, + response_body_loader=status_retort, + query_param_dumper=status_retort, + error_raiser=ErrorRaiser(need_body=True), +) + + +class BaseNetboxStatusClient: + @status_rest.get("status") + def status(self) -> Status: ... diff --git a/src/annetbox/v42/client_async.py b/src/annetbox/v42/client_async.py index edd156a..304e4aa 100644 --- a/src/annetbox/v42/client_async.py +++ b/src/annetbox/v42/client_async.py @@ -8,6 +8,7 @@ from annetbox.base.client_async import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import BaseNetboxV42 from .models import ( Cable, ConsolePort, @@ -25,270 +26,44 @@ ) -class NetboxV42(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) +class NetboxV42(BaseNetboxClient, BaseNetboxV42): + dcim_all_interfaces = collect(BaseNetboxV42.dcim_interfaces, field="device_id") + dcim_all_interfaces_by_id = collect(BaseNetboxV42.dcim_interfaces, field="id") - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - - # dcim - @get("dcim/interfaces/") - async def dcim_interfaces( - self, - id: list[int] | None = None, - cable_id: list[int] | None = None, - cable_id__n: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Interface]: - pass - - dcim_all_interfaces = collect(dcim_interfaces, field="device_id") - dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - - @get("dcim/interfaces/{id}/") - async def dcim_interface(self, id: int) -> Interface: - pass - - @get("dcim/console-ports/") - async def dcim_console_ports( - self, - id: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[ConsolePort]: - pass - - dcim_all_console_ports = collect(dcim_console_ports, field="device_id") - dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - - @get("dcim/console-ports/{id}/") - async def dcim_console_port(self, id: int) -> ConsolePort: - pass - - @get("dcim/cables/") - async def dcim_cables( - self, - device: list[str] | None = None, - device_id: list[int] | None = None, - interface_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Cable]: - pass - - dcim_all_cables = collect(dcim_cables, field="interface_id") - - @post("dcim/cables/") - async def dcim_cable_create(self, body: NewCable) -> Cable: - pass - - @post("dcim/cables/") - async def dcim_cable_bulk_create( - self, - body: list[NewCable], - ) -> list[Cable]: - pass + dcim_all_console_ports = collect(BaseNetboxV42.dcim_console_ports, field="device_id") + dcim_all_console_ports_by_id = collect(BaseNetboxV42.dcim_console_ports, field="id") - @delete("dcim/cables/") - async def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: - pass + dcim_all_cables = collect(BaseNetboxV42.dcim_cables, field="interface_id") async def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: return await self._dcim_cable_bulk_delete( [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") - async def dcim_cable_delete(self, id: int) -> None: - pass - - @get("dcim/devices/") - async def dcim_devices( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Device]: - pass - - dcim_all_devices = collect(dcim_devices) - dcim_all_devices_by_id = collect(dcim_devices, field="id") - - - @get("dcim/devices/?brief=1") - async def dcim_devices_brief( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - has_oob_ip: bool | None = None, - has_primary_ip: bool | None = None, - location_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Entity]: - pass - - dcim_all_devices_brief = collect(dcim_devices_brief) - dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - - @get("dcim/devices/{device_id}/") - async def dcim_device( - self, - device_id: int, - ) -> Device: - pass - - # ipam - @get("ipam/ip-addresses/") - async def ipam_ip_addresses( - self, - interface_id: list[int] | None = None, - device_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[IpAddress]: - pass - - ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - - @get("ipam/ip-addresses/{id}/") - async def ipam_ip_address( - self, - id: int, - ) -> IpAddress: - pass - - @get("ipam/prefixes/") - async def prefixes( - self, - prefix: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Prefix]: - pass - - ipam_all_prefixes = collect(prefixes, field="prefix") + dcim_all_devices = collect(BaseNetboxV42.dcim_devices) + dcim_all_devices_by_id = collect(BaseNetboxV42.dcim_devices, field="id") - @get("ipam/vlans/") - async def ipam_vlans( - self, - id: list[int] | None = None, - vid: list[int] | None = None, - group: list[str] | None = None, - status: list[str] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - role: list[str] | None = None, - site: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vlan]: - pass - ipam_all_vlans = collect(ipam_vlans, field="vid") - ipam_all_vlans_by_id = collect(ipam_vlans, field="id") + dcim_all_devices_brief = collect(BaseNetboxV42.dcim_devices_brief) + dcim_all_devices_brief_by_id = collect(BaseNetboxV42.dcim_devices_brief, field="id") - @get("ipam/vrfs/") - async def ipam_vrfs( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vrf]: - pass + ipam_all_ip_addresses = collect(BaseNetboxV42.ipam_ip_addresses, field="interface_id") - ipam_all_vrfs = collect(ipam_vrfs, field="vid") - ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") + ipam_all_prefixes = collect(BaseNetboxV42.prefixes, field="prefix") - @get("ipam/fhrp-groups/") - async def ipam_fhrp_groups( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - protocol: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - related_ip: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroup]: - pass - ipam_all_fhrp_groups = collect(ipam_fhrp_groups) - ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") + ipam_all_vlans = collect(BaseNetboxV42.ipam_vlans, field="vid") + ipam_all_vlans_by_id = collect(BaseNetboxV42.ipam_vlans, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") - async def ipam_fhrp_group_assignments_brief( - self, - id: list[int] | None = None, - interface_id: list[int] | None = None, - device: list[str] | None = None, - device_id: list[int] | None = None, - group_id: list[int] | None = None, + ipam_all_vrfs = collect(BaseNetboxV42.ipam_vrfs, field="vid") + ipam_all_vrfs_by_id = collect(BaseNetboxV42.ipam_vrfs, field="id") - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroupAssignmentBrief]: - pass + ipam_all_fhrp_groups = collect(BaseNetboxV42.ipam_fhrp_groups) + ipam_all_fhrp_groups_by_id = collect(BaseNetboxV42.ipam_fhrp_groups, field="id") ipam_all_fhrp_group_assignments = collect( - ipam_fhrp_group_assignments_brief, + BaseNetboxV42.ipam_fhrp_group_assignments_brief, ) ipam_all_fhrp_group_assignments_by_interface = collect( - ipam_fhrp_group_assignments_brief, field="interface_id", + BaseNetboxV42.ipam_fhrp_group_assignments_brief, field="interface_id", ) diff --git a/src/annetbox/v42/client_base.py b/src/annetbox/v42/client_base.py new file mode 100644 index 0000000..704a495 --- /dev/null +++ b/src/annetbox/v42/client_base.py @@ -0,0 +1,296 @@ +from collections.abc import Iterable +from datetime import datetime + +import dateutil.parser +from adaptix import Retort, loader, name_mapping +from descanso import RestBuilder +from descanso.response_transformers import ErrorRaiser + +from annetbox.base.client_sync import collect +from annetbox.base.models import PagingResponse +from .models import ( + Cable, + ConsolePort, + Device, + Entity, + FHRPGroup, + FHRPGroupAssignmentBrief, + Interface, + IpAddress, + ItemToDelete, + NewCable, + Prefix, + Vlan, + Vrf, +) + +retort = Retort(recipe=[ + loader(datetime, dateutil.parser.parse), + name_mapping(NewCable, omit_default=True), +]) +rest = RestBuilder( + request_body_dumper=retort, + response_body_loader=retort, + query_param_dumper=retort, + error_raiser=ErrorRaiser(need_body=True), +) + +class BaseNetboxV42: + # dcim + @rest.get("dcim/interfaces/") + def dcim_interfaces( + self, + id: list[int] | None = None, + cable_id: list[int] | None = None, + cable_id__n: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Interface]: + pass + + dcim_all_interfaces = collect( + dcim_interfaces, field="device_id", batch_size=10, # heavy request + ) + dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") + + @rest.get("dcim/interfaces/{id}/") + def dcim_interface(self, id: int) -> Interface: + pass + + @rest.get("dcim/console-ports/") + def dcim_console_ports( + self, + id: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[ConsolePort]: + pass + + dcim_all_console_ports = collect(dcim_console_ports, field="device_id") + dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") + + @rest.get("dcim/console-ports/{id}/") + def dcim_console_port(self, id: int) -> ConsolePort: + pass + + @rest.get("dcim/cables/") + def dcim_cables( + self, + device: list[str] | None = None, + device_id: list[int] | None = None, + interface_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Cable]: + pass + + dcim_all_cables = collect(dcim_cables, field="interface_id") + + @rest.post("dcim/cables/") + def dcim_cable_create(self, body: NewCable) -> Cable: + pass + + @rest.post("dcim/cables/") + def dcim_cable_bulk_create( + self, + body: list[NewCable], + ) -> list[Cable]: + pass + + @rest.delete("dcim/cables/") + def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: + pass + + def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: + return self._dcim_cable_bulk_delete( + [ItemToDelete(id=x) for x in body], + ) + + @rest.delete("dcim/cables/{id}/") + def dcim_cable_delete(self, id: int) -> None: + pass + + @rest.get("dcim/devices/") + def dcim_devices( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Device]: + pass + + dcim_all_devices = collect(dcim_devices) + dcim_all_devices_by_id = collect(dcim_devices, field="id") + + + @rest.get("dcim/devices/?brief=1") + def dcim_devices_brief( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + has_oob_ip: bool | None = None, + has_primary_ip: bool | None = None, + location_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Entity]: + pass + + dcim_all_devices_brief = collect(dcim_devices_brief) + dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") + + @rest.get("dcim/devices/{device_id}/") + def dcim_device( + self, + device_id: int, + ) -> Device: + pass + + # ipam + @rest.get("ipam/ip-addresses/") + def ipam_ip_addresses( + self, + interface_id: list[int] | None = None, + device_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[IpAddress]: + pass + + ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") + + @rest.get("ipam/ip-addresses/{id}/") + def ipam_ip_address( + self, + id: int, + ) -> IpAddress: + pass + + @rest.get("ipam/prefixes/") + def prefixes( + self, + prefix: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Prefix]: + pass + + ipam_all_prefixes = collect(prefixes, field="prefix") + + @rest.get("ipam/vlans/") + def ipam_vlans( + self, + id: list[int] | None = None, + vid: list[int] | None = None, + group: list[str] | None = None, + status: list[str] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + role: list[str] | None = None, + site: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vlan]: + pass + + ipam_all_vlans = collect(ipam_vlans, field="vid") + ipam_all_vlans_by_id = collect(ipam_vlans, field="id") + + @rest.get("ipam/vrfs/") + def ipam_vrfs( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vrf]: + pass + + ipam_all_vrfs = collect(ipam_vrfs, field="vid") + ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") + + @rest.get("ipam/fhrp-groups/") + def ipam_fhrp_groups( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + protocol: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + related_ip: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroup]: + pass + + ipam_all_fhrp_groups = collect(ipam_fhrp_groups) + ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") + + @rest.get("ipam/fhrp-group-assignments/?brief=1") + def ipam_fhrp_group_assignments_brief( + self, + id: list[int] | None = None, + interface_id: list[int] | None = None, + device: list[str] | None = None, + device_id: list[int] | None = None, + group_id: list[int] | None = None, + + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroupAssignmentBrief]: + pass + + ipam_all_fhrp_group_assignments = collect( + ipam_fhrp_group_assignments_brief, + ) + ipam_all_fhrp_group_assignments_by_interface = collect( + ipam_fhrp_group_assignments_brief, field="interface_id", + ) diff --git a/src/annetbox/v42/client_sync.py b/src/annetbox/v42/client_sync.py index e02149a..d356bcc 100644 --- a/src/annetbox/v42/client_sync.py +++ b/src/annetbox/v42/client_sync.py @@ -1,296 +1,51 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_sync import BaseNetboxClient, collect -from annetbox.base.models import PagingResponse +from .client_base import BaseNetboxV42 from .models import ( - Cable, - ConsolePort, - Device, - Entity, - FHRPGroup, - FHRPGroupAssignmentBrief, - Interface, - IpAddress, ItemToDelete, - NewCable, - Prefix, - Vlan, - Vrf, ) -class NetboxV42(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - - # dcim - @get("dcim/interfaces/") - def dcim_interfaces( - self, - id: list[int] | None = None, - cable_id: list[int] | None = None, - cable_id__n: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Interface]: - pass - +class NetboxV42(BaseNetboxClient, BaseNetboxV42): dcim_all_interfaces = collect( - dcim_interfaces, field="device_id", batch_size=10, # heavy request + BaseNetboxV42.dcim_interfaces, field="device_id", batch_size=10, # heavy request ) - dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - - @get("dcim/interfaces/{id}/") - def dcim_interface(self, id: int) -> Interface: - pass - - @get("dcim/console-ports/") - def dcim_console_ports( - self, - id: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[ConsolePort]: - pass - - dcim_all_console_ports = collect(dcim_console_ports, field="device_id") - dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - - @get("dcim/console-ports/{id}/") - def dcim_console_port(self, id: int) -> ConsolePort: - pass - - @get("dcim/cables/") - def dcim_cables( - self, - device: list[str] | None = None, - device_id: list[int] | None = None, - interface_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Cable]: - pass - - dcim_all_cables = collect(dcim_cables, field="interface_id") - - @post("dcim/cables/") - def dcim_cable_create(self, body: NewCable) -> Cable: - pass + dcim_all_interfaces_by_id = collect(BaseNetboxV42.dcim_interfaces, field="id") - @post("dcim/cables/") - def dcim_cable_bulk_create( - self, - body: list[NewCable], - ) -> list[Cable]: - pass + dcim_all_console_ports = collect(BaseNetboxV42.dcim_console_ports, field="device_id") + dcim_all_console_ports_by_id = collect(BaseNetboxV42.dcim_console_ports, field="id") - @delete("dcim/cables/") - def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: - pass + dcim_all_cables = collect(BaseNetboxV42.dcim_cables, field="interface_id") def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: return self._dcim_cable_bulk_delete( [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") - def dcim_cable_delete(self, id: int) -> None: - pass - - @get("dcim/devices/") - def dcim_devices( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Device]: - pass - - dcim_all_devices = collect(dcim_devices) - dcim_all_devices_by_id = collect(dcim_devices, field="id") - - - @get("dcim/devices/?brief=1") - def dcim_devices_brief( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - has_oob_ip: bool | None = None, - has_primary_ip: bool | None = None, - location_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Entity]: - pass - - dcim_all_devices_brief = collect(dcim_devices_brief) - dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - - @get("dcim/devices/{device_id}/") - def dcim_device( - self, - device_id: int, - ) -> Device: - pass - - # ipam - @get("ipam/ip-addresses/") - def ipam_ip_addresses( - self, - interface_id: list[int] | None = None, - device_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[IpAddress]: - pass - - ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - - @get("ipam/ip-addresses/{id}/") - def ipam_ip_address( - self, - id: int, - ) -> IpAddress: - pass - - @get("ipam/prefixes/") - def prefixes( - self, - prefix: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Prefix]: - pass - - ipam_all_prefixes = collect(prefixes, field="prefix") - - @get("ipam/vlans/") - def ipam_vlans( - self, - id: list[int] | None = None, - vid: list[int] | None = None, - group: list[str] | None = None, - status: list[str] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - role: list[str] | None = None, - site: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vlan]: - pass + dcim_all_devices = collect(BaseNetboxV42.dcim_devices) + dcim_all_devices_by_id = collect(BaseNetboxV42.dcim_devices, field="id") - ipam_all_vlans = collect(ipam_vlans, field="vid") - ipam_all_vlans_by_id = collect(ipam_vlans, field="id") + dcim_all_devices_brief = collect(BaseNetboxV42.dcim_devices_brief) + dcim_all_devices_brief_by_id = collect(BaseNetboxV42.dcim_devices_brief, field="id") - @get("ipam/vrfs/") - def ipam_vrfs( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vrf]: - pass + ipam_all_ip_addresses = collect(BaseNetboxV42.ipam_ip_addresses, field="interface_id") - ipam_all_vrfs = collect(ipam_vrfs, field="vid") - ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") + ipam_all_prefixes = collect(BaseNetboxV42.prefixes, field="prefix") - @get("ipam/fhrp-groups/") - def ipam_fhrp_groups( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - protocol: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - related_ip: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroup]: - pass + ipam_all_vlans = collect(BaseNetboxV42.ipam_vlans, field="vid") + ipam_all_vlans_by_id = collect(BaseNetboxV42.ipam_vlans, field="id") - ipam_all_fhrp_groups = collect(ipam_fhrp_groups) - ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") + ipam_all_vrfs = collect(BaseNetboxV42.ipam_vrfs, field="vid") + ipam_all_vrfs_by_id = collect(BaseNetboxV42.ipam_vrfs, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") - def ipam_fhrp_group_assignments_brief( - self, - id: list[int] | None = None, - interface_id: list[int] | None = None, - device: list[str] | None = None, - device_id: list[int] | None = None, - group_id: list[int] | None = None, + ipam_all_fhrp_groups = collect(BaseNetboxV42.ipam_fhrp_groups) + ipam_all_fhrp_groups_by_id = collect(BaseNetboxV42.ipam_fhrp_groups, field="id") - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroupAssignmentBrief]: - pass ipam_all_fhrp_group_assignments = collect( - ipam_fhrp_group_assignments_brief, + BaseNetboxV42.ipam_fhrp_group_assignments_brief, ) ipam_all_fhrp_group_assignments_by_interface = collect( - ipam_fhrp_group_assignments_brief, field="interface_id", + BaseNetboxV42.ipam_fhrp_group_assignments_brief, field="interface_id", ) From 9fe1b7a5e6e9618eca6e70703a16b946196e4ea9 Mon Sep 17 00:00:00 2001 From: Andrey Tikhonov Date: Wed, 26 Nov 2025 12:14:52 +0100 Subject: [PATCH 2/2] Switch from dataclass-rest to descanso --- pyproject.toml | 2 +- src/annetbox/base/status.py | 3 +- src/annetbox/v24/client_async.py | 18 +- src/annetbox/v24/client_base.py | 16 ++ src/annetbox/v24/client_sync.py | 18 +- src/annetbox/v37/client_async.py | 51 ++---- src/annetbox/v37/client_base.py | 21 +++ src/annetbox/v37/client_sync.py | 51 ++---- src/annetbox/v41/client_async.py | 51 ++---- src/annetbox/v41/client_base.py | 21 +++ src/annetbox/v41/client_sync.py | 51 ++---- src/annetbox/v42/client_async.py | 264 ++++++++++++++++++++++++++--- src/annetbox/v42/client_base.py | 275 ------------------------------- src/annetbox/v42/client_sync.py | 272 +++++++++++++++++++++++++++--- 14 files changed, 630 insertions(+), 484 deletions(-) create mode 100644 src/annetbox/v24/client_base.py create mode 100644 src/annetbox/v37/client_base.py create mode 100644 src/annetbox/v41/client_base.py diff --git a/pyproject.toml b/pyproject.toml index 56fb187..1c39713 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,7 +33,7 @@ classifiers = [ ] dependencies = [ 'adaptix~=3.0.0b2', - 'dataclass-rest~=0.4', + 'descanso~=0.7.0', 'python-dateutil~=2.8', ] [project.optional-dependencies] diff --git a/src/annetbox/base/status.py b/src/annetbox/base/status.py index 38e7024..99b76c3 100644 --- a/src/annetbox/base/status.py +++ b/src/annetbox/base/status.py @@ -1,10 +1,9 @@ -from adaptix import Retort, name_mapping, NameStyle +from adaptix import NameStyle, Retort, name_mapping from descanso import RestBuilder from descanso.response_transformers import ErrorRaiser from annetbox.base.models import Status - status_retort = Retort(recipe=[name_mapping(name_style=NameStyle.LOWER_KEBAB)]) status_rest = RestBuilder( request_body_dumper=status_retort, diff --git a/src/annetbox/v24/client_async.py b/src/annetbox/v24/client_async.py index 3b49369..9830a55 100644 --- a/src/annetbox/v24/client_async.py +++ b/src/annetbox/v24/client_async.py @@ -1,20 +1,12 @@ -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader -from dataclass_rest import get - from annetbox.base.client_async import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import Device, Interface, IpAddress class NetboxV24(BaseNetboxClient): - def _init_response_body_factory(self) -> Retort: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") async def dcim_interfaces( self, device_id: list[int] | None = None, @@ -25,7 +17,7 @@ async def dcim_interfaces( dcim_all_interfaces = collect(dcim_interfaces, field="device_id") - @get("dcim/devices/") + @rest.get("dcim/devices/") async def dcim_devices( self, name: list[str] | None = None, @@ -37,7 +29,7 @@ async def dcim_devices( dcim_all_devices = collect(dcim_devices) - @get("dcim/devices/{device_id}/") + @rest.get("dcim/devices/{device_id}/") async def dcim_device( self, device_id: int, @@ -45,7 +37,7 @@ async def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") async def ipam_ip_addresses( self, interface_id: list[int] | None = None, diff --git a/src/annetbox/v24/client_base.py b/src/annetbox/v24/client_base.py new file mode 100644 index 0000000..dd53c1e --- /dev/null +++ b/src/annetbox/v24/client_base.py @@ -0,0 +1,16 @@ +from datetime import datetime + +import dateutil.parser +from adaptix import Retort, loader +from descanso import RestBuilder +from descanso.response_transformers import ErrorRaiser + +retort = Retort(recipe=[ + loader(datetime, dateutil.parser.parse), +]) +rest = RestBuilder( + request_body_dumper=retort, + response_body_loader=retort, + query_param_dumper=retort, + error_raiser=ErrorRaiser(need_body=True), +) diff --git a/src/annetbox/v24/client_sync.py b/src/annetbox/v24/client_sync.py index 9a4ff72..e7a1f51 100644 --- a/src/annetbox/v24/client_sync.py +++ b/src/annetbox/v24/client_sync.py @@ -1,20 +1,12 @@ -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader -from dataclass_rest import get - from annetbox.base.client_sync import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import Device, Interface, IpAddress class NetboxV24(BaseNetboxClient): - def _init_response_body_factory(self) -> Retort: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") def dcim_interfaces( self, device_id: list[int] | None = None, @@ -25,7 +17,7 @@ def dcim_interfaces( dcim_all_interfaces = collect(dcim_interfaces, field="device_id") - @get("dcim/devices/") + @rest.get("dcim/devices/") def dcim_devices( self, name: list[str] | None = None, @@ -37,7 +29,7 @@ def dcim_devices( dcim_all_devices = collect(dcim_devices) - @get("dcim/devices/{device_id}") + @rest.get("dcim/devices/{device_id}") def dcim_device( self, device_id: int, @@ -45,7 +37,7 @@ def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") def ipam_ip_addresses( self, interface_id: list[int] | None = None, diff --git a/src/annetbox/v37/client_async.py b/src/annetbox/v37/client_async.py index 64a9b5d..b86a3d4 100644 --- a/src/annetbox/v37/client_async.py +++ b/src/annetbox/v37/client_async.py @@ -1,13 +1,8 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_async import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import ( Cable, ConsolePort, @@ -24,18 +19,8 @@ class NetboxV37(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") async def dcim_interfaces( self, id: list[int] | None = None, @@ -55,11 +40,11 @@ async def dcim_interfaces( ) dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - @get("dcim/interfaces/{id}/") + @rest.get("dcim/interfaces/{id}/") async def dcim_interface(self, id: int) -> Interface: pass - @get("dcim/console-ports/") + @rest.get("dcim/console-ports/") async def dcim_console_ports( self, id: list[int] | None = None, @@ -75,11 +60,11 @@ async def dcim_console_ports( dcim_all_console_ports = collect(dcim_console_ports, field="device_id") dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - @get("dcim/console-ports/{id}/") + @rest.get("dcim/console-ports/{id}/") async def dcim_console_port(self, id: int) -> ConsolePort: pass - @get("dcim/cables/") + @rest.get("dcim/cables/") async def dcim_cables( self, device: list[str] | None = None, @@ -92,18 +77,18 @@ async def dcim_cables( dcim_all_cables = collect(dcim_cables, field="interface_id") - @post("dcim/cables/") + @rest.post("dcim/cables/") async def dcim_cable_create(self, body: NewCable) -> Cable: pass - @post("dcim/cables/") + @rest.post("dcim/cables/") async def dcim_cable_bulk_create( self, body: list[NewCable], ) -> list[Cable]: pass - @delete("dcim/cables/") + @rest.delete("dcim/cables/") async def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: pass @@ -112,11 +97,11 @@ async def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") + @rest.delete("dcim/cables/{id}/") async def dcim_cable_delete(self, id: int) -> None: pass - @get("dcim/devices/") + @rest.get("dcim/devices/") async def dcim_devices( self, name: list[str] | None = None, @@ -147,7 +132,7 @@ async def dcim_devices( dcim_all_devices_by_id = collect(dcim_devices, field="id") - @get("dcim/devices/?brief=1") + @rest.get("dcim/devices/?brief=1") async def dcim_devices_brief( self, name: list[str] | None = None, @@ -180,7 +165,7 @@ async def dcim_devices_brief( dcim_all_devices_brief = collect(dcim_devices_brief) dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - @get("dcim/devices/{device_id}/") + @rest.get("dcim/devices/{device_id}/") async def dcim_device( self, device_id: int, @@ -188,7 +173,7 @@ async def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") async def ipam_ip_addresses( self, interface_id: list[int] | None = None, @@ -200,14 +185,14 @@ async def ipam_ip_addresses( ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - @get("ipam/ip-addresses/{id}/") + @rest.get("ipam/ip-addresses/{id}/") async def ipam_ip_address( self, id: int, ) -> IpAddress: pass - @get("ipam/prefixes/") + @rest.get("ipam/prefixes/") async def prefixes( self, prefix: list[str] | None = None, @@ -218,7 +203,7 @@ async def prefixes( ipam_all_prefixes = collect(prefixes, field="prefix") - @get("ipam/fhrp-groups/") + @rest.get("ipam/fhrp-groups/") async def ipam_fhrp_groups( self, id: list[int] | None = None, @@ -235,7 +220,7 @@ async def ipam_fhrp_groups( ipam_all_fhrp_groups = collect(ipam_fhrp_groups) ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") + @rest.get("ipam/fhrp-group-assignments/?brief=1") async def ipam_fhrp_group_assignments_brief( self, id: list[int] | None = None, diff --git a/src/annetbox/v37/client_base.py b/src/annetbox/v37/client_base.py new file mode 100644 index 0000000..083e2b9 --- /dev/null +++ b/src/annetbox/v37/client_base.py @@ -0,0 +1,21 @@ +from datetime import datetime + +import dateutil.parser +from adaptix import Retort, loader, name_mapping +from descanso import RestBuilder +from descanso.response_transformers import ErrorRaiser + +from .models import ( + NewCable, +) + +retort = Retort(recipe=[ + loader(datetime, dateutil.parser.parse), + name_mapping(NewCable, omit_default=True), +]) +rest = RestBuilder( + request_body_dumper=retort, + response_body_loader=retort, + query_param_dumper=retort, + error_raiser=ErrorRaiser(need_body=True), +) diff --git a/src/annetbox/v37/client_sync.py b/src/annetbox/v37/client_sync.py index 975744d..1bbc4da 100644 --- a/src/annetbox/v37/client_sync.py +++ b/src/annetbox/v37/client_sync.py @@ -1,13 +1,8 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_sync import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import ( Cable, ConsolePort, @@ -24,18 +19,8 @@ class NetboxV37(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") def dcim_interfaces( self, id: list[int] | None = None, @@ -55,11 +40,11 @@ def dcim_interfaces( ) dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - @get("dcim/interfaces/{id}/") + @rest.get("dcim/interfaces/{id}/") def dcim_interface(self, id: int) -> Interface: pass - @get("dcim/console-ports/") + @rest.get("dcim/console-ports/") def dcim_console_ports( self, id: list[int] | None = None, @@ -75,11 +60,11 @@ def dcim_console_ports( dcim_all_console_ports = collect(dcim_console_ports, field="device_id") dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - @get("dcim/console-ports/{id}/") + @rest.get("dcim/console-ports/{id}/") def dcim_console_port(self, id: int) -> ConsolePort: pass - @get("dcim/cables/") + @rest.get("dcim/cables/") def dcim_cables( self, device: list[str] | None = None, @@ -92,18 +77,18 @@ def dcim_cables( dcim_all_cables = collect(dcim_cables, field="interface_id") - @post("dcim/cables/") + @rest.post("dcim/cables/") def dcim_cable_create(self, body: NewCable) -> Cable: pass - @post("dcim/cables/") + @rest.post("dcim/cables/") def dcim_cable_bulk_create( self, body: list[NewCable], ) -> list[Cable]: pass - @delete("dcim/cables/") + @rest.delete("dcim/cables/") def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: pass @@ -112,11 +97,11 @@ def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") + @rest.delete("dcim/cables/{id}/") def dcim_cable_delete(self, id: int) -> None: pass - @get("dcim/devices/") + @rest.get("dcim/devices/") def dcim_devices( self, name: list[str] | None = None, @@ -147,7 +132,7 @@ def dcim_devices( dcim_all_devices_by_id = collect(dcim_devices, field="id") - @get("dcim/devices/?brief=1") + @rest.get("dcim/devices/?brief=1") def dcim_devices_brief( self, name: list[str] | None = None, @@ -180,7 +165,7 @@ def dcim_devices_brief( dcim_all_devices_brief = collect(dcim_devices_brief) dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - @get("dcim/devices/{device_id}/") + @rest.get("dcim/devices/{device_id}/") def dcim_device( self, device_id: int, @@ -188,7 +173,7 @@ def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") def ipam_ip_addresses( self, interface_id: list[int] | None = None, @@ -200,14 +185,14 @@ def ipam_ip_addresses( ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - @get("ipam/ip-addresses/{id}/") + @rest.get("ipam/ip-addresses/{id}/") def ipam_ip_address( self, id: int, ) -> IpAddress: pass - @get("ipam/prefixes/") + @rest.get("ipam/prefixes/") def prefixes( self, prefix: list[str] | None = None, @@ -218,7 +203,7 @@ def prefixes( ipam_all_prefixes = collect(prefixes, field="prefix") - @get("ipam/fhrp-groups/") + @rest.get("ipam/fhrp-groups/") def ipam_fhrp_groups( self, id: list[int] | None = None, @@ -235,7 +220,7 @@ def ipam_fhrp_groups( ipam_all_fhrp_groups = collect(ipam_fhrp_groups) ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") + @rest.get("ipam/fhrp-group-assignments/?brief=1") def ipam_fhrp_group_assignments_brief( self, id: list[int] | None = None, diff --git a/src/annetbox/v41/client_async.py b/src/annetbox/v41/client_async.py index 4a5ccc8..fa94a77 100644 --- a/src/annetbox/v41/client_async.py +++ b/src/annetbox/v41/client_async.py @@ -1,13 +1,8 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_async import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import ( Cable, ConsolePort, @@ -24,18 +19,8 @@ class NetboxV41(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") async def dcim_interfaces( self, id: list[int] | None = None, @@ -53,11 +38,11 @@ async def dcim_interfaces( dcim_all_interfaces = collect(dcim_interfaces, field="device_id") dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - @get("dcim/interfaces/{id}/") + @rest.get("dcim/interfaces/{id}/") async def dcim_interface(self, id: int) -> Interface: pass - @get("dcim/console-ports/") + @rest.get("dcim/console-ports/") async def dcim_console_ports( self, id: list[int] | None = None, @@ -73,11 +58,11 @@ async def dcim_console_ports( dcim_all_console_ports = collect(dcim_console_ports, field="device_id") dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - @get("dcim/console-ports/{id}/") + @rest.get("dcim/console-ports/{id}/") async def dcim_console_port(self, id: int) -> ConsolePort: pass - @get("dcim/cables/") + @rest.get("dcim/cables/") async def dcim_cables( self, device: list[str] | None = None, @@ -90,18 +75,18 @@ async def dcim_cables( dcim_all_cables = collect(dcim_cables, field="interface_id") - @post("dcim/cables/") + @rest.post("dcim/cables/") async def dcim_cable_create(self, body: NewCable) -> Cable: pass - @post("dcim/cables/") + @rest.post("dcim/cables/") async def dcim_cable_bulk_create( self, body: list[NewCable], ) -> list[Cable]: pass - @delete("dcim/cables/") + @rest.delete("dcim/cables/") async def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: pass @@ -110,11 +95,11 @@ async def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") + @rest.delete("dcim/cables/{id}/") async def dcim_cable_delete(self, id: int) -> None: pass - @get("dcim/devices/") + @rest.get("dcim/devices/") async def dcim_devices( self, name: list[str] | None = None, @@ -145,7 +130,7 @@ async def dcim_devices( dcim_all_devices_by_id = collect(dcim_devices, field="id") - @get("dcim/devices/?brief=1") + @rest.get("dcim/devices/?brief=1") async def dcim_devices_brief( self, name: list[str] | None = None, @@ -178,7 +163,7 @@ async def dcim_devices_brief( dcim_all_devices_brief = collect(dcim_devices_brief) dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - @get("dcim/devices/{device_id}/") + @rest.get("dcim/devices/{device_id}/") async def dcim_device( self, device_id: int, @@ -186,7 +171,7 @@ async def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") async def ipam_ip_addresses( self, interface_id: list[int] | None = None, @@ -198,14 +183,14 @@ async def ipam_ip_addresses( ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - @get("ipam/ip-addresses/{id}/") + @rest.get("ipam/ip-addresses/{id}/") async def ipam_ip_address( self, id: int, ) -> IpAddress: pass - @get("ipam/prefixes/") + @rest.get("ipam/prefixes/") async def prefixes( self, prefix: list[str] | None = None, @@ -216,7 +201,7 @@ async def prefixes( ipam_all_prefixes = collect(prefixes, field="prefix") - @get("ipam/fhrp-groups/") + @rest.get("ipam/fhrp-groups/") async def ipam_fhrp_groups( self, id: list[int] | None = None, @@ -233,7 +218,7 @@ async def ipam_fhrp_groups( ipam_all_fhrp_groups = collect(ipam_fhrp_groups) ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") + @rest.get("ipam/fhrp-group-assignments/?brief=1") async def ipam_fhrp_group_assignments_brief( self, id: list[int] | None = None, diff --git a/src/annetbox/v41/client_base.py b/src/annetbox/v41/client_base.py new file mode 100644 index 0000000..083e2b9 --- /dev/null +++ b/src/annetbox/v41/client_base.py @@ -0,0 +1,21 @@ +from datetime import datetime + +import dateutil.parser +from adaptix import Retort, loader, name_mapping +from descanso import RestBuilder +from descanso.response_transformers import ErrorRaiser + +from .models import ( + NewCable, +) + +retort = Retort(recipe=[ + loader(datetime, dateutil.parser.parse), + name_mapping(NewCable, omit_default=True), +]) +rest = RestBuilder( + request_body_dumper=retort, + response_body_loader=retort, + query_param_dumper=retort, + error_raiser=ErrorRaiser(need_body=True), +) diff --git a/src/annetbox/v41/client_sync.py b/src/annetbox/v41/client_sync.py index 9baef2c..30ceaf7 100644 --- a/src/annetbox/v41/client_sync.py +++ b/src/annetbox/v41/client_sync.py @@ -1,13 +1,8 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_sync import BaseNetboxClient, collect from annetbox.base.models import PagingResponse +from .client_base import rest from .models import ( Cable, ConsolePort, @@ -24,18 +19,8 @@ class NetboxV41(BaseNetboxClient): - def _init_response_body_factory(self) -> FactoryProtocol: - return Retort(recipe=[loader(datetime, dateutil.parser.parse)]) - - def _init_request_body_factory(self) -> FactoryProtocol: - return Retort( - recipe=[ - name_mapping(NewCable, omit_default=True), - ], - ) - # dcim - @get("dcim/interfaces/") + @rest.get("dcim/interfaces/") def dcim_interfaces( self, id: list[int] | None = None, @@ -55,11 +40,11 @@ def dcim_interfaces( ) dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - @get("dcim/interfaces/{id}/") + @rest.get("dcim/interfaces/{id}/") def dcim_interface(self, id: int) -> Interface: pass - @get("dcim/console-ports/") + @rest.get("dcim/console-ports/") def dcim_console_ports( self, id: list[int] | None = None, @@ -75,11 +60,11 @@ def dcim_console_ports( dcim_all_console_ports = collect(dcim_console_ports, field="device_id") dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - @get("dcim/console-ports/{id}/") + @rest.get("dcim/console-ports/{id}/") def dcim_console_port(self, id: int) -> ConsolePort: pass - @get("dcim/cables/") + @rest.get("dcim/cables/") def dcim_cables( self, device: list[str] | None = None, @@ -92,18 +77,18 @@ def dcim_cables( dcim_all_cables = collect(dcim_cables, field="interface_id") - @post("dcim/cables/") + @rest.post("dcim/cables/") def dcim_cable_create(self, body: NewCable) -> Cable: pass - @post("dcim/cables/") + @rest.post("dcim/cables/") def dcim_cable_bulk_create( self, body: list[NewCable], ) -> list[Cable]: pass - @delete("dcim/cables/") + @rest.delete("dcim/cables/") def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: pass @@ -112,11 +97,11 @@ def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: [ItemToDelete(id=x) for x in body], ) - @delete("dcim/cables/{id}/") + @rest.delete("dcim/cables/{id}/") def dcim_cable_delete(self, id: int) -> None: pass - @get("dcim/devices/") + @rest.get("dcim/devices/") def dcim_devices( self, name: list[str] | None = None, @@ -147,7 +132,7 @@ def dcim_devices( dcim_all_devices_by_id = collect(dcim_devices, field="id") - @get("dcim/devices/?brief=1") + @rest.get("dcim/devices/?brief=1") def dcim_devices_brief( self, name: list[str] | None = None, @@ -180,7 +165,7 @@ def dcim_devices_brief( dcim_all_devices_brief = collect(dcim_devices_brief) dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - @get("dcim/devices/{device_id}/") + @rest.get("dcim/devices/{device_id}/") def dcim_device( self, device_id: int, @@ -188,7 +173,7 @@ def dcim_device( pass # ipam - @get("ipam/ip-addresses/") + @rest.get("ipam/ip-addresses/") def ipam_ip_addresses( self, interface_id: list[int] | None = None, @@ -200,14 +185,14 @@ def ipam_ip_addresses( ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - @get("ipam/ip-addresses/{id}/") + @rest.get("ipam/ip-addresses/{id}/") def ipam_ip_address( self, id: int, ) -> IpAddress: pass - @get("ipam/prefixes/") + @rest.get("ipam/prefixes/") def prefixes( self, prefix: list[str] | None = None, @@ -218,7 +203,7 @@ def prefixes( ipam_all_prefixes = collect(prefixes, field="prefix") - @get("ipam/fhrp-groups/") + @rest.get("ipam/fhrp-groups/") def ipam_fhrp_groups( self, id: list[int] | None = None, @@ -235,7 +220,7 @@ def ipam_fhrp_groups( ipam_all_fhrp_groups = collect(ipam_fhrp_groups) ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - @get("ipam/fhrp-group-assignments/?brief=1") + @rest.get("ipam/fhrp-group-assignments/?brief=1") def ipam_fhrp_group_assignments_brief( self, id: list[int] | None = None, diff --git a/src/annetbox/v42/client_async.py b/src/annetbox/v42/client_async.py index 304e4aa..3c2b1ba 100644 --- a/src/annetbox/v42/client_async.py +++ b/src/annetbox/v42/client_async.py @@ -1,14 +1,8 @@ from collections.abc import Iterable -from datetime import datetime - -import dateutil.parser -from adaptix import Retort, loader, name_mapping -from dataclass_rest import delete, get, post -from dataclass_rest.client_protocol import FactoryProtocol from annetbox.base.client_async import BaseNetboxClient, collect from annetbox.base.models import PagingResponse -from .client_base import BaseNetboxV42 +from .client_base import rest from .models import ( Cable, ConsolePort, @@ -26,44 +20,260 @@ ) -class NetboxV42(BaseNetboxClient, BaseNetboxV42): - dcim_all_interfaces = collect(BaseNetboxV42.dcim_interfaces, field="device_id") - dcim_all_interfaces_by_id = collect(BaseNetboxV42.dcim_interfaces, field="id") +class NetboxV42(BaseNetboxClient): + # dcim + @rest.get("dcim/interfaces/") + async def dcim_interfaces( + self, + id: list[int] | None = None, + cable_id: list[int] | None = None, + cable_id__n: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Interface]: + pass + + dcim_all_interfaces = collect(dcim_interfaces, field="device_id") + dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") + + @rest.get("dcim/interfaces/{id}/") + async def dcim_interface(self, id: int) -> Interface: + pass + + @rest.get("dcim/console-ports/") + async def dcim_console_ports( + self, + id: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[ConsolePort]: + pass + + dcim_all_console_ports = collect(dcim_console_ports, field="device_id") + dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - dcim_all_console_ports = collect(BaseNetboxV42.dcim_console_ports, field="device_id") - dcim_all_console_ports_by_id = collect(BaseNetboxV42.dcim_console_ports, field="id") + @rest.get("dcim/console-ports/{id}/") + async def dcim_console_port(self, id: int) -> ConsolePort: + pass - dcim_all_cables = collect(BaseNetboxV42.dcim_cables, field="interface_id") + @rest.get("dcim/cables/") + async def dcim_cables( + self, + device: list[str] | None = None, + device_id: list[int] | None = None, + interface_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Cable]: + pass + + dcim_all_cables = collect(dcim_cables, field="interface_id") + + @rest.post("dcim/cables/") + async def dcim_cable_create(self, body: NewCable) -> Cable: + pass + + @rest.post("dcim/cables/") + async def dcim_cable_bulk_create( + self, + body: list[NewCable], + ) -> list[Cable]: + pass + + @rest.delete("dcim/cables/") + async def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: + pass async def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: return await self._dcim_cable_bulk_delete( [ItemToDelete(id=x) for x in body], ) - dcim_all_devices = collect(BaseNetboxV42.dcim_devices) - dcim_all_devices_by_id = collect(BaseNetboxV42.dcim_devices, field="id") + @rest.delete("dcim/cables/{id}/") + async def dcim_cable_delete(self, id: int) -> None: + pass + + @rest.get("dcim/devices/") + async def dcim_devices( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Device]: + pass + + dcim_all_devices = collect(dcim_devices) + dcim_all_devices_by_id = collect(dcim_devices, field="id") + + + @rest.get("dcim/devices/?brief=1") + async def dcim_devices_brief( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + has_oob_ip: bool | None = None, + has_primary_ip: bool | None = None, + location_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Entity]: + pass + + dcim_all_devices_brief = collect(dcim_devices_brief) + dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") + + @rest.get("dcim/devices/{device_id}/") + async def dcim_device( + self, + device_id: int, + ) -> Device: + pass + + # ipam + @rest.get("ipam/ip-addresses/") + async def ipam_ip_addresses( + self, + interface_id: list[int] | None = None, + device_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[IpAddress]: + pass + + ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") + + @rest.get("ipam/ip-addresses/{id}/") + async def ipam_ip_address( + self, + id: int, + ) -> IpAddress: + pass + + @rest.get("ipam/prefixes/") + async def prefixes( + self, + prefix: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Prefix]: + pass + + ipam_all_prefixes = collect(prefixes, field="prefix") + @rest.get("ipam/vlans/") + async def ipam_vlans( + self, + id: list[int] | None = None, + vid: list[int] | None = None, + group: list[str] | None = None, + status: list[str] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + role: list[str] | None = None, + site: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vlan]: + pass - dcim_all_devices_brief = collect(BaseNetboxV42.dcim_devices_brief) - dcim_all_devices_brief_by_id = collect(BaseNetboxV42.dcim_devices_brief, field="id") + ipam_all_vlans = collect(ipam_vlans, field="vid") + ipam_all_vlans_by_id = collect(ipam_vlans, field="id") - ipam_all_ip_addresses = collect(BaseNetboxV42.ipam_ip_addresses, field="interface_id") + @rest.get("ipam/vrfs/") + async def ipam_vrfs( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vrf]: + pass - ipam_all_prefixes = collect(BaseNetboxV42.prefixes, field="prefix") + ipam_all_vrfs = collect(ipam_vrfs, field="vid") + ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") + @rest.get("ipam/fhrp-groups/") + async def ipam_fhrp_groups( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + protocol: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + related_ip: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroup]: + pass - ipam_all_vlans = collect(BaseNetboxV42.ipam_vlans, field="vid") - ipam_all_vlans_by_id = collect(BaseNetboxV42.ipam_vlans, field="id") + ipam_all_fhrp_groups = collect(ipam_fhrp_groups) + ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - ipam_all_vrfs = collect(BaseNetboxV42.ipam_vrfs, field="vid") - ipam_all_vrfs_by_id = collect(BaseNetboxV42.ipam_vrfs, field="id") + @rest.get("ipam/fhrp-group-assignments/?brief=1") + async def ipam_fhrp_group_assignments_brief( + self, + id: list[int] | None = None, + interface_id: list[int] | None = None, + device: list[str] | None = None, + device_id: list[int] | None = None, + group_id: list[int] | None = None, - ipam_all_fhrp_groups = collect(BaseNetboxV42.ipam_fhrp_groups) - ipam_all_fhrp_groups_by_id = collect(BaseNetboxV42.ipam_fhrp_groups, field="id") + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroupAssignmentBrief]: + pass ipam_all_fhrp_group_assignments = collect( - BaseNetboxV42.ipam_fhrp_group_assignments_brief, + ipam_fhrp_group_assignments_brief, ) ipam_all_fhrp_group_assignments_by_interface = collect( - BaseNetboxV42.ipam_fhrp_group_assignments_brief, field="interface_id", + ipam_fhrp_group_assignments_brief, field="interface_id", ) diff --git a/src/annetbox/v42/client_base.py b/src/annetbox/v42/client_base.py index 704a495..083e2b9 100644 --- a/src/annetbox/v42/client_base.py +++ b/src/annetbox/v42/client_base.py @@ -1,4 +1,3 @@ -from collections.abc import Iterable from datetime import datetime import dateutil.parser @@ -6,22 +5,8 @@ from descanso import RestBuilder from descanso.response_transformers import ErrorRaiser -from annetbox.base.client_sync import collect -from annetbox.base.models import PagingResponse from .models import ( - Cable, - ConsolePort, - Device, - Entity, - FHRPGroup, - FHRPGroupAssignmentBrief, - Interface, - IpAddress, - ItemToDelete, NewCable, - Prefix, - Vlan, - Vrf, ) retort = Retort(recipe=[ @@ -34,263 +19,3 @@ query_param_dumper=retort, error_raiser=ErrorRaiser(need_body=True), ) - -class BaseNetboxV42: - # dcim - @rest.get("dcim/interfaces/") - def dcim_interfaces( - self, - id: list[int] | None = None, - cable_id: list[int] | None = None, - cable_id__n: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Interface]: - pass - - dcim_all_interfaces = collect( - dcim_interfaces, field="device_id", batch_size=10, # heavy request - ) - dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") - - @rest.get("dcim/interfaces/{id}/") - def dcim_interface(self, id: int) -> Interface: - pass - - @rest.get("dcim/console-ports/") - def dcim_console_ports( - self, - id: list[int] | None = None, - device: list[str] | None = None, - device__n: list[str] | None = None, - device_id: list[int] | None = None, - device_id__n: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[ConsolePort]: - pass - - dcim_all_console_ports = collect(dcim_console_ports, field="device_id") - dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") - - @rest.get("dcim/console-ports/{id}/") - def dcim_console_port(self, id: int) -> ConsolePort: - pass - - @rest.get("dcim/cables/") - def dcim_cables( - self, - device: list[str] | None = None, - device_id: list[int] | None = None, - interface_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Cable]: - pass - - dcim_all_cables = collect(dcim_cables, field="interface_id") - - @rest.post("dcim/cables/") - def dcim_cable_create(self, body: NewCable) -> Cable: - pass - - @rest.post("dcim/cables/") - def dcim_cable_bulk_create( - self, - body: list[NewCable], - ) -> list[Cable]: - pass - - @rest.delete("dcim/cables/") - def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: - pass - - def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: - return self._dcim_cable_bulk_delete( - [ItemToDelete(id=x) for x in body], - ) - - @rest.delete("dcim/cables/{id}/") - def dcim_cable_delete(self, id: int) -> None: - pass - - @rest.get("dcim/devices/") - def dcim_devices( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Device]: - pass - - dcim_all_devices = collect(dcim_devices) - dcim_all_devices_by_id = collect(dcim_devices, field="id") - - - @rest.get("dcim/devices/?brief=1") - def dcim_devices_brief( - self, - name: list[str] | None = None, - name__empty: bool | None = None, - name__ic: list[str] | None = None, - name__ie: list[str] | None = None, - name__iew: list[str] | None = None, - name__isw: list[str] | None = None, - name__n: list[str] | None = None, - name__nic: list[str] | None = None, - name__nie: list[str] | None = None, - name__niew: list[str] | None = None, - name__nisw: list[str] | None = None, - id: list[int] | None = None, - tag: list[str] | None = None, - site: list[str] | None = None, - role: list[str] | None = None, - device_type: list[str] | None = None, - tenant: list[str] | None = None, - status: list[str] | None = None, - asset_tag: list[str] | None = None, - has_oob_ip: bool | None = None, - has_primary_ip: bool | None = None, - location_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Entity]: - pass - - dcim_all_devices_brief = collect(dcim_devices_brief) - dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") - - @rest.get("dcim/devices/{device_id}/") - def dcim_device( - self, - device_id: int, - ) -> Device: - pass - - # ipam - @rest.get("ipam/ip-addresses/") - def ipam_ip_addresses( - self, - interface_id: list[int] | None = None, - device_id: list[int] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[IpAddress]: - pass - - ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") - - @rest.get("ipam/ip-addresses/{id}/") - def ipam_ip_address( - self, - id: int, - ) -> IpAddress: - pass - - @rest.get("ipam/prefixes/") - def prefixes( - self, - prefix: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Prefix]: - pass - - ipam_all_prefixes = collect(prefixes, field="prefix") - - @rest.get("ipam/vlans/") - def ipam_vlans( - self, - id: list[int] | None = None, - vid: list[int] | None = None, - group: list[str] | None = None, - status: list[str] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - role: list[str] | None = None, - site: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vlan]: - pass - - ipam_all_vlans = collect(ipam_vlans, field="vid") - ipam_all_vlans_by_id = collect(ipam_vlans, field="id") - - @rest.get("ipam/vrfs/") - def ipam_vrfs( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - tenant: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[Vrf]: - pass - - ipam_all_vrfs = collect(ipam_vrfs, field="vid") - ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") - - @rest.get("ipam/fhrp-groups/") - def ipam_fhrp_groups( - self, - id: list[int] | None = None, - tag: list[str] | None = None, - protocol: list[str] | None = None, - name: list[str] | None = None, - name__ic: list[str] | None = None, - related_ip: list[str] | None = None, - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroup]: - pass - - ipam_all_fhrp_groups = collect(ipam_fhrp_groups) - ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - - @rest.get("ipam/fhrp-group-assignments/?brief=1") - def ipam_fhrp_group_assignments_brief( - self, - id: list[int] | None = None, - interface_id: list[int] | None = None, - device: list[str] | None = None, - device_id: list[int] | None = None, - group_id: list[int] | None = None, - - limit: int = 20, - offset: int = 0, - ) -> PagingResponse[FHRPGroupAssignmentBrief]: - pass - - ipam_all_fhrp_group_assignments = collect( - ipam_fhrp_group_assignments_brief, - ) - ipam_all_fhrp_group_assignments_by_interface = collect( - ipam_fhrp_group_assignments_brief, field="interface_id", - ) diff --git a/src/annetbox/v42/client_sync.py b/src/annetbox/v42/client_sync.py index d356bcc..1019b13 100644 --- a/src/annetbox/v42/client_sync.py +++ b/src/annetbox/v42/client_sync.py @@ -1,51 +1,281 @@ from collections.abc import Iterable from annetbox.base.client_sync import BaseNetboxClient, collect -from .client_base import BaseNetboxV42 +from annetbox.base.models import PagingResponse +from .client_base import rest from .models import ( + Cable, + ConsolePort, + Device, + Entity, + FHRPGroup, + FHRPGroupAssignmentBrief, + Interface, + IpAddress, ItemToDelete, + NewCable, + Prefix, + Vlan, + Vrf, ) -class NetboxV42(BaseNetboxClient, BaseNetboxV42): +class NetboxV42(BaseNetboxClient): + # dcim + @rest.get("dcim/interfaces/") + def dcim_interfaces( + self, + id: list[int] | None = None, + cable_id: list[int] | None = None, + cable_id__n: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Interface]: + pass + dcim_all_interfaces = collect( - BaseNetboxV42.dcim_interfaces, field="device_id", batch_size=10, # heavy request + dcim_interfaces, field="device_id", batch_size=10, # heavy request ) - dcim_all_interfaces_by_id = collect(BaseNetboxV42.dcim_interfaces, field="id") + dcim_all_interfaces_by_id = collect(dcim_interfaces, field="id") + + @rest.get("dcim/interfaces/{id}/") + def dcim_interface(self, id: int) -> Interface: + pass + + @rest.get("dcim/console-ports/") + def dcim_console_ports( + self, + id: list[int] | None = None, + device: list[str] | None = None, + device__n: list[str] | None = None, + device_id: list[int] | None = None, + device_id__n: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[ConsolePort]: + pass + + dcim_all_console_ports = collect(dcim_console_ports, field="device_id") + dcim_all_console_ports_by_id = collect(dcim_console_ports, field="id") + + @rest.get("dcim/console-ports/{id}/") + def dcim_console_port(self, id: int) -> ConsolePort: + pass - dcim_all_console_ports = collect(BaseNetboxV42.dcim_console_ports, field="device_id") - dcim_all_console_ports_by_id = collect(BaseNetboxV42.dcim_console_ports, field="id") + @rest.get("dcim/cables/") + def dcim_cables( + self, + device: list[str] | None = None, + device_id: list[int] | None = None, + interface_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Cable]: + pass - dcim_all_cables = collect(BaseNetboxV42.dcim_cables, field="interface_id") + dcim_all_cables = collect(dcim_cables, field="interface_id") + + @rest.post("dcim/cables/") + def dcim_cable_create(self, body: NewCable) -> Cable: + pass + + @rest.post("dcim/cables/") + def dcim_cable_bulk_create( + self, + body: list[NewCable], + ) -> list[Cable]: + pass + + @rest.delete("dcim/cables/") + def _dcim_cable_bulk_delete(self, body: list[ItemToDelete]) -> None: + pass def dcim_cable_bulk_delete(self, body: Iterable[int]) -> None: return self._dcim_cable_bulk_delete( [ItemToDelete(id=x) for x in body], ) - dcim_all_devices = collect(BaseNetboxV42.dcim_devices) - dcim_all_devices_by_id = collect(BaseNetboxV42.dcim_devices, field="id") + @rest.delete("dcim/cables/{id}/") + def dcim_cable_delete(self, id: int) -> None: + pass + + @rest.get("dcim/devices/") + def dcim_devices( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Device]: + pass + + dcim_all_devices = collect(dcim_devices) + dcim_all_devices_by_id = collect(dcim_devices, field="id") + + + @rest.get("dcim/devices/?brief=1") + def dcim_devices_brief( + self, + name: list[str] | None = None, + name__empty: bool | None = None, + name__ic: list[str] | None = None, + name__ie: list[str] | None = None, + name__iew: list[str] | None = None, + name__isw: list[str] | None = None, + name__n: list[str] | None = None, + name__nic: list[str] | None = None, + name__nie: list[str] | None = None, + name__niew: list[str] | None = None, + name__nisw: list[str] | None = None, + id: list[int] | None = None, + tag: list[str] | None = None, + site: list[str] | None = None, + role: list[str] | None = None, + device_type: list[str] | None = None, + tenant: list[str] | None = None, + status: list[str] | None = None, + asset_tag: list[str] | None = None, + has_oob_ip: bool | None = None, + has_primary_ip: bool | None = None, + location_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Entity]: + pass + + dcim_all_devices_brief = collect(dcim_devices_brief) + dcim_all_devices_brief_by_id = collect(dcim_devices_brief, field="id") + + @rest.get("dcim/devices/{device_id}/") + def dcim_device( + self, + device_id: int, + ) -> Device: + pass + + # ipam + @rest.get("ipam/ip-addresses/") + def ipam_ip_addresses( + self, + interface_id: list[int] | None = None, + device_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[IpAddress]: + pass + + ipam_all_ip_addresses = collect(ipam_ip_addresses, field="interface_id") + + @rest.get("ipam/ip-addresses/{id}/") + def ipam_ip_address( + self, + id: int, + ) -> IpAddress: + pass + + @rest.get("ipam/prefixes/") + def prefixes( + self, + prefix: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Prefix]: + pass + + ipam_all_prefixes = collect(prefixes, field="prefix") + + @rest.get("ipam/vlans/") + def ipam_vlans( + self, + id: list[int] | None = None, + vid: list[int] | None = None, + group: list[str] | None = None, + status: list[str] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + role: list[str] | None = None, + site: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vlan]: + pass - dcim_all_devices_brief = collect(BaseNetboxV42.dcim_devices_brief) - dcim_all_devices_brief_by_id = collect(BaseNetboxV42.dcim_devices_brief, field="id") + ipam_all_vlans = collect(ipam_vlans, field="vid") + ipam_all_vlans_by_id = collect(ipam_vlans, field="id") - ipam_all_ip_addresses = collect(BaseNetboxV42.ipam_ip_addresses, field="interface_id") + @rest.get("ipam/vrfs/") + def ipam_vrfs( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + tenant: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[Vrf]: + pass - ipam_all_prefixes = collect(BaseNetboxV42.prefixes, field="prefix") + ipam_all_vrfs = collect(ipam_vrfs, field="vid") + ipam_all_vrfs_by_id = collect(ipam_vrfs, field="id") - ipam_all_vlans = collect(BaseNetboxV42.ipam_vlans, field="vid") - ipam_all_vlans_by_id = collect(BaseNetboxV42.ipam_vlans, field="id") + @rest.get("ipam/fhrp-groups/") + def ipam_fhrp_groups( + self, + id: list[int] | None = None, + tag: list[str] | None = None, + protocol: list[str] | None = None, + name: list[str] | None = None, + name__ic: list[str] | None = None, + related_ip: list[str] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroup]: + pass - ipam_all_vrfs = collect(BaseNetboxV42.ipam_vrfs, field="vid") - ipam_all_vrfs_by_id = collect(BaseNetboxV42.ipam_vrfs, field="id") + ipam_all_fhrp_groups = collect(ipam_fhrp_groups) + ipam_all_fhrp_groups_by_id = collect(ipam_fhrp_groups, field="id") - ipam_all_fhrp_groups = collect(BaseNetboxV42.ipam_fhrp_groups) - ipam_all_fhrp_groups_by_id = collect(BaseNetboxV42.ipam_fhrp_groups, field="id") + @rest.get("ipam/fhrp-group-assignments/?brief=1") + def ipam_fhrp_group_assignments_brief( + self, + id: list[int] | None = None, + interface_id: list[int] | None = None, + device: list[str] | None = None, + device_id: list[int] | None = None, + group_id: list[int] | None = None, + limit: int = 20, + offset: int = 0, + ) -> PagingResponse[FHRPGroupAssignmentBrief]: + pass ipam_all_fhrp_group_assignments = collect( - BaseNetboxV42.ipam_fhrp_group_assignments_brief, + ipam_fhrp_group_assignments_brief, ) ipam_all_fhrp_group_assignments_by_interface = collect( - BaseNetboxV42.ipam_fhrp_group_assignments_brief, field="interface_id", + ipam_fhrp_group_assignments_brief, field="interface_id", )