pgstorm 0.1.8

A lightweight PostgreSQL query builder and mini-ORM for Python

pgstorm

A lightweight PostgreSQL query builder and mini-ORM for Python. Compose type-safe queries that compile to parameterized SQL, then execute them using a pluggable engine (sync or async).

Features

• Type-safe models — Define models with type annotations; __table__ or __tablename__ for table names
• pgstorm.schema — BaseModel, scalar column types, and the same relation / annotation helpers as pgstorm.types (Annotated, ForeignKey, ON_DELETE_*, FK_FIELD, Self, …), in one import
• Rich QuerySet API — filter, exclude, order_by, limit, offset, join, aggregate, annotate, alias
• Q objects — Combine conditions with | (OR), & (AND), ~ (NOT)
• Subqueries — Subquery and OuterRef for correlated subqueries
• F expressions — Reference annotations/aliases in filters and order_by
• SQL functions — Concat, Coalesce, Upper, Lower, Now, DateTrunc, Func_, and more
• Aggregates — Min, Max, Count, Sum, Avg
• Writes included — create, bulk_create, update, delete (sync or await with async engines)
• Engine abstraction — Sync (psycopg2, psycopg3) and async (psycopg3_async, asyncpg) interfaces
• Transactions — with pgstorm.transaction(): or async with pgstorm.transaction():
• Observers & hooks — Django-style pre_* / post_* and connection/transaction hooks via pgstorm.observers
• Schema support — using_schema() and per-join rhs_schema

Requirements

• Python 3.10+
• A PostgreSQL database
• A driver (installed automatically via extras): psycopg3 (default), psycopg2, or asyncpg

Installation

pip install pgstorm

This installs pgstorm with psycopg3 (the default driver). To use a different driver:

# psycopg2: choose normal (requires libpq) or binary (pre-built)
pip install pgstorm[psycopg2]        # psycopg2 (sync, normal build)
pip install pgstorm[psycopg2-binary] # psycopg2-binary (sync, pre-built)

# psycopg3: choose normal or binary (default uses binary)
pip install pgstorm[psycopg3]        # psycopg3 (normal build)
pip install pgstorm[psycopg3-binary] # psycopg3 binary (pre-built)

pip install pgstorm[asyncpg]         # asyncpg (async)
pip install pgstorm[all]             # all drivers (binary variants)

From source: pip install -e . or pip install -e ".[asyncpg]"

Quick Start

from pgstorm import BaseModel, types, create_engine

class User(BaseModel):
    __table__ = "users"
    id: types.Integer[types.IS_PRIMARY_KEY_FIELD]
    age: types.Integer
    email: types.String

class UserProfile(BaseModel):
    __table__ = "user_profile"
    user: types.ForeignKey[User, types.ON_DELETE_CASCADE]

# Create engine (sets global context for querysets)
engine = create_engine("postgresql://user:pass@localhost/dbname", interface="psycopg3")

# Build and compile a query
qs = UserProfile.objects.filter(
    UserProfile.user.email.like("[e-mail]")
).join(User, UserProfile.user.id == User.id)

compiled = qs.compiled()
print(compiled.sql.as_string(None))
print(compiled.params)

# Execute and iterate (uses engine from context)
for profile in qs:
    print(profile.user.email)

Async example (asyncpg)

QuerySet.fetch() and other methods return an awaitable when the configured engine is async.

import asyncio
from pgstorm import create_engine, Subquery
from example.model import User, AuditLog

db_credentials = {
    "host": "localhost",
    "port": 5432,
    "user": "postgres",
    "password": "admin",
    "dbname": "testdb",
}

async def main():
    create_engine(db_credentials, interface="asyncpg")

    log = await AuditLog.objects.create(
        user=Subquery(
            User.objects.using_schema("tenant1")
                .filter(User.email == "[e-mail]")
                .columns("id")
        ),
        action="INSERT",
        target_table="user",
        target_id=2,
    )
    print("log id:", log.id)

    rows = await User.objects.using_schema("tenant1").filter(User.email.like("[e-mail]")).fetch()
    for user in rows:
        print(user.email)

    print("count:", await User.objects.count())

