Descanso

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()
 

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()
 

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]

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):
+    ...

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):
+    ...

src/annetbox/base/status.py

@@ -0,0 +1,18 @@
+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,
+    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: ...

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,15 +29,15 @@ 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,
     ) -> Device:
         pass
 
     # ipam
-    @get("ipam/ip-addresses/")
+    @rest.get("ipam/ip-addresses/")
     async def ipam_ip_addresses(
         self,
         interface_id: list[int] | None = None,

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),
+)