QuerySets¶
QuerySet classes provide the interface for querying and filtering data.
Synchronous QuerySets¶
- class django_api_orm.base.QuerySet(model_class, manager)[source]¶
Bases:
Generic[T]Django-like QuerySet for filtering and retrieving API resources.
Provides lazy evaluation with result caching and chainable filter methods.
Example
>>> queryset = Policy.objects.filter(status='active') >>> queryset = queryset.order_by('-created_at') >>> for policy in queryset: # Executes query here ... print(policy.policy_number)
- filter(**kwargs)[source]¶
Filter QuerySet by given parameters.
- Parameters:
**kwargs (Any) – Field lookups (e.g., status=’active’, id=123)
- Returns:
New QuerySet with filters applied
- Return type:
QuerySet[T]
Example
>>> Policy.objects.filter(status='active', premium_amount__gte=1000)
- exclude(**kwargs)[source]¶
Exclude results matching given parameters.
- Parameters:
**kwargs (Any) – Field lookups to exclude
- Returns:
New QuerySet with exclusions applied
- Return type:
QuerySet[T]
Example
>>> Policy.objects.exclude(status='cancelled')
- all()[source]¶
Return a copy of this QuerySet.
- Returns:
New QuerySet (clone)
- Return type:
QuerySet[T]
- order_by(*fields)[source]¶
Order results by given fields.
- Parameters:
*fields (str) – Field names (prefix with ‘-’ for descending)
- Returns:
New QuerySet with ordering applied
- Return type:
QuerySet[T]
Example
>>> Policy.objects.order_by('-created_at', 'policy_number')
- first()[source]¶
Get the first result or None.
- Returns:
First model instance or None
- Return type:
T | None
Example
>>> policy = Policy.objects.filter(status='active').first()
- last()[source]¶
Get the last result or None.
- Returns:
Last model instance or None
- Return type:
T | None
- __getitem__(key)[source]¶
Support slicing and indexing.
- Parameters:
- Returns:
Model instance (for int) or QuerySet (for slice)
- Return type:
T | QuerySet[T]
Example
>>> policies = Policy.objects.all() >>> first = policies[0] # Get first >>> subset = policies[10:20] # Get slice
- get(**kwargs)[source]¶
Get a single object matching the criteria.
- Parameters:
**kwargs (Any) – Field lookups
- Returns:
Single model instance
- Raises:
DoesNotExist – If no results found
MultipleObjectsReturned – If multiple results found
- Return type:
T
Example
>>> policy = Policy.objects.get(id=123)
- exists()[source]¶
Check if any results exist.
- Returns:
True if results exist, False otherwise
- Return type:
Example
>>> if Policy.objects.filter(status='active').exists(): ... print("Active policies found")
- count()[source]¶
Get count of results.
- Returns:
Number of results
- Return type:
Example
>>> count = Policy.objects.filter(status='active').count()
- __iter__()[source]¶
Make QuerySet iterable.
- Returns:
Iterator over model instances
- Return type:
Iterator[T]
Example
>>> for policy in Policy.objects.filter(status='active'): ... print(policy.policy_number)
- values(*fields)[source]¶
Return list of dictionaries instead of model instances.
- Parameters:
*fields (str) – Field names to include (all if not specified)
- Returns:
List of dictionaries
- Return type:
Example
>>> policies = Policy.objects.values('id', 'policy_number') >>> # [{'id': 1, 'policy_number': 'POL-001'}, ...]
- values_list(*fields, flat=False)[source]¶
Return list of tuples instead of model instances.
- Parameters:
- Returns:
List of tuples (or flat list if flat=True)
- Return type:
Example
>>> ids = Policy.objects.values_list('id', flat=True) >>> # [1, 2, 3, ...] >>> pairs = Policy.objects.values_list('id', 'policy_number') >>> # [(1, 'POL-001'), (2, 'POL-002'), ...]
Example Usage¶
from django_api_orm import APIModel
class User(APIModel):
_schema_class = UserSchema
_endpoint = "/api/v1/users/"
# Get QuerySet via manager
queryset = User.objects.all()
# Filter
active_users = User.objects.filter(active=True)
# Chain filters
admin_users = User.objects.filter(active=True).filter(role="admin")
# Order
ordered_users = User.objects.order_by("-created_at")
# Slice
first_ten = User.objects.all()[:10]
# Iterate
for user in User.objects.filter(active=True):
print(user.name)
Asynchronous QuerySets¶
- class django_api_orm.async_base.AsyncQuerySet(model_class, manager)[source]¶
Bases:
Generic[T]Django-like AsyncQuerySet for filtering and retrieving API resources.
Provides lazy evaluation with result caching and chainable filter methods. All methods that fetch data are async.
Example
>>> queryset = Policy.objects.filter(status='active') >>> queryset = queryset.order_by('-created_at') >>> async for policy in queryset: # Executes query here ... print(policy.policy_number)
- __init__(model_class, manager)[source]¶
Initialize AsyncQuerySet.
- Parameters:
model_class (type[T]) – The model class this QuerySet represents
manager (AsyncManager[T]) – The manager that created this QuerySet
- __getitem__(key)[source]¶
Support slicing (indexing not supported for async).
- Parameters:
key (int | slice) – Slice object (int indexing not supported in async context)
- Returns:
AsyncQuerySet with limits applied
- Return type:
Example
>>> subset = Policy.objects.all()[10:20] # Get slice >>> async for policy in subset: ... print(policy.policy_number)
Note
Integer indexing like queryset[0] is not supported for async querysets. Use await queryset.first() instead.
- filter(**kwargs)[source]¶
Filter QuerySet by given parameters.
- Parameters:
**kwargs (Any) – Field lookups
- Returns:
New AsyncQuerySet with filters applied
- Return type:
Example
>>> Policy.objects.filter(status='active', premium_amount__gte=1000)
- exclude(**kwargs)[source]¶
Exclude results matching given parameters.
- Parameters:
**kwargs (Any) – Field lookups to exclude
- Returns:
New AsyncQuerySet with exclusions applied
- Return type:
Example
>>> Policy.objects.exclude(status='cancelled')
- order_by(*fields)[source]¶
Order results by given fields.
- Parameters:
*fields (str) – Field names (prefix with ‘-’ for descending)
- Returns:
New AsyncQuerySet with ordering applied
- Return type:
Example
>>> Policy.objects.order_by('-created_at', 'policy_number')
- async first()[source]¶
Get the first result or None.
- Returns:
First model instance or None
- Return type:
T | None
Example
>>> policy = await Policy.objects.filter(status='active').first()
- async last()[source]¶
Get the last result or None.
- Returns:
Last model instance or None
- Return type:
T | None
- async get(**kwargs)[source]¶
Get a single object matching the criteria.
- Parameters:
**kwargs (Any) – Field lookups
- Returns:
Single model instance
- Raises:
DoesNotExist – If no results found
MultipleObjectsReturned – If multiple results found
- Return type:
T
Example
>>> policy = await Policy.objects.get(id=123)
- async exists()[source]¶
Check if any results exist.
- Returns:
True if results exist, False otherwise
- Return type:
Example
>>> if await Policy.objects.filter(status='active').exists(): ... print("Active policies found")
- async count()[source]¶
Get count of results.
- Returns:
Number of results
- Return type:
Example
>>> count = await Policy.objects.filter(status='active').count()
- async __aiter__()[source]¶
Make QuerySet async iterable.
- Returns:
Async iterator over model instances
- Return type:
Example
>>> async for policy in Policy.objects.filter(status='active'): ... print(policy.policy_number)
- async alen()[source]¶
Get length of results (async version).
- Returns:
Number of cached results
- Return type:
- async values(*fields)[source]¶
Return list of dictionaries instead of model instances.
- Parameters:
*fields (str) – Field names to include (all if not specified)
- Returns:
List of dictionaries
- Return type:
Example
>>> policies = await Policy.objects.values('id', 'policy_number') >>> # [{'id': 1, 'policy_number': 'POL-001'}, ...]
Example Usage¶
from django_api_orm import AsyncAPIModel
class User(AsyncAPIModel):
_schema_class = UserSchema
_endpoint = "/api/v1/users/"
# Get QuerySet via manager
queryset = User.objects.all()
# Filter (returns QuerySet, no await needed)
active_users = User.objects.filter(active=True)
# Chain filters
admin_users = User.objects.filter(active=True).filter(role="admin")
# Order
ordered_users = User.objects.order_by("-created_at")
# Slice
first_ten = User.objects.all()[:10]
# Iterate (await needed)
async for user in User.objects.filter(active=True):
print(user.name)
# Get single (await needed)
user = await User.objects.get(id=1)
# Count (await needed)
count = await User.objects.count()