DEV Community

kernz
kernz

Posted on

I built a 57-line asyncpg wrapper because SQLAlchemy was driving me insane

I came from Rust where I used sqlx — you write raw SQL, you get typed structs back. Simple, honest, fast.

Then I had to write Python and reached for SQLAlchemy. Big mistake.

Suddenly I was learning a DSL on top of SQL. Debugging what ORM decided to generate behind my back. Fighting N+1 queries I didn't even know were happening. Writing text() to escape into raw SQL anyway.

So I built EzQL.

What is it?

A minimal wrapper around asyncpg. You write SQL. You get typed Pydantic models back. That's literally it.

The entire core is 57 lines of code.

How it works

Define your model:

from pydantic import BaseModel

class User(BaseModel):
    __table__ = "users"  # Marks this model as an EzQL model

    id: int
    name: str
Enter fullscreen mode Exit fullscreen mode

Connect and query:

from ezql import create_client

client = await create_client(
    user, password, database, host,
    min_connections=2,
    max_connections=10
)

# Fire and forget
await client.execute("INSERT INTO users (name) VALUES ($1)", "Nazar")

# Returns List[User] or []
users = await client.query_as(User, "SELECT id, name FROM users WHERE name = $1", "Nazar")

assert users[0].name == "Nazar"
assert isinstance(users[0], User)
Enter fullscreen mode Exit fullscreen mode

No sessions. No .commit(). No magic lazy loading. You know exactly what hits the database.

Joins

For joins you just define a DTO — a plain Pydantic model without __table__:

class UserWithPosts(BaseModel):
    # No __table__ — this is a DTO, not a table model
    user_name: str
    post_title: str

users_with_posts = await client.query_as(UserWithPosts, """
    SELECT users.name as user_name, posts.title as post_title
    FROM users
    JOIN posts ON posts.user_id = users.id
    WHERE users.id = $1
""", user_id)
Enter fullscreen mode Exit fullscreen mode

Same approach as sqlx in Rust — separate struct for each query shape. No relationship magic, no hidden queries.

Warning: Always select columns explicitly in joins. SELECT * may cause a ValidationError at runtime since column names can collide across tables.

CLI validator

EzQL also ships with a CLI tool that validates your models against the actual DB schema before you deploy:

ezql ./models --dsn postgresql://user:password@localhost:5432/mydb
Enter fullscreen mode Exit fullscreen mode
Found 1 models. Validating against DB...
Validating User  table users
┏━━━━━━━┳━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━┓
 Field  Model type     DB type  Status 
┡━━━━━━━╇━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━┩
 id     <class 'int'> │ integer │ ✓      │
│ name  │ <class 'str'> │ text    │ ✓      │
└───────┴───────────────┴─────────┴────────┘
All models are valid ✓
Enter fullscreen mode Exit fullscreen mode

Catches missing columns, type mismatches, and tables that don't exist — before production blows up.

Why not just use SQLAlchemy?

SQLAlchemy makes sense when:

  • Your team doesn't know SQL
  • You need to support multiple databases simultaneously
  • You're maintaining a legacy codebase that already uses it

Otherwise it's just an extra layer of abstraction you'll eventually punch through with text() anyway.

If you know SQL — just write SQL.

GitHub

GitHub logo kernz / ezql

Why use another bloated ORM when you can interact with your database in the most intuitive and simple way?

EzQL

EzQL - a simple wrapper around asyncpg that makes writing raw SQL in python a little easier

SQLAlchemy is overkill and you need something simpler? Have you tried writing raw SQL in Python but ended up in tears because there's no type safety? Then EzQL is your choice

Why use another bloated ORM when you can interact with your database in the most intuitive and simple way?

Good place to start

PostgreSQL is an open-source relational database beloved by most developers for its reliability, performance, and rich feature set - from advanced indexing and full-text search to JSON support and powerful extensions like PostGIS.

If you're new to PostgreSQL, here are the best places to start:

Example

import asyncio
from pydantic import BaseModel
from ezql import create_client
class User(BaseModel):
    __table__ = "users"
Enter fullscreen mode Exit fullscreen mode

Feedback welcome — especially if you find a case where the type mapping breaks.

Top comments (0)