rest

django_spire.contrib.rest

__all__ = ['BaseRestHttpConnector', 'DjangoModelRestSchema', 'RestSchema', 'RestSchemaSet'] module-attribute

RestSchema

Bases: ABC, BaseModel

Base class for REST API schemas that should map to django models with Django-like .objects API.

Provides a Django QuerySet - like interface for interacting with external REST-based data sources.

Subclasses need to assign an instance of a RestSchemaSet subclass to the objects class: objects = MySchemaSet

objects class-attribute

__pydantic_init_subclass__ classmethod

Source code in django_spire/contrib/rest/schema/schema.py

@classmethod
def __pydantic_init_subclass__(cls, **kwargs):
    super().__pydantic_init_subclass__(**kwargs)

    if not isabstract(cls):
        from django_spire.contrib.rest.schema.schemaset import RestSchemaSet
        objects = getattr(cls, 'objects', None)

        if not isinstance(objects, RestSchemaSet):
            message = f'{cls.__name__}.objects must be an instance of a RestSchemaSet subclass.'
            raise TypeError(message)

        cls.objects = objects.__class__(schema_class=cls)

RestSchemaSet

Bases: ABC, Generic[TSchema]

Source code in django_spire/contrib/rest/schema/schemaset.py

def __init__(
    self,
    schema_class: type[TSchema] | None = None,
    *,
    # Internal state for cloning
    _request_params: dict[str, Any] | None = None,
    _filters: list[Callable[[TSchema], bool]] | None = None,
    _excludes: list[Callable[[TSchema], bool]] | None = None,
    _ordering: list[tuple[str, bool]] | None = None,
    _limit: int | None = None,
    _offset: int = 0,
    _cached_results: list[TSchema] | None = None,
):
    self._request_params = _request_params
    self.schema_class = schema_class

    self._filters = _filters or []
    self._excludes = _excludes or []
    self._ordering = _ordering or []
    self._limit = _limit
    self._offset = _offset
    self._cached_results = _cached_results

connector instance-attribute

schema_class = schema_class instance-attribute

as_manager classmethod

Simply returns an instance of itself with no schema class. Provides no real value other than a more Django model-like objects assignment in RestSchema classes.

Source code in django_spire/contrib/rest/schema/schemaset.py

@classmethod
def as_manager(cls) -> Self:
    """
    Simply returns an instance of itself with no schema class. Provides no real
    value other than a more Django model-like objects assignment in RestSchema classes.
    """
    return cls()

__iter__

Source code in django_spire/contrib/rest/schema/schemaset.py

def __iter__(self) -> Iterator[TSchema]:
    return iter(self._evaluate())

__len__

Source code in django_spire/contrib/rest/schema/schemaset.py

def __len__(self) -> int:
    return len(self._evaluate())

__bool__

Source code in django_spire/contrib/rest/schema/schemaset.py

def __bool__(self) -> bool:
    return bool(self._evaluate())

__repr__

Source code in django_spire/contrib/rest/schema/schemaset.py

def __repr__(self) -> str:
    name = self.schema_class.__name__ if self.schema_class else 'Unknown'
    return f"<RestSchemaSet [{name}]>"

__getitem__

Source code in django_spire/contrib/rest/schema/schemaset.py

def __getitem__(self, key: int | slice) -> TSchema | Self:
    if isinstance(key, int):
        if key < 0:
            return self._evaluate()[key]
        result = self.offset(key).limit(1).first()
        if result is None:
            raise IndexError("RestSchemaQuerySet index out of range")
        return result
    elif isinstance(key, slice):
        clone = self
        start = key.start or 0
        if start:
            clone = clone.offset(start)
        if key.stop is not None:
            clone = clone.limit(key.stop - start)
        return clone
    raise TypeError(f"Invalid index type: {type(key)}")

with_request_params

Set or merge request parameters for downstream API calls.

Source code in django_spire/contrib/rest/schema/schemaset.py

def with_request_params(
    self,
    **kwargs,
) -> Self:
    """Set or merge request parameters for downstream API calls."""
    if self._request_params:
        kwargs = {
            **self._request_params,
            **kwargs
        }

    return self._clone(_request_params=kwargs)

all

Return a clone of the schema set with all results.

