Skip to content

katana_public_api_client.api_wrapper._resource

katana_public_api_client.api_wrapper._resource

Generic async CRUD resource that delegates to generated API modules.

Classes

Resource(client, config)

Thin async CRUD wrapper around a single Katana API resource.

Each method delegates to the corresponding generated asyncio_detailed function, unwraps the response, and returns the raw attrs model.

Generated modules are imported lazily on first call and cached for reuse.

Source code in katana_public_api_client/api_wrapper/_resource.py 
28
29
30
31
def __init__(self, client: AuthenticatedClient, config: ResourceConfig) -> None:
    self._client = client
    self._config = config
    self._module_cache: dict[str, ModuleType] = {}
Functions
create(body, **kwargs) async

Create a new resource.

Source code in katana_public_api_client/api_wrapper/_resource.py 
86
87
88
89
90
91
async def create(self, body: Any, **kwargs: Any) -> Any:
    """Create a new resource."""
    name = self._require("create", self._config.create)
    mod = self._load_module(name)
    response = await mod.asyncio_detailed(client=self._client, body=body, **kwargs)
    return unwrap(response)
delete(resource_id, **kwargs) async

Delete a resource by ID. Raises on error, returns None.

Source code in katana_public_api_client/api_wrapper/_resource.py 
102
103
104
105
106
107
108
109
110
111
112
async def delete(self, resource_id: int, **kwargs: Any) -> None:
    """Delete a resource by ID.  Raises on error, returns ``None``."""
    name = self._require("delete", self._config.delete)
    mod = self._load_module(name)
    response = await mod.asyncio_detailed(
        resource_id, client=self._client, **kwargs
    )
    # DELETE endpoints return 204 with parsed=None on success;
    # unwrap() would raise on None, so check status first.
    if not is_success(response):
        unwrap(response)  # raises the appropriate typed error
get(resource_id, **kwargs) async

Fetch a single resource by ID.

Source code in katana_public_api_client/api_wrapper/_resource.py 
70
71
72
73
74
75
76
77
async def get(self, resource_id: int, **kwargs: Any) -> Any:
    """Fetch a single resource by ID."""
    name = self._require("get", self._config.get_one)
    mod = self._load_module(name)
    response = await mod.asyncio_detailed(
        resource_id, client=self._client, **kwargs
    )
    return unwrap(response)
list(**kwargs) async

Fetch all resources (with optional filters).

Source code in katana_public_api_client/api_wrapper/_resource.py 
79
80
81
82
83
84
async def list(self, **kwargs: Any) -> _list[Any]:
    """Fetch all resources (with optional filters)."""
    name = self._require("list", self._config.get_all)
    mod = self._load_module(name)
    response = await mod.asyncio_detailed(client=self._client, **kwargs)
    return unwrap_data(response, default=[])
update(resource_id, body, **kwargs) async

Update an existing resource by ID.

Source code in katana_public_api_client/api_wrapper/_resource.py 
 93
 94
 95
 96
 97
 98
 99
100
async def update(self, resource_id: int, body: Any, **kwargs: Any) -> Any:
    """Update an existing resource by ID."""
    name = self._require("update", self._config.update)
    mod = self._load_module(name)
    response = await mod.asyncio_detailed(
        resource_id, client=self._client, body=body, **kwargs
    )
    return unwrap(response)

Functions