asyncio.run(main())

Observers & hooks

pgstorm includes an observer system for registering callbacks around queries, connections, cursors, and transactions via pgstorm.observers. Observers can be global (for all tables) or table-specific, and can be sync or async (async callbacks are awaited automatically when you use an async engine).

from pgstorm.observers import (
    ObserverContext,
    table_observers,
    on_fetch,
    on_query_before_execute,
    on_query_after_execute,
    POST_CREATE,
)
from example.model import User


@on_fetch
def log_fetch(ctx: ObserverContext) -> None:
    # Called for any SELECT
    print(f"[fetch] table={ctx.table}, model={ctx.model}")


@table_observers(action=POST_CREATE, table=User)
def user_created(ctx: ObserverContext) -> None:
    # Called only for User inserts
    instance = ctx.extra.get("instance")
    print(f"[post_create] User created: {instance}")


@on_query_before_execute
def before_any_query(ctx: ObserverContext) -> None:
    action = ctx.extra.get("query_action", "?")
    print(f"[before] action={action}, table={ctx.table}")


@on_query_after_execute
def after_any_query(ctx: ObserverContext) -> None:
    rows = ctx.result if isinstance(ctx.result, list) else []
    print(f"[after] rows={len(rows)}")

See example/observers_example.py and the Observers & hooks section in the docs for a fuller walkthrough and list of supported actions.

Models

Define models by subclassing BaseModel and annotating attributes with types:

from pgstorm import BaseModel, types

class Product(BaseModel):
    __table__ = "products"
    id: types.Integer[types.IS_PRIMARY_KEY_FIELD]
    name: types.String
    price: types.Integer

Use __table__ or __tablename__ to set the table name; otherwise the class name (lowercased) is used.

Types

• Scalars (convenience API): types.Integer, types.String, types.BigSerial, types.Jsonb, types.Inet, types.Varchar(20), types.TimestampTZ(default=...)
• Scalars + model base (full catalog): from pgstorm import schema gives schema.BaseModel and every PostgreSQL column class (schema.UUID, schema.Text, schema.Numeric, schema.Vector, geometric types, schema.JsonPythonType, schema.Column, schema.ScalarField, etc.). Same types as pgstorm.columns, but only canonical names (no *Descriptor / *Column aliases).
• Relations & annotation helpers: Either pgstorm.types or pgstorm.schema — same symbols (schema.ForeignKey, schema.Annotated, schema.ON_DELETE_CASCADE, schema.FK_FIELD, schema.Self, schema.String, schema.Varchar(20), …). schema also exposes ref types FKFieldRef, FKColumnRef, ReverseNameRef for advanced use.
• Relations (concise): ForeignKey[User], OneToOne, ManyToMany
• Relation metadata: ON_DELETE_CASCADE, FK_FIELD("email"), FK_COLUMN("user_email"), ReverseName("profiles")

user: schema.ForeignKey[User, schema.ON_DELETE_CASCADE, schema.FK_FIELD("email")]

Self-referential relations: reply_to: schema.ForeignKey[schema.Self] (or types.Self / types.ForeignKey — same objects).

pgstorm.schema (models + column types)

Use one import for table definitions: schema.BaseModel, schema.IS_PRIMARY_KEY_FIELD, all scalar column classes, and relation / annotation helpers (schema.ForeignKey, schema.Annotated, schema.ON_DELETE_*, … — aligned with pgstorm.types).

from pgstorm import schema

class Appointment(schema.BaseModel):
    __tablename__ = "appointments"
    id: schema.UUID[schema.IS_PRIMARY_KEY_FIELD]
    note: schema.Text

Use bracket metadata Field[IS_PRIMARY_KEY_FIELD] on any scalar column class (e.g. schema.Text[schema.IS_PRIMARY_KEY_FIELD]), or primary_key=True in the constructor when using call-style annotations. schema.Numeric is the decimal type (maps to NUMERIC in PostgreSQL).

Engine & Execution

Create an engine with create_engine(). By default it sets the engine in a context variable so querysets use it automatically.