Source code in django_spire/contrib/rest/schema/schemaset.py

def all(
    self,
) -> Self:
    """Return a clone of the schema set with all results."""
    return self._clone()

filter

Filter results by predicate and/or field lookups.

Examples: .filter(lambda p: p.weight > 100) .filter(name="pikachu") .filter(type__name="electric")

Source code in django_spire/contrib/rest/schema/schemaset.py

def filter(
    self,
    predicate: Callable[[TSchema], bool] | None = None,
    **kwargs,
) -> Self:
    """
    Filter results by predicate and/or field lookups.

    Examples:
        .filter(lambda p: p.weight > 100)
        .filter(name="pikachu")
        .filter(type__name="electric")
    """
    new_filters = list(self._filters)
    if predicate:
        new_filters.append(predicate)
    for key, value in kwargs.items():
        new_filters.append(self._make_predicate(key, value))
    return self._clone(_filters=new_filters)

exclude

Exclude results matching predicate or field lookups.

Source code in django_spire/contrib/rest/schema/schemaset.py

def exclude(
    self,
    predicate: Callable[[TSchema], bool] | None = None,
    **kwargs,
) -> Self:
    """Exclude results matching predicate or field lookups."""
    new_excludes = list(self._excludes)
    if predicate:
        new_excludes.append(predicate)
    for key, value in kwargs.items():
        new_excludes.append(self._make_predicate(key, value))
    return self._clone(_excludes=new_excludes)

order_by

Order by fields. Prefix with '-' for descending.

Examples: .order_by('name') .order_by('-weight', 'name')

Source code in django_spire/contrib/rest/schema/schemaset.py

def order_by(self, *fields: str) -> Self:
    """
    Order by fields. Prefix with '-' for descending.

    Examples:
        .order_by('name')
        .order_by('-weight', 'name')
    """
    ordering = []
    for field in fields:
        if field.startswith('-'):
            ordering.append((field[1:], True))
        else:
            ordering.append((field, False))
    return self._clone(_ordering=ordering)

limit

Limit the number of results to n.

Source code in django_spire/contrib/rest/schema/schemaset.py

def limit(self, n: int) -> Self:
    """Limit the number of results to n."""
    return self._clone(_limit=n)

offset

Skip the first n results.

Source code in django_spire/contrib/rest/schema/schemaset.py

def offset(self, n: int) -> Self:
    """Skip the first n results."""
    return self._clone(_offset=n)

first

Return the first result, or None if the set is empty.

Source code in django_spire/contrib/rest/schema/schemaset.py

def first(self) -> TSchema | None:
    """Return the first result, or None if the set is empty."""
    results = self._evaluate()
    return results[0] if results else None

last

Return the last result, or None if the set is empty.

Source code in django_spire/contrib/rest/schema/schemaset.py

def last(self) -> TSchema | None:
    """Return the last result, or None if the set is empty."""
    results = self._evaluate()
    return results[-1] if results else None

count

Return the number of results.

Source code in django_spire/contrib/rest/schema/schemaset.py

def count(self) -> int:
    """Return the number of results."""
    return len(self)

exists

Return True if there is at least one result.

Source code in django_spire/contrib/rest/schema/schemaset.py

def exists(self) -> bool:
    """Return True if there is at least one result."""
    return bool(self)

get

Return exactly one result matching kwargs. Raises LookupError if zero or multiple results.

Source code in django_spire/contrib/rest/schema/schemaset.py

def get(
    self,
    request_params: dict[str, Any] | None = None,
    **kwargs,
) -> TSchema:
    """
    Return exactly one result matching kwargs.
    Raises LookupError if zero or multiple results.
    """

    if not self._cached_results:
        try:
            # Direct fetch by ID/params - try using _read_one from connector
            if request_params:
                result =  self._read_one(**request_params)
            else:
                result = self._read_one()

            if result and not isinstance(result, self.schema_class):
                raise ValueError(
                    f'_read_one for RestSchemaSet subclass {self.__class__.__name__} returned invalid type. It must return an instance of {self.schema_class.__name__}')

        except NotImplementedError:
            # TODO: log warning if request_params are passed signalling that there is no _read_one available to use the params on
            pass

    results = list(self.filter(**kwargs) if kwargs else self)
    schema_name = self.schema_class.__name__ if self.schema_class else 'object'
    if len(results) == 0:
        raise LookupError(f"No {schema_name} found")
    if len(results) > 1:
        raise LookupError(f"Multiple {schema_name} found")
    return results[0]

