Managers¶

Manager classes provide the interface for creating and managing model instances.

Synchronous Managers¶

class django_api_orm.base.Manager(model_class, client)[source]¶

Bases: Generic[T]

Django-like Manager for model querying.

Handles creation and retrieval of model instances.

Example

>>> class Policy(APIModel):
...     objects = Manager()
>>> policies = Policy.objects.filter(status='active')

__init__(model_class, client)[source]¶

Initialize Manager.

Parameters:

model_class (type[T]) – The model class this manager handles
client (ServiceClient) – HTTP client for API requests

get_endpoint()[source]¶

Get the API endpoint for this model.

Returns:: API endpoint path
Return type:: str

all()[source]¶

Get all objects.

Returns:: QuerySet of all objects
Return type:: QuerySet[T]

Example

>>> all_policies = Policy.objects.all()

filter(**kwargs)[source]¶

Filter objects by criteria.

Parameters:: **kwargs (Any) – Filter parameters
Returns:: Filtered QuerySet
Return type:: QuerySet[T]

Example

>>> active_policies = Policy.objects.filter(status='active')

exclude(**kwargs)[source]¶

Exclude objects matching criteria.

Parameters:: **kwargs (Any) – Exclusion parameters
Returns:: Filtered QuerySet
Return type:: QuerySet[T]

get(**kwargs)[source]¶

Get a single object.

Parameters:

**kwargs (Any) – Lookup parameters

Returns:

Single model instance

Raises:

DoesNotExist – If not found
MultipleObjectsReturned – If multiple found

Return type:

Example

>>> policy = Policy.objects.get(id=123)

order_by(*fields)[source]¶

Order results by given fields.

Parameters:: *fields (str) – Field names to order by (prefix with ‘-’ for descending)
Returns:: Ordered QuerySet
Return type:: QuerySet[T]

Example

>>> policies = Policy.objects.order_by('-created_at')

first()[source]¶

Get the first object or None.

Returns:: First model instance or None
Return type:: T | None

Example

>>> policy = Policy.objects.first()

last()[source]¶

Get the last object or None.

Returns:: Last model instance or None
Return type:: T | None

Example

>>> policy = Policy.objects.last()

exists()[source]¶

Check if any objects exist.

Returns:: True if results exist, False otherwise
Return type:: bool

Example

>>> has_policies = Policy.objects.filter(status='active').exists()

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:: list[dict[str, Any]]

Example

>>> policies = Policy.objects.values('id', 'policy_number')

values_list(*fields, flat=False)[source]¶

Return list of tuples instead of model instances.

Parameters:

*fields (str) – Field names to include
flat (bool) – If True and one field, return flat list

Returns:

List of tuples (or flat list if flat=True)

Return type:

list[Any]

Example

>>> ids = Policy.objects.values_list('id', flat=True)

count()[source]¶

Count the number of objects.

Returns:: Number of objects
Return type:: int

Example

>>> total = Policy.objects.count()

create(**kwargs)[source]¶

Create a new object in the API.

Parameters:: **kwargs (Any) – Field values
Returns:: Created model instance
Return type:: T

Example

>>> policy = Policy.objects.create(
...     policy_number='POL-001',
...     premium_amount=1500.00
... )

get_or_create(defaults=None, **kwargs)[source]¶

Get an existing object or create a new one.

Parameters:

defaults (dict[str, Any] | None) – Values to use when creating
**kwargs (Any) – Lookup parameters

Returns:

Tuple of (object, created) where created is a boolean

Return type:

tuple[T, bool]

Example

>>> policy, created = Policy.objects.get_or_create(
...     policy_number='POL-001',
...     defaults={'premium_amount': 1500.00}
... )

update_or_create(defaults=None, **kwargs)[source]¶

Update an existing object or create a new one.

Parameters:

defaults (dict[str, Any] | None) – Values to update/create with
**kwargs (Any) – Lookup parameters

Returns:

Tuple of (object, created) where created is a boolean

Return type:

tuple[T, bool]

Example

>>> policy, created = Policy.objects.update_or_create(
...     policy_number='POL-001',
...     defaults={'premium_amount': 2000.00}
... )

bulk_create(objs)[source]¶

Create multiple objects in bulk.

Parameters:: objs (list[dict[str, Any]]) – List of dictionaries with object data
Returns:: List of created model instances
Return type:: list[T]

Example

>>> policies = Policy.objects.bulk_create([
...     {'policy_number': 'POL-001', 'premium_amount': 1500},
...     {'policy_number': 'POL-002', 'premium_amount': 2000},
... ])

Example Usage¶

from django_api_orm import APIModel

class User(APIModel):
    _schema_class = UserSchema
    _endpoint = "/api/v1/users/"

# Access manager
manager = User.objects

# Create instance
user = User.objects.create(name="Alice", email="alice@example.com")

# Get instance
user = User.objects.get(id=1)

# Filter instances
active_users = User.objects.filter(active=True)

# Get or create
user, created = User.objects.get_or_create(
    email="alice@example.com",
    defaults={"name": "Alice"}
)

# Update or create
user, created = User.objects.update_or_create(
    email="alice@example.com",
    defaults={"name": "Alice Updated"}
)

# Bulk create
users = User.objects.bulk_create([
    {"name": "Bob", "email": "bob@example.com"},
    {"name": "Charlie", "email": "charlie@example.com"}
])

Asynchronous Managers¶

class django_api_orm.async_base.AsyncManager(model_class, client)[source]¶

Bases: Generic[T]

Django-like AsyncManager for model querying.

Handles creation and retrieval of model instances using async methods.

Example

>>> class Policy(AsyncAPIModel):
...     objects = AsyncManager()
>>> policies = await Policy.objects.filter(status='active').alen()

