Ditching Heavy ORMs: Building a Zero-Overhead Type-Safe Query Pipeline with Raw SQL and Pydantic v2

Ditching Heavy ORMs: Building a Zero-Overhead Type-Safe Query Pipeline with Raw SQL and Pydantic v2


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.


Why Raw SQL + Pydantic v2

Raw SQL gives you complete control over the query plan, indexes, and joins. Pydantic v2 supplies fast runtime validation, excellent IDE support, and automatic conversion from database records into typed Python objects. Together they remove the abstraction tax of an ORM while preserving (and often improving) developer experience.

Pydantic v2 is particularly well suited for this role because of its performance improvements and the model_validate / from_attributes capabilities. Official documentation is available at the Pydantic site.


Ditching Heavy ORMs: Building a Zero-Overhead Type-Safe Query Pipeline with Raw SQL and Pydantic v2




The Core Pipeline

Here is a production-ready implementation using asyncpg:


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))


Handling Complex Queries and Joins


For multi-table queries you can still keep everything type-safe. Define nested or composite models and map the result rows explicitly. This approach forces you to think about the exact data shape you need, which usually leads to more efficient queries than the automatic relationship loading offered by ORMs.


Connection Pooling and Transaction Management

Use a proper connection pool (asyncpg.create_pool) and keep transactions short. Because you control the SQL, it is easy to place critical statements inside explicit transactions when atomicity is required.


Testing and Maintainability

SQL can live in separate .sql files or as carefully named constants. Repository methods become the single place where database interaction occurs, making unit and integration tests straightforward. You can mock the pool or run tests against a real (or containerized) database with confidence that the queries are identical to production.


When This Approach Shines

  • 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

Trade-offs and Honest Limitations

You lose automatic relationship loading, change tracking, and some convenience features of mature ORMs. Schema evolution requires coordinated updates to both the SQL and the Pydantic models. For applications dominated by simple CRUD and heavy relationship traversal, a well-tuned ORM can still be more productive. The raw-SQL approach rewards teams that already think in terms of queries and indexes.


Final Thoughts

Ditching a heavy ORM in favor of raw SQL paired with Pydantic v2 removes an entire abstraction layer and its associated cost. You gain predictable performance, clearer code, and complete visibility into every statement that reaches the database. For many modern services this zero-overhead pipeline is the cleaner, faster, and more maintainable path. The combination of explicit SQL and strict typing delivers both speed and safety without forcing you to accept the complexity of a full ORM.

Further reading on the underlying tools can be found in the asyncpg documentation and the Pydantic v2 documentation.

Post a Comment

Previous Post Next Post