values_list

Extract field values from results.

Source code in django_spire/contrib/rest/schema/schemaset.py

def values_list(self, *fields: str, flat: bool = False) -> list[tuple] | list[Any]:
    """Extract field values from results."""
    if flat and len(fields) != 1:
        raise ValueError("flat=True requires exactly one field")

    results = self._evaluate()
    if flat:
        return [self._get_attr(item, fields[0]) for item in results]
    return [tuple(self._get_attr(item, f) for f in fields) for item in results]

DjangoModelRestSchema

Bases: RestSchema, Generic[TModel], ABC

from_django_model abstractmethod classmethod

Source code in django_spire/contrib/rest/schema/django_model_schema.py

@classmethod
@abstractmethod
def from_django_model(cls, model: type[TModel]) -> Self:
    raise NotImplementedError()

to_django_model abstractmethod classmethod

Source code in django_spire/contrib/rest/schema/django_model_schema.py

@classmethod
@abstractmethod
def to_django_model(cls) -> TModel:
    raise NotImplementedError()

BaseRestHttpConnector

Bases: ABC

Source code in django_spire/contrib/rest/connector/connector.py

def __init__(self):
    self._validate_url(self.base_url)

base_url instance-attribute

base_path = '' class-attribute instance-attribute

base_headers = {} class-attribute instance-attribute

timeout = 30 class-attribute instance-attribute

max_retries = 3 class-attribute instance-attribute

auth property

__init_subclass__

Source code in django_spire/contrib/rest/connector/connector.py

def __init_subclass__(cls, **kwargs):
    if not inspect.isabstract(cls):
        required_attributes = ['base_url']
        for attribute in required_attributes:
            if getattr(cls, attribute, None) is None:
                message = f'{attribute} is required'
                raise ImproperlyConfigured(message)

        cls._validate_url(cls.base_url)

request

Source code in django_spire/contrib/rest/connector/connector.py

def request(
    self,
    method: str,
    path: str | None = None,
    headers: dict[str, str] | None = None,
    auth: bool | AuthBase | None = True,
    **kwargs,
) -> requests.Response:
    merged_headers = {**self.base_headers, **(headers or {})}

    if isinstance(auth, bool):
        if auth:
            auth = self.auth
        else:
            auth = None

    retries = 0

    response = requests.request(
        method=method,
        url=self._build_url(path),
        headers=merged_headers,
        auth=auth,
        timeout=self.timeout,
        **kwargs,
    )

    for i in range(self.max_retries):
        try:
            response.raise_for_status()

            break
        except requests.exceptions.Timeout as e:
            retries += 1

            if retries > 1:
                time.sleep(retries * 2)

            if retries >= self.max_retries:
                raise RestConnectorTimeoutException from e

            response = requests.request(
                method=method,
                url=self._build_url(path),
                headers=merged_headers,
                auth=auth,
                timeout=self.timeout,
                **kwargs,
            )

        except HTTPError as e:
            raise RestConnectorError from e

    return response

get

Source code in django_spire/contrib/rest/connector/connector.py

def get(self, path: str | None = None, **kwargs) -> requests.Response:
    return self.request('GET', path, **kwargs)

post

Source code in django_spire/contrib/rest/connector/connector.py

def post(self, path: str | None = None, **kwargs) -> requests.Response:
    return self.request('POST', path, **kwargs)

put

Source code in django_spire/contrib/rest/connector/connector.py

def put(self, path: str | None = None, **kwargs) -> requests.Response:
    return self.request('PUT', path, **kwargs)

patch

Source code in django_spire/contrib/rest/connector/connector.py

def patch(self, path: str | None = None, **kwargs) -> requests.Response:
    return self.request('PATCH', path, **kwargs)

delete

Source code in django_spire/contrib/rest/connector/connector.py

def delete(self, path: str | None = None, **kwargs) -> requests.Response:
    return self.request('DELETE', path, **kwargs)