__init__(model_class, client)[source]¶

Initialize AsyncManager.

Parameters:

model_class (type[T]) – The model class this manager handles
client (AsyncServiceClient) – Async HTTP client for API requests

get_endpoint()[source]¶

Get the API endpoint for this model.

Returns:: API endpoint path
Return type:: str

all()[source]¶

Get all objects.

Returns:: AsyncQuerySet of all objects
Return type:: AsyncQuerySet[T]

Example

>>> all_policies = Policy.objects.all()
>>> async for policy in all_policies:
...     print(policy)

filter(**kwargs)[source]¶

Filter objects by criteria.

Parameters:: **kwargs (Any) – Filter parameters
Returns:: Filtered AsyncQuerySet
Return type:: AsyncQuerySet[T]

Example

>>> active_policies = Policy.objects.filter(status='active')

exclude(**kwargs)[source]¶

Exclude objects matching criteria.

Parameters:: **kwargs (Any) – Exclusion parameters
Returns:: Filtered AsyncQuerySet
Return type:: AsyncQuerySet[T]

async get(**kwargs)[source]¶

Get a single object.

Parameters:

**kwargs (Any) – Lookup parameters

Returns:

Single model instance

Raises:

DoesNotExist – If not found
MultipleObjectsReturned – If multiple found

Return type:

Example

>>> policy = await Policy.objects.get(id=123)

order_by(*fields)[source]¶

Order results by given fields.

Parameters:: *fields (str) – Field names to order by (prefix with ‘-’ for descending)
Returns:: Ordered AsyncQuerySet
Return type:: AsyncQuerySet[T]

Example

>>> policies = Policy.objects.order_by('-created_at')

async first()[source]¶

Get the first object or None.

Returns:: First model instance or None
Return type:: T | None

async last()[source]¶

Get the last object or None.

Returns:: Last model instance or None
Return type:: T | None

async exists()[source]¶

Check if any objects exist.

Returns:: True if results exist, False otherwise
Return type:: bool

async count()[source]¶

Count the number of objects.

Returns:: Number of objects
Return type:: int

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:: list[dict[str, Any]]

Example

>>> policies = await Policy.objects.values('id', 'policy_number')

async values_list(*fields, flat=False)[source]¶

Return list of tuples instead of model instances.

Parameters:

*fields (str) – Field names to include
flat (bool) – If True and one field, return flat list

Returns:

List of tuples (or flat list if flat=True)

Return type:

list[Any]

Example

>>> ids = await Policy.objects.values_list('id', flat=True)

async create(**kwargs)[source]¶

Create a new object in the API.

Parameters:: **kwargs (Any) – Field values
Returns:: Created model instance
Return type:: T

Example

>>> policy = await Policy.objects.create(
...     policy_number='POL-001',
...     premium_amount=1500.00
... )

async get_or_create(defaults=None, **kwargs)[source]¶

Get an existing object or create a new one.

Parameters:

defaults (dict[str, Any] | None) – Values to use when creating
**kwargs (Any) – Lookup parameters

Returns:

Tuple of (object, created) where created is a boolean

Return type:

tuple[T, bool]

Example

>>> policy, created = await Policy.objects.get_or_create(
...     policy_number='POL-001',
...     defaults={'premium_amount': 1500.00}
... )

async update_or_create(defaults=None, **kwargs)[source]¶

Update an existing object or create a new one.

Parameters:

defaults (dict[str, Any] | None) – Values to update/create with
**kwargs (Any) – Lookup parameters

Returns:

Tuple of (object, created) where created is a boolean

Return type:

tuple[T, bool]

Example

>>> policy, created = await Policy.objects.update_or_create(
...     policy_number='POL-001',
...     defaults={'premium_amount': 2000.00}
... )

async bulk_create(objs)[source]¶

Create multiple objects in bulk.

Parameters:: objs (list[dict[str, Any]]) – List of dictionaries with object data
Returns:: List of created model instances
Return type:: list[T]

Example

>>> policies = await Policy.objects.bulk_create([
...     {'policy_number': 'POL-001', 'premium_amount': 1500},
...     {'policy_number': 'POL-002', 'premium_amount': 2000},
... ])

Example Usage¶

from django_api_orm import AsyncAPIModel

class User(AsyncAPIModel):
    _schema_class = UserSchema
    _endpoint = "/api/v1/users/"

# Access manager
manager = User.objects

# Create instance (await needed)
user = await User.objects.create(name="Alice", email="alice@example.com")

# Get instance (await needed)
user = await User.objects.get(id=1)

# Filter instances (no await needed for QuerySet)
active_users = User.objects.filter(active=True)

# Get or create (await needed)
user, created = await User.objects.get_or_create(
    email="alice@example.com",
    defaults={"name": "Alice"}
)

# Update or create (await needed)
user, created = await User.objects.update_or_create(
    email="alice@example.com",
    defaults={"name": "Alice Updated"}
)

# Bulk create (await needed)
users = await User.objects.bulk_create([
    {"name": "Bob", "email": "bob@example.com"},
    {"name": "Charlie", "email": "charlie@example.com"}
])

Custom Managers¶

You can create custom managers to encapsulate common queries:

from django_api_orm import Manager, APIModel

class UserManager(Manager):
    """Custom manager for User model."""

    def active(self):
        """Get only active users."""
        return self.filter(active=True)

    def admins(self):
        """Get only admin users."""
        return self.filter(role="admin")

class User(APIModel):
    _schema_class = UserSchema
    _endpoint = "/api/v1/users/"
    objects = UserManager()

# Usage
active_users = User.objects.active()
admins = User.objects.admins()

Managers¶

Synchronous Managers¶

Example Usage¶

Asynchronous Managers¶

Example Usage¶

Custom Managers¶

See Also¶