Examples

This section provides practical examples of using django-api-orm in various scenarios.

Basic CRUD Operations

Complete example of Create, Read, Update, Delete operations:

from pydantic import BaseModel, EmailStr
from django_api_orm import APIModel, ServiceClient, register_models

# Define schema
class UserSchema(BaseModel):
    id: int | None = None
    name: str
    email: EmailStr
    active: bool = True
    role: str = "user"

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

# Use the model
with ServiceClient(
    base_url="https://api.example.com",
    auth_token="your-token-here"
) as client:
    register_models(client, User)

    # CREATE
    user = User.objects.create(
        name="Alice Smith",
        email="alice@example.com",
        role="admin"
    )
    print(f"Created user {user.id}: {user.name}")

    # READ
    user = User.objects.get(id=user.id)
    print(f"Retrieved user: {user.name}")

    # List all users
    all_users = list(User.objects.all())
    print(f"Total users: {len(all_users)}")

    # Filter users
    active_users = list(User.objects.filter(active=True))
    print(f"Active users: {len(active_users)}")

    # UPDATE
    user.email = "alice.updated@example.com"
    user.save(update_fields=["email"])
    print(f"Updated user email: {user.email}")

    # DELETE
    user.delete()
    print("User deleted")

Async CRUD Operations

Async version with concurrent operations:

import asyncio
from django_api_orm import AsyncAPIModel, AsyncServiceClient, register_async_models

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

async def main():
    async with AsyncServiceClient(
        base_url="https://api.example.com",
        auth_token="your-token-here",
        http2=True
    ) as client:
        register_async_models(client, User)

        # CREATE - concurrent creation
        users = await asyncio.gather(
            User.objects.create(name="Alice", email="alice@example.com"),
            User.objects.create(name="Bob", email="bob@example.com"),
            User.objects.create(name="Charlie", email="charlie@example.com")
        )
        print(f"Created {len(users)} users concurrently")

        # READ - concurrent retrieval
        retrieved_users = await asyncio.gather(
            User.objects.get(id=users[0].id),
            User.objects.get(id=users[1].id),
            User.objects.get(id=users[2].id)
        )
        print(f"Retrieved {len(retrieved_users)} users concurrently")

        # UPDATE - concurrent updates
        for user in users:
            user.active = True
        await asyncio.gather(*[user.save() for user in users])
        print("Updated all users concurrently")

        # DELETE - concurrent deletion
        await asyncio.gather(*[user.delete() for user in users])
        print("Deleted all users concurrently")

asyncio.run(main())

Pagination

Paginating through large result sets:

def paginate_users(page_size=20):
    """Paginate through all users."""
    page = 1
    while True:
        offset = (page - 1) * page_size
        users = list(User.objects.all()[offset:offset + page_size])

        if not users:
            break

        print(f"\nPage {page}:")
        for user in users:
            print(f"  {user.id}: {user.name}")

        page += 1

# Usage
with ServiceClient(base_url="https://api.example.com") as client:
    register_models(client, User)
    paginate_users(page_size=10)

Search and Filter

Advanced filtering and search:

with ServiceClient(base_url="https://api.example.com") as client:
    register_models(client, User)

    # Find active admin users
    active_admins = User.objects.filter(active=True, role="admin")

    # Exclude banned users
    valid_users = User.objects.exclude(status="banned")

    # Combine filters
    users = (User.objects
             .filter(active=True)
             .exclude(role="guest")
             .order_by("-created_at"))

    # Get count
    admin_count = User.objects.filter(role="admin").count()
    print(f"Total admins: {admin_count}")

    # Check existence
    has_admins = User.objects.filter(role="admin").exists()
    if has_admins:
        print("Admin users exist")

    # Get first/last
    newest_user = User.objects.order_by("-created_at").first()
    oldest_user = User.objects.order_by("created_at").first()

Batch Operations

Processing records in batches:

# Synchronous batch processing
def process_users_in_batches(batch_size=100):
    """Process users in batches."""
    offset = 0

    while True:
        batch = list(User.objects.all()[offset:offset + batch_size])

        if not batch:
            break

        # Process batch
        for user in batch:
            # Do something with user
            print(f"Processing {user.name}")

        offset += batch_size

# Async batch processing with concurrency
async def process_users_async(batch_size=100, concurrent=10):
    """Process users in batches with concurrency."""
    offset = 0

    while True:
        batch = []
        async for user in User.objects.all()[offset:offset + batch_size]:
            batch.append(user)

        if not batch:
            break

        # Process batch concurrently
        await asyncio.gather(*[process_user(user) for user in batch])

        offset += batch_size

async def process_user(user):
    """Process a single user."""
    # Do something async with user
    await asyncio.sleep(0.1)

Error Handling

Comprehensive error handling:

import logging
from django_api_orm.exceptions import (
    DoesNotExist,
    MultipleObjectsReturned,
    ValidationError,
    UnauthorizedError,
    ForbiddenError,
    NotFoundError,
    TooManyRequestsError,
    APIServerError,
    APITimeoutError,
    APIConnectionError,
)

