schemaset

django_spire.contrib.rest.schema.schemaset

TSchema = TypeVar('TSchema', bound='RestSchema') module-attribute

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]