Build a zero-overhead, type-safe data access layer with raw SQL and Pydantic v2. Faster queries, full control, and excellent developer experience.
Heavy ORMs such as SQLAlchemy, Django ORM, or Tortoise introduce convenience at a measurable cost. They generate SQL dynamically, maintain complex identity maps, perform automatic relationship loading, and often hide the exact queries that reach the database. In high-throughput services this overhead becomes visible as higher latency, increased memory consumption, and harder-to-debug performance problems.
A cleaner alternative is to write the SQL yourself and pair it with Pydantic v2 for strict type safety, validation, and serialization. The result is a zero-overhead, fully explicit data-access layer that remains pleasant to work with.
I migrated a production API that previously used SQLAlchemy to this pattern. Average query latency dropped noticeably, memory usage decreased, and the team gained full visibility into every statement executed against PostgreSQL.
import asyncpg
from pydantic import BaseModel, Field, ConfigDict
from typing import List, Optional
from datetime import datetime
class User(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
email: str
full_name: str
is_active: bool
created_at: datetime
class CreateUser(BaseModel):
email: str = Field(..., min_length=5, max_length=255)
full_name: str = Field(..., min_length=1)
is_active: bool = True
class UserRepository:
def __init__(self, pool: asyncpg.Pool):
self.pool = pool
async def get_by_id(self, user_id: int) -> Optional[User]:
query = """
SELECT id, email, full_name, is_active, created_at
FROM users
WHERE id = $1
"""
row = await self.pool.fetchrow(query, user_id)
if row is None:
return None
return User.model_validate(dict(row))
async def list_active(self, limit: int = 50, offset: int = 0) -> List[User]:
query = """
SELECT id, email, full_name, is_active, created_at
FROM users
WHERE is_active = true
ORDER BY created_at DESC
LIMIT $1 OFFSET $2
"""
rows = await self.pool.fetch(query, limit, offset)
return [User.model_validate(dict(row)) for row in rows]
async def create(self, data: CreateUser) -> User:
query = """
INSERT INTO users (email, full_name, is_active)
VALUES ($1, $2, $3)
RETURNING id, email, full_name, is_active, created_at
"""
row = await self.pool.fetchrow(
query, data.email, data.full_name, data.is_active
)
return User.model_validate(dict(row))
async def update_status(self, user_id: int, is_active: bool) -> Optional[User]:
query = """
UPDATE users
SET is_active = $2
WHERE id = $1
RETURNING id, email, full_name, is_active, created_at
"""
row = await self.pool.fetchrow(query, user_id, is_active)
if row is None:
return None
return User.model_validate(dict(row))- High-throughput APIs and microservices
- Services with complex, performance-sensitive queries
- Teams that value explicit SQL and full control over the database
- Projects that want to avoid ORM magic, migration complexity, and N+1 surprises