logger = logging.getLogger(__name__)

def safe_get_user(user_id):
    """Safely get a user with comprehensive error handling."""
    try:
        return User.objects.get(id=user_id)

    except DoesNotExist:
        logger.warning(f"User {user_id} not found")
        return None

    except UnauthorizedError:
        logger.error("Authentication failed - please log in")
        raise

    except ForbiddenError:
        logger.error(f"Access denied to user {user_id}")
        raise

    except TooManyRequestsError as e:
        logger.warning(f"Rate limited. Retry after {e.retry_after}s")
        raise

    except APIServerError as e:
        logger.error(f"Server error: {e.status_code}")
        return None

    except APITimeoutError:
        logger.error(f"Request timed out for user {user_id}")
        return None

    except APIConnectionError as e:
        logger.error(f"Connection failed: {e.message}")
        return None

Custom Managers

Creating custom managers for common queries:

from django_api_orm import Manager, APIModel

class UserManager(Manager):
    """Custom manager with convenience methods."""

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

    def inactive(self):
        """Get inactive users."""
        return self.filter(active=False)

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

    def users(self):
        """Get regular users."""
        return self.filter(role="user")

    def active_admins(self):
        """Get active admin users."""
        return self.active().admins()

    def create_admin(self, **kwargs):
        """Create an admin user."""
        kwargs["role"] = "admin"
        return self.create(**kwargs)

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

# Usage
with ServiceClient(base_url="https://api.example.com") as client:
    register_models(client, User)

    # Use custom manager methods
    active_users = User.objects.active()
    admins = User.objects.admins()
    active_admins = User.objects.active_admins()

    # Create admin
    admin = User.objects.create_admin(
        name="Admin User",
        email="admin@example.com"
    )

Multiple Related Models

Working with multiple related models:

from pydantic import BaseModel

# Define schemas
class UserSchema(BaseModel):
    id: int | None = None
    name: str
    email: str

class PostSchema(BaseModel):
    id: int | None = None
    user_id: int
    title: str
    content: str
    published: bool = False

class CommentSchema(BaseModel):
    id: int | None = None
    post_id: int
    user_id: int
    text: str

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

class Post(APIModel):
    _schema_class = PostSchema
    _endpoint = "/api/v1/posts/"

class Comment(APIModel):
    _schema_class = CommentSchema
    _endpoint = "/api/v1/comments/"

# Usage
with ServiceClient(base_url="https://api.example.com") as client:
    register_models(client, User, Post, Comment)

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

    # Create posts for user
    post1 = Post.objects.create(
        user_id=user.id,
        title="First Post",
        content="Hello World",
        published=True
    )

    post2 = Post.objects.create(
        user_id=user.id,
        title="Second Post",
        content="More content",
        published=False
    )

    # Create comments
    comment = Comment.objects.create(
        post_id=post1.id,
        user_id=user.id,
        text="Great post!"
    )

    # Query related data
    user_posts = Post.objects.filter(user_id=user.id)
    published_posts = Post.objects.filter(user_id=user.id, published=True)
    post_comments = Comment.objects.filter(post_id=post1.id)

Testing with Mock APIs

Testing your code with mock APIs using respx:

import respx
import httpx
from django_api_orm import ServiceClient, APIModel, register_models

@respx.mock
def test_user_creation():
    """Test creating a user."""
    # Mock the API response
    respx.post("https://api.example.com/api/v1/users/").mock(
        return_value=httpx.Response(
            201,
            json={"id": 1, "name": "Alice", "email": "alice@example.com"}
        )
    )

    with ServiceClient(base_url="https://api.example.com") as client:
        register_models(client, User)

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

        assert user.id == 1
        assert user.name == "Alice"
        assert user.email == "alice@example.com"

FastAPI Integration

Integrating with FastAPI:

from fastapi import FastAPI, Depends, HTTPException
from django_api_orm import AsyncServiceClient, AsyncAPIModel, register_async_models

app = FastAPI()

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

async def get_client():
    """Dependency to provide async client."""
    async with AsyncServiceClient(
        base_url="https://api.example.com",
        auth_token="your-token-here"
    ) as client:
        register_async_models(client, User)
        yield client

@app.get("/users/{user_id}")
async def get_user(user_id: int, client = Depends(get_client)):
    """Get a user by ID."""
    try:
        user = await User.objects.get(id=user_id)
        return user.to_dict()
    except User.DoesNotExist:
        raise HTTPException(status_code=404, detail="User not found")

@app.get("/users/")
async def list_users(active: bool = True, client = Depends(get_client)):
    """List users."""
    users = []
    async for user in User.objects.filter(active=active):
        users.append(user.to_dict())
    return users

@app.post("/users/")
async def create_user(
    name: str,
    email: str,
    client = Depends(get_client)
):
    """Create a user."""
    user = await User.objects.create(name=name, email=email)
    return user.to_dict()

See Also

• Quick Start for basic usage

• Models for model documentation

• Async Support for async patterns