A Django ORM-like interface for external REST APIs with Pydantic validation and httpx.
django-api-orm provides a familiar Django ORM interface for interacting with REST APIs, making it easy to work with external services using the same patterns you already know. Built with modern Python features:
- Pydantic 2.0 for robust data validation and serialization
- httpx for modern HTTP client support (both sync and async)
- Full type safety with comprehensive type hints
- Django-like API for intuitive, familiar usage
- 91% test coverage with 152 comprehensive tests
- Sync and Async Support - Use synchronous or asynchronous clients based on your needs
- Type Safety - Full type hints and Pydantic validation throughout
- Nested Schemas - Full support for nested Pydantic models with dot notation access
- Django-like QuerySet API - Familiar filter(), exclude(), get(), first(), last(), etc.
- HTTP/2 Support - Optional HTTP/2 for better performance (async only)
- Connection Pooling - Built-in connection pooling with httpx
- Automatic Retries - Configurable retry logic for failed requests
- Lazy Evaluation - QuerySets only execute when needed
- Chainable Queries - Build complex queries with method chaining
- CRUD Operations - Full create, read, update, delete support
- Concurrent Operations - Easy async/await with asyncio.gather()
- Test Server Included - FastAPI test server with insurance domain for development and testing
# Basic installation
pip install django-api-orm
# With async and HTTP/2 support
pip install django-api-orm[async]
# Development installation
pip install django-api-orm[dev]Using uv (recommended):
uv add django-api-orm
# or with async support
uv add "django-api-orm[async]"from pydantic import BaseModel
from django_api_orm import APIModel, ServiceClient, register_models
# Define your Pydantic schema
class UserSchema(BaseModel):
id: int | None = None # Optional for creation
name: str
email: str
active: bool = True
# Define your API model
class User(APIModel):
_schema_class = UserSchema
_endpoint = "/api/v1/users/"# Create a client and register models
with ServiceClient(
base_url="https://api.example.com",
auth_token="your-token-here"
) as client:
register_models(client, User)
# Query users (Django-like!)
active_users = User.objects.filter(active=True)
for user in active_users:
print(f"{user.name} - {user.email}")
# Get a single user
user = User.objects.get(id=1)
# Create a new user
new_user = User.objects.create(
name="Alice Smith",
email="alice@example.com"
)
# Update a user
user.email = "newemail@example.com"
user.save(update_fields=["email"])
# Delete a user
user.delete()
# Chain queries
recent_active = (User.objects
.filter(active=True)
.order_by("-created_at")
.first())
# Get or create
user, created = User.objects.get_or_create(
email="bob@example.com",
defaults={"name": "Bob Jones"}
)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 # Enable HTTP/2
) as client:
register_async_models(client, User)
# Async iteration
async for user in User.objects.filter(active=True):
print(f"{user.name} - {user.email}")
# Async retrieval
user = await User.objects.get(id=1)
# Async creation
new_user = await User.objects.create(
name="Charlie Brown",
email="charlie@example.com"
)
# Async update
user.email = "updated@example.com"
await user.save()
# Concurrent operations
users_count, posts_count = await asyncio.gather(
User.objects.filter(active=True).count(),
Post.objects.filter(published=True).count()
)
asyncio.run(main())django-api-orm fully supports nested Pydantic models with dot notation access:
from pydantic import BaseModel, EmailStr
from datetime import date
# Define nested schema
class AddressSchema(BaseModel):
street: str
city: str
state: str
zip_code: str
country: str = "USA"
# Use in parent schema
class PolicyHolderSchema(BaseModel):
id: int | None = None
first_name: str
last_name: str
email: EmailStr
date_of_birth: date
address: AddressSchema # Nested schema
active: bool = True
class PolicyHolder(APIModel):
_schema_class = PolicyHolderSchema
_endpoint = "/api/v1/policyholders/"
# Access nested fields with dot notation
with ServiceClient(base_url="https://api.example.com") as client:
register_models(client, PolicyHolder)
# Create with nested data
address = AddressSchema(
street="123 Main St",
city="Chicago",
state="IL",
zip_code="60601"
)
holder = PolicyHolder.objects.create(
first_name="John",
last_name="Doe",
email="john@example.com",
date_of_birth=date(1980, 1, 1),
address=address, # Nested schema
active=True
)
# Access nested fields
print(holder.address.city) # "Chicago"
print(holder.address.state) # "IL"
print(holder.address.zip_code) # "60601"
# Update nested fields
holder.address.street = "456 Oak Ave"
holder.save()See examples/INSURANCE_DOMAIN.md for a complete insurance domain example with nested schemas.
django-api-orm includes a comprehensive FastAPI test server for development and testing. The test server implements a complete insurance domain API with PolicyHolder, Policy, and Claim models.
# Install development dependencies
uv sync --all-extras
# Start the server
uvicorn examples.test_server:app --host 127.0.0.1 --port 8700 --reloadThe server will be available at http://localhost:8700 with interactive API docs at http://localhost:8700/docs.
# Run synchronous test script
uv run python examples/test_with_server.py
# Run asynchronous test script
uv run python examples/test_with_server_async.pyThe test scripts demonstrate all features including:
- Creating, reading, updating, and deleting records
- Filtering and querying with nested schemas
- Relationship-based filtering
- Concurrent async operations
- And much more!
See examples/README.md for complete test server documentation.
# Filtering
User.objects.filter(active=True, role="admin")
User.objects.exclude(status="banned")
# Ordering
User.objects.order_by("-created_at")
User.objects.order_by("name", "-email")
# Slicing
User.objects.all()[10:20] # Offset and limit
User.objects.all()[0] # Get first by index
# Retrieval
User.objects.get(id=123) # Get single (raises DoesNotExist)
User.objects.first() # Get first or None
User.objects.last() # Get last or None
User.objects.exists() # Check if any exist
User.objects.count() # Get count
# Value extraction
User.objects.all().values("id", "name") # List of dicts
User.objects.all().values_list("id", flat=True) # Flat list# Creation
User.objects.create(name="Alice", email="alice@example.com")
User.objects.bulk_create([
{"name": "Bob", "email": "bob@example.com"},
{"name": "Charlie", "email": "charlie@example.com"}
])
# 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"}
)# Create
user = User(name="Alice", email="alice@example.com", _client=client)
user.save()
# Update
user.email = "newemail@example.com"
user.save() # Updates all fields
user.save(update_fields=["email"]) # Partial update
# Refresh
user.refresh_from_api() # Re-fetch from API
# Delete
user.delete()
# Convert to dict
data = user.to_dict()
data = user.to_dict(exclude_unset=True)The library has comprehensive test coverage (91%):
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=src/django_api_orm
# Run specific tests
uv run pytest tests/test_orm_integration.py
uv run pytest tests/test_async_orm_integration.py# Type checking with mypy
uv run mypy src/
# Linting with ruff
uv run ruff check src/ tests/
# Auto-fix linting issues
uv run ruff check --fix src/ tests/
# Format code
uv run ruff format src/ tests/Version: 0.1.0 (Beta)
Test Coverage: 91% (152 tests)
- Complete exception system
- Synchronous HTTP client (ServiceClient)
- Asynchronous HTTP client (AsyncServiceClient)
- QuerySet implementation (sync and async)
- Manager implementation (sync and async)
- APIModel base class (sync and async)
- Model registration system
- Nested Pydantic schema support with dot notation
- Manager delegation methods (values, values_list)
- AsyncQuerySet slicing support
- Comprehensive test suite (152 tests, 91% coverage)
- FastAPI test server with insurance domain examples
- Full type safety with mypy
- CI/CD with GitHub Actions
- Connection pooling and retries
- HTTP/2 support (async)
- Advanced caching layer
- Streaming support for large responses
- Related field support (select_related, prefetch_related)
- Query optimization and request batching
- More example projects
- Performance benchmarks
We chose httpx for several important reasons:
- Native async/sync support - Single codebase for both patterns
- HTTP/2 support - Better performance with multiplexing
- Better connection pooling - Optimized and built-in
- Modern API - Cleaner, more Pythonic interface
- Active development - Better maintained with more features
- Type hints - Better IDE support and type safety
- Better timeout control - More granular timeout configuration
# Clone the repository
git clone https://github.com/mrb101/django-api-orm.git
cd django-api-orm
# Install dependencies
uv sync --all-extras
# Run tests
uv run pytest
# Type checking
uv run mypy src/
# Linting
uv run ruff check src/ tests/src/django_api_orm/
├── __init__.py # Public API exports
├── py.typed # PEP 561 type marker
├── exceptions.py # Custom exceptions
├── client.py # Sync HTTP client (httpx)
├── async_client.py # Async HTTP client (httpx)
├── base.py # APIModel, QuerySet, Manager
├── async_base.py # Async versions
├── utils.py # Helper functions
└── typing.py # Type hints and protocols
tests/
├── test_exceptions.py # Exception tests
├── test_client.py # Sync client tests
├── test_async_client.py # Async client tests
├── test_orm_integration.py # Sync ORM tests
├── test_async_orm_integration.py # Async ORM tests
└── test_utils.py # Utility tests
examples/
├── basic_usage.py # Basic sync example
├── async_usage.py # Basic async example
├── test_server.py # FastAPI test server (insurance domain)
├── test_with_server.py # Comprehensive sync test script
├── test_with_server_async.py # Comprehensive async test script
├── README.md # Test server documentation
├── INSURANCE_DOMAIN.md # Insurance domain guide
└── QUICKSTART.md # Quick start guide
Contributions are welcome! Please feel free to submit a Pull Request.
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Make your changes
- Add tests for your changes
- Ensure tests pass (
uv run pytest) - Ensure type checking passes (
uv run mypy src/) - Ensure linting passes (
uv run ruff check src/ tests/) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
MIT License - see LICENSE file for details.
Created by Bassel J. Hamadeh at TigerLab.
Inspired by Django ORM, Pydantic, and httpx.
- Issues: https://github.com/mrb101/django-api-orm/issues
- Discussions: https://github.com/mrb101/django-api-orm/discussions
- Documentation: https://django-api-orm.readthedocs.io/en/latest/
See CHANGELOG.md for detailed version history.
- Nested Pydantic schema support - Full support for nested models with dot notation access
- FastAPI test server - Comprehensive insurance domain test server for development
- Enhanced Manager methods - Added values() and values_list() delegation methods
- Bug fixes - Fixed nested schema preservation, attribute syncing, and JSON serialization
- Initial release with full sync and async support
- Complete QuerySet, Manager, and APIModel implementations
- 91% test coverage with 152 tests
- Full type safety with mypy
- CI/CD with GitHub Actions