|
| 1 | +import math |
| 2 | +from dataclasses import dataclass, field |
| 3 | +from typing import Any, ClassVar, Self, override |
| 4 | + |
| 5 | +from box import Box |
| 6 | +from httpx import Response |
| 7 | + |
| 8 | + |
| 9 | +@dataclass |
| 10 | +class Pagination: |
| 11 | + """Provides pagination information.""" |
| 12 | + |
| 13 | + limit: int = 0 |
| 14 | + offset: int = 0 |
| 15 | + total: int = 0 |
| 16 | + |
| 17 | + def has_next(self) -> bool: |
| 18 | + """Returns True if there is a next page.""" |
| 19 | + return self.num_page() + 1 < self.total_pages() |
| 20 | + |
| 21 | + def num_page(self) -> int: |
| 22 | + """Returns the current page number starting the first page as 0.""" |
| 23 | + if self.limit == 0: |
| 24 | + return 0 |
| 25 | + return self.offset // self.limit |
| 26 | + |
| 27 | + def total_pages(self) -> int: |
| 28 | + """Returns the total number of pages.""" |
| 29 | + if self.limit == 0: |
| 30 | + return 0 |
| 31 | + return math.ceil(self.total / self.limit) |
| 32 | + |
| 33 | + def next_offset(self) -> int: |
| 34 | + """Returns the next offset as an integer for the next page.""" |
| 35 | + return self.offset + self.limit |
| 36 | + |
| 37 | + |
| 38 | +@dataclass |
| 39 | +class Meta: |
| 40 | + """Provides meta-information about the pagination, ignored fields and the response.""" |
| 41 | + |
| 42 | + response: Response |
| 43 | + pagination: Pagination = field(default_factory=Pagination) |
| 44 | + ignored: list[str] = field(default_factory=list) |
| 45 | + |
| 46 | + @classmethod |
| 47 | + def from_response(cls, response: Response) -> Self: |
| 48 | + """Creates a meta object from response.""" |
| 49 | + meta_data = response.json().get("$meta", {}) |
| 50 | + if not isinstance(meta_data, dict): |
| 51 | + raise TypeError("Response $meta must be a dict.") |
| 52 | + |
| 53 | + return cls( |
| 54 | + ignored=meta_data.get("ignored", []), |
| 55 | + pagination=Pagination(**meta_data.get("pagination", {})), |
| 56 | + response=response, |
| 57 | + ) |
| 58 | + |
| 59 | + |
| 60 | +ResourceData = dict[str, Any] |
| 61 | + |
| 62 | + |
| 63 | +class GenericResource: |
| 64 | + """Provides a base resource to interact with api data using fluent interfaces.""" |
| 65 | + |
| 66 | + _data_key: ClassVar[str] = "data" |
| 67 | + _safe_attributes: ClassVar[list[str]] = ["meta", "_resource_data"] |
| 68 | + |
| 69 | + def __init__(self, resource_data: ResourceData | None = None, meta: Meta | None = None) -> None: |
| 70 | + self.meta = meta |
| 71 | + self._resource_data = Box(resource_data or {}, camel_killer_box=True, default_box=False) |
| 72 | + |
| 73 | + def __getattr__(self, attribute: str) -> Box | Any: |
| 74 | + """Returns the resource data.""" |
| 75 | + return self._resource_data.__getattr__(attribute) # type: ignore[no-untyped-call] |
| 76 | + |
| 77 | + @override |
| 78 | + def __setattr__(self, attribute: str, attribute_value: Any) -> None: |
| 79 | + """Sets the resource data.""" |
| 80 | + if attribute in self._safe_attributes: |
| 81 | + object.__setattr__(self, attribute, attribute_value) |
| 82 | + return |
| 83 | + |
| 84 | + self._resource_data.__setattr__(attribute, attribute_value) # type: ignore[no-untyped-call] |
| 85 | + |
| 86 | + @classmethod |
| 87 | + def from_response(cls, response: Response) -> Self: |
| 88 | + """Creates a resource from a response. |
| 89 | +
|
| 90 | + Expected a Response with json data with two keys: data and $meta. |
| 91 | + """ |
| 92 | + response_data = response.json().get(cls._data_key) |
| 93 | + if not isinstance(response_data, dict): |
| 94 | + raise TypeError("Response data must be a dict.") |
| 95 | + meta = Meta.from_response(response) |
| 96 | + return cls(response_data, meta) |
| 97 | + |
| 98 | + def to_dict(self) -> dict[str, Any]: |
| 99 | + """Returns the resource as a dictionary.""" |
| 100 | + return self._resource_data.to_dict() |
0 commit comments