from pgstorm import create_engine

# Sync (default)
engine = create_engine("postgresql://user:pass@localhost/db", interface="psycopg3")

# Async
engine = create_engine("postgresql://...", interface="psycopg3_async")
# or
engine = create_engine("postgresql://...", interface="asyncpg")

Interfaces: psycopg2, psycopg3, psycopg3_sync, psycopg3_async, asyncpg

Fetching results

# Sync: iterate or index
users = list(User.objects.filter(User.age > 18))
user = User.objects.filter(User.id == 1)[0]

# Async: use await fetch()
users = await User.objects.filter(User.age > 18).fetch()

Transactions

import pgstorm

# Sync
with pgstorm.transaction():
    # queries run in transaction
    pass

# Async
async with pgstorm.transaction():
    await User.objects.all().fetch()

QuerySet API

Filtering

# Simple comparisons (==, !=, <, <=, >, >=)
User.objects.filter(User.age >= 18)
User.objects.filter(User.email == "[e-mail]")

# LIKE / ILIKE
User.objects.filter(User.email.like("[e-mail]"))
User.objects.filter(User.name.ilike("%john%"))

# IN
User.objects.filter(User.id.in_([1, 2, 3]))
User.objects.filter(User.id.in_(Subquery(Order.objects.columns("user_id"))))

# Exclude
User.objects.filter(User.age > 18).exclude(User.deleted)

Q objects (AND / OR / NOT)

from pgstorm import Q, and_, or_, not_

User.objects.filter(Q(User.age > 18) | Q(User.age < 5))
User.objects.filter(and_(Q(User.active), Q(User.verified)))
User.objects.filter(~Q(User.deleted))

Joins

UserProfile.objects.join(
    User,
    UserProfile.user.email == User.email,
    join_type="LEFT"
)

Schemas

User.objects.using_schema("tenant_1").filter(...)
UserProfile.objects.join(User, ..., rhs_schema="tenant_2")

Aggregates

from pgstorm import Min, Max, Count, Sum, Avg

# Positional: alias = col_name_function_name (e.g. price_min)
Product.objects.aggregate(Min(Product.price), Max(Product.price))

# Keyword: alias = key
Product.objects.aggregate(total=Sum(Product.price), cnt=Count())

# COUNT(*)
Product.objects.aggregate(row_count=Count())

Annotate & Alias

from pgstorm import Concat, F

# annotate: add computed columns to SELECT; results include them
User.objects.annotate(full_name=Concat(User.first_name, " ", User.last_name))

# alias: define expressions for filter/order_by without including in SELECT
User.objects.alias(full_name=Concat(User.first_name, " ", User.last_name)).filter(
    F("full_name").ilike("%mohamed%")
)

Subqueries & OuterRef

from pgstorm import Subquery, OuterRef

# Users who have at least one order
User.objects.filter(
    User.id.in_(
        Subquery(
            Order.objects.filter(Order.user_id == OuterRef(User.id)).columns("user_id")
        )
    )
)

Other

• order_by(User.age) — ORDER BY
• limit(10), offset(20) — LIMIT / OFFSET
• distinct() — SELECT DISTINCT
• defer("col"), columns("col1", "col2") — column selection
• as_cte(name) — use queryset as CTE

Compiling to SQL

qs = User.objects.filter(User.age > 18).limit(10)
compiled = qs.compiled()

# For psycopg
sql, params = qs.as_sql()
cursor.execute(sql, params)

SQL Functions

Built-in: Concat, Coalesce, Upper, Lower, Length, Trim, Replace, NullIf, Abs, Round, Floor, Ceil, Now, CurrentDate, CurrentTimestamp, DateTrunc. Use Func_("name", arg1, arg2) for any other PostgreSQL function.

Documentation

See the docs/ folder for detailed documentation:

• Installation & Setup
• Models & Types
• QuerySet API
• Engine & Execution
• Functions & Aggregates
• Subqueries
• Observers & hooks
• API Reference

License

Not specified yet (README previously said MIT, but a LICENSE file is not currently present in this repository).