snekql

snekql is a Python async-first query builder and query runtime for SQL. It gives applications explicit SQL-shaped operations, typed model declarations, runtime validation, startup schema checks, and transaction- based execution without becoming an ORM.

Install

uv add snekql                 # Query Builder and backend namespaces only
uv add 'snekql[aiosqlite]'    # SQLite Query Runtime
uv add 'snekql[aiomysql]'     # MariaDB Query Runtime

snekql requires Python 3.14 or newer. Database drivers are optional backend extras. The base snekql install is enough for importing the Query Builder and backend namespaces, but runtime initialization requires the matching backend extra.

Bundled docs and examples

Installed packages include copyable docs and examples:

snekql --agent-docs
snekql --llms
snekql --examples
snekql --example basic
snekql example typed_queries
python -m snekql --agent-docs

Quick start

from __future__ import annotations

from datetime import datetime
from pathlib import Path

from snekql import sqlite
from snekql.sqlite import (
    Database,
    Fetched,
    Pending,
    insert,
    select,
)


class User[S = Pending](sqlite.Model[S, "User[Fetched]"]):
    id: sqlite.GenCol[int] = sqlite.Integer(
        primary_key=True,
        auto_increment=True,
        default=sqlite.PENDING_GENERATION,
    )
    email: sqlite.Col[str] = sqlite.Text(unique=True)
    status: sqlite.Col[str] = sqlite.Text(default="active")
    created_at: sqlite.GenCol[datetime] = sqlite.Text(default=sqlite.CurrentTimestamp)


MIGRATIONS = {
    "0001_create_user": (
        'CREATE TABLE "user" ('
        '"id" INTEGER PRIMARY KEY AUTOINCREMENT, '
        '"email" TEXT NOT NULL, '
        '"status" TEXT NOT NULL, '
        '"created_at" TEXT NOT NULL DEFAULT '
        "(strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"
        ") STRICT"
    ),
    "0002_user_email_unique": (
        'CREATE UNIQUE INDEX "ux_user_email" ON "user" ("email")'
    ),
}


async def main() -> None:
    # The migration SQL is generated once during development, reviewed, and
    # committed as a literal. Runtime model changes cannot alter its checksum.
    async with await Database.initialize(
        sqlite.Config(
            database=Path("app.db"),
            pool_size=5,
            acquire_timeout=30.0,
        ),
    ) as db:
        await db.migrate(MIGRATIONS)
        await db.verify_migrations(MIGRATIONS)
        await db.verify([User], policy="strict")
        async with db.transaction(timeout=5.0) as tx:
            await tx.execute(insert(User(email="alice@example.com")))
            # fetch_one is exactly-one: it raises NoResultError if the row is
            # absent, so the result is never None.
            user = await tx.fetch_one(
                select(User).where(User.email.eq("alice@example.com")),
            )
            print(user.email)

Model declaration

Table models are declared through a backend namespace such as sqlite or mariadb. Application-created instances are Pending; database reads return Fetched instances.

from datetime import datetime

from snekql import sqlite
from snekql.sqlite import Fetched, Pending


class AuditLog[S = Pending](sqlite.Model[S, "AuditLog[Fetched]"]):
    __tablename__ = "audit_log"

    id: sqlite.GenCol[int] = sqlite.Integer(
        primary_key=True,
        auto_increment=True,
        default=sqlite.PENDING_GENERATION,
    )
    message: sqlite.Col[str] = sqlite.Text()
    created_at: sqlite.GenCol[datetime] = sqlite.Text(
        default=sqlite.CurrentTimestamp,
    )

Rules to remember:

• Col[T] is a normal persisted column.
• GenCol[T] is server/generated; pending values may be PENDING_GENERATION, fetched values are T.
• If __tablename__ is omitted, class names become snake_case table names.
• Models are immutable after construction/materialization.
• Fetched models are produced by database reads only.
• Instance methods that assume a state should annotate self, e.g. self: User[Pending] or self: User[Fetched].

State-specific instance methods

Model classes are generic in state. If a method uses pending-only or fetched-only assumptions, write that state on self:

class User[S = Pending](sqlite.Model[S, "User[Fetched]"]):
    id: sqlite.GenCol[int] = sqlite.Integer(
        primary_key=True, default=sqlite.PENDING_GENERATION
    )
    email: sqlite.Col[str] = sqlite.Text()

    def insert_payload(self: User[Pending]) -> dict[str, str]:
        return {"email": self.email}

    def cache_key(self: User[Fetched]) -> str:
        return f"user:{self.id}"

A bare User means User[Pending]; spell User[Fetched] for methods that require a materialized row.

Ruff/Pyflakes unused-import note

Fetched appears in model declarations as part of a string forward reference, for example sqlite.Model[S, "User[Fetched]"]. Type checkers resolve that name, but Ruff's Pyflakes F401 check does not count names inside string literals as import usage. If a project imports Fetched only for those model self-types, allow that import in Ruff:

[tool.ruff.lint.pyflakes]
allowed-unused-imports = [
  "snekql.sqlite.Fetched",
  "snekql.mariadb.Fetched",
]

Column types and logical types

A column is two coordinates (see ADR 0005):

• The column type is the constructor. It names a storage primitive of the backend and decides where the value is physically stored — nothing else.
• The logical type is the field annotation (Col[T]). It is the single source of truth for the column's Python value and all validation, which is delegated to Pydantic.

Read a declaration as a sentence — created_at: Col[datetime] = Text() is "a datetime, stored as text." The codec that bridges the two is derived from the pair; you never name it.

SQLite exposes exactly its four storage classes as column types:

• sqlite.Integer — INTEGER storage. A Col[bool] stores as 0/1.
• sqlite.Real — REAL storage.
• sqlite.Text — TEXT storage. Holds Col[str], Col[UtcDatetime] (canonical UTC millisecond text), Col[datetime] (raw ISO text), Col[uuid.UUID] (string form), or a Col[pydantic.Json[T]] payload.
• sqlite.Blob — BLOB storage for Col[bytes].

JSON uses Pydantic's marker, not a snekql type: annotate Col[pydantic.Json[T]] = Text(). Serialization and validation both run through T, so any type Pydantic can validate (datetime, Pydantic models, list[Model], ...) round-trips, not just dict/list/primitives.

MariaDB additionally exposes its native types as column types — mariadb.Json, mariadb.Boolean, mariadb.DateTime, mariadb.Uuid (native UUID), and mariadb.Decimal(precision, scale) (native DECIMAL(p,s)). To store a UUID as raw bytes instead, pair Col[uuid.UUID] with Blob().

There is no declaration-time storage/logical compatibility check: any pairing is allowed and an impossible one fails at encode/decode via a Pydantic error. Timezone policy is the logical type's job — over a primitive storage class (SQLite Text(), or Integer() with an epoch type) a naive datetime round-trips naive. Use UtcDatetime for database timestamp columns: it rejects naive values, normalizes aware values to UTC milliseconds, and serializes SQLite text so =, ORDER BY, and ranges compare by instant. Bare SQLite Col[datetime] = Text() and Col[AwareDatetime] = Text() columns emit a suppressible LexicalDatetimeWarning because their raw ISO text compares lexically. The one exception is MariaDB's native DateTime, which stores offset-less UTC text: it has no way to record a naive value's zone, so encoding a naive datetime there is rejected with a ModelValidationError rather than silently assuming the writer's local zone. Attach a timezone (or annotate Col[UtcDatetime]) for those columns.

Decimal storage has the same two-coordinate rule:

• Use Col[CanonicalDecimal] = Text() when you need portable exact decimal identity/equality over text storage. It normalizes Decimal("1.50") to Decimal("1.5"), Decimal("1E+2") to Decimal("100"), and negative zero to zero, then stores minimal plain text. Equality, IN, and unique indexes are safe; lexical ordering and range predicates are not.
• On SQLite, store integer minor units (Col[int] = Integer(), e.g. cents) when the database must order, range-filter, or aggregate decimal quantities.
• On MariaDB, use Col[decimal.Decimal] = mariadb.Decimal(precision, scale) for native numeric equality, ordering, range predicates, and aggregation. Values that would overflow or require rounding for the declared (precision, scale) are rejected before they reach the driver.

Bare Col[decimal.Decimal] = Text() emits LexicalDecimalWarning on both backends because Pydantic's default decimal text can represent the same value in multiple ways and still sorts lexically, not numerically. Suppress it only when a custom Annotated[..., Canonical] or Annotated[..., OrderPreserving] logical type owns the wire-form invariant.

Because the logical type is whatever Pydantic can validate, the UUID-version aliases work as drop-in logical types and add version validation for free: Col[pydantic.UUID4] = Text(), or Col[pydantic.UUID7] = mariadb.Uuid(). Pydantic ships UUID1/UUID3/UUID4/UUID5/UUID6/UUID7/UUID8; all of them store the same as a plain Col[uuid.UUID] and round-trip through both Text and MariaDB's native Uuid. Pair the annotation with a matching factory (Col[pydantic.UUID7] = mariadb.Uuid(default_factory=uuid.uuid7)) — nothing forces the factory and the annotation to agree, so a mismatched version fails construct-time validation.

All column constructors accept unique=True for column-level unique indexes. SQLite allows multiple NULL values in a unique index, so use a non-optional annotation such as Col[str] when uniqueness should also require a value. Primary-key columns reject unique=True because it is redundant.

For a plain non-unique single-column index, pass index=True instead — sugar for an Index(col) entry in __indexes__ (named ix_<table>_<col>). It is rejected on primary-key columns and alongside unique=True, since both are already indexed.

A server default is declared by passing a marker as the column's default: sqlite.CurrentTimestamp and mariadb.CurrentTimestamp are the only v1 server defaults. The marker means the database computes the value, so the field is valid only on GenCol columns, is omittable at construction (it is PENDING_GENERATION until the database fills it), and accepts an explicit value when you pass one.

To refresh a column to the server clock on update, pass the same marker to an update assignment: update(Doc).set(Doc.edited_at.to(CurrentTimestamp)). It renders the backend's current-timestamp SQL inline (no bound parameter) and is identical on SQLite and MariaDB. SQLite has no native ON UPDATE, so this keeps the refresh explicit at the call site -- include it in each update that should bump the timestamp.

Indexes

Use the backend namespace Index(...) in __indexes__ for table-level indexes:

from snekql import sqlite
from snekql.sqlite import Fetched, Pending


class User[S = Pending](sqlite.Model[S, "User[Fetched]"]):
    email: sqlite.Col[str] = sqlite.Text(unique=True)
    status: sqlite.Col[str] = sqlite.Text()
    tenant_id: sqlite.Col[int] = sqlite.Integer()

    __indexes__ = [
        sqlite.Index(status),
        sqlite.Index(tenant_id, email, unique=True),
        sqlite.Index(tenant_id, name="ix_user_tenant_custom"),
    ]

Index declarations accept column descriptors only. Names are inferred as ix_<table>_<columns> or ux_<table>_<columns> unless name= is supplied. A column-level index=True collides with an equivalent Index(col) here and is rejected as a duplicate.

Queries

Queries are immutable. Chaining returns new query objects.

from snekql.sqlite import delete, insert, select, update

select(User).all()
select(User.email).where(User.status.eq("active"))
select(User.email, User.status).where(User.email.like("%@example.com"))

insert(User(email="alice@example.com"))

update(User).set(User.status.to("disabled")).where(
    User.email.eq("alice@example.com"),
)

delete(User).where(User.email.eq("retired@example.com"))
delete(User).all()  # explicit full-table delete

Inserts can handle a primary-key or unique-index conflict atomically. DoUpdate accepts one or more assignments. .to_inserted() takes the value from the row whose insert conflicted, while .to(...) assigns a literal or CurrentTimestamp:

from snekql.sqlite import DoNothing, DoUpdate, insert

insert(User(email=email, name=name, status=status)).on_conflict(
    User.email,
    action=DoUpdate(
        User.name.to_inserted(),
        User.status.to("active"),
    ),
)

insert(User(email=email, name=name, status=status)).on_conflict(
    User.email,
    action=DoNothing,
)

SQLite compiles these actions as ON CONFLICT (...) DO UPDATE or DO NOTHING. MariaDB compiles them as ON DUPLICATE KEY UPDATE. MariaDB checks every primary key and unique index, so its SQL cannot limit detection to the columns passed to on_conflict; those columns select the no-op assignment used for DoNothing. On SQLite, the target columns must match a primary key or unique index.

DoNothing cannot be combined with .returning(...) because SQLite may return no row. DoUpdate supports .returning(...) for single and bulk inserts.

Filtering is explicit: select, update, and delete must choose exactly one of .where(...) or .all() before execution. Predicates use methods such as .eq(...), .ne(...), .is_null(), .in_(...), .like(...), .gt(...)/.gte(...)/.lt(...)/.lte(...), and .between(low, high); Python comparison operators are not part of the v1 API.

Combine predicates with | (OR), & (AND), and ~ (NOT); use parentheses to group. Repeated .where(...) calls AND together, so & is mainly useful inside an OR. Python's and/or/not keywords are rejected — a predicate raises if used as a boolean.

# WHERE status = 'active' OR status = 'trialing'
select(User).where(User.status.eq("active") | User.status.eq("trialing"))

# WHERE tenant_id = 1 AND (status = 'active' OR email LIKE '%@vip.com')
select(User).where(
    User.tenant_id.eq(1) & (User.status.eq("active") | User.email.like("%@vip.com")),
)

# WHERE NOT (status = 'disabled')
select(User).where(~User.status.eq("disabled"))

A select can be nested inside another query as a subquery:

from snekql.sqlite import exists, not_exists, scalar, select

# IN / NOT IN against a single-column subquery
select(User).where(
    User.id.in_subquery(select(Order.user_id).where(Order.amount.gt(100))),
)

# EXISTS / NOT EXISTS, correlated to the outer row via a column comparison
select(User).where(
    exists(select(Order.id).where(Order.user_id.eq_col(User.id))),
)
select(User).where(
    not_exists(select(Order.id).where(Order.user_id.eq_col(User.id))),
)

# A scalar subquery used in a projection (or as a comparison operand)
select(
    User.id,
    scalar(select(Order.amount.sum()).where(Order.user_id.eq_col(User.id))),
).all()

in_subquery/not_in_subquery and scalar(...) require a single-column select; exists/not_exists accept any select. The *_col comparisons (.eq_col, .ne_col, .gt_col, .gte_col, .lt_col, .lte_col) compare a column against another column or a scalar subquery, which is how a correlated subquery references the outer query. A reference to a table in neither the subquery nor an enclosing query is rejected when the query compiles.

Inspecting the generated SQL

Any query object renders its own SQL through repr() and str(), resolving the dialect from its model's backend — no Database or transaction needed. Because queries are immutable, the object you hold after composing (query = query.where(...)) already carries the full state, so inspecting it shows the final SQL.

query = select(User).where(User.status.eq("active"))
query = query.where(User.email.like("%@example.com"))

repr(query)
# <SelectModelQuery: SELECT ... FROM "user"
#  WHERE ("status" = ?) AND ("email" LIKE ?) | params=('active', '%@example.com')>

print(query)  # str(): the parameterized form plus an inlined-literals form
# -- parameterized (executes):
# SELECT ... WHERE ("status" = ?) AND ("email" LIKE ?)
# -- params: ('active', '%@example.com')
#
# -- inlined literals (approximate, not executed):
# SELECT ... WHERE ("status" = 'active') AND ("email" LIKE '%@example.com')

The parameterized form is what executes. The inlined form substitutes the encoded parameters as SQL literals for pasting into a database console; it is approximate and must not be executed. A query that has not yet chosen .where(...)/.all() renders as <SelectModelQuery incomplete: ...> rather than raising, so it is always safe to repr a query in a debugger.

Runtime

Database.initialize(...) is the only public construction path and is connect-only: it opens connectivity and a connection pool and does no schema work. Select the backend with its namespace config. The legacy SQLite keyword form remains supported for compatibility, but new code should use sqlite.Config.

from pathlib import Path

from snekql import sqlite
from snekql.sqlite import Database


db = await Database.initialize(
    sqlite.Config(database=Path("app.db"), pool_size=5),
)
# MIGRATIONS is the committed literal chain from the quick start above.
await db.migrate(MIGRATIONS)
await db.verify_migrations(MIGRATIONS)
await db.verify([User])

memory_db = await Database.initialize(
    sqlite.Config(database=":memory:"),
)

A Database is an async context manager, so the runtime is closed for you on block exit (including when the body raises) — async with await Database.initialize(...) as db:. Call await db.close() directly only when you manage the lifecycle by hand.

snekql logs through the standard library logging module. Every snekql logger is a child of the snekql logger (snekql.runtime, snekql.sqlite.runtime, …), so an application controls all snekql output from one place:

import logging

# Route snekql logs wherever the app sends its own logs.
logging.basicConfig(level=logging.INFO)

# Or silence snekql while keeping the rest of the app verbose.
logging.getLogger("snekql").setLevel(logging.WARNING)

snekql attaches a NullHandler to the snekql logger, so it emits nothing until the application configures logging. To capture snekql's structured fields, point a JSON/structured formatter (e.g. structlog's ProcessorFormatter) at the handler that receives snekql records — snekql itself stays pure stdlib.

MariaDB models should use the MariaDB namespace so backend-specific columns and runtime checks agree:

from snekql import mariadb
from snekql.mariadb import Database, Fetched, Pending, insert, select


class Account[S = Pending](mariadb.Model[S, "Account[Fetched]"]):
    id: mariadb.GenCol[int] = mariadb.Integer(
        primary_key=True,
        auto_increment=True,
        default=mariadb.PENDING_GENERATION,
    )
    email: mariadb.Col[str] = mariadb.Text(unique=True)


config = mariadb.Config(
    database="app",
    host="127.0.0.1",
    port=3306,
    user="snekql",
    password="secret",
)

async with await Database.initialize(config) as db:
    migrations = {
        "0001_create_account": (
            "CREATE TABLE `account` ("
            "`id` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, "
            "`email` VARCHAR(255) CHARACTER SET utf8mb4 "
            "COLLATE utf8mb4_bin NOT NULL"
            ") ENGINE=InnoDB"
        ),
        "0002_account_email_unique": (
            "CREATE UNIQUE INDEX `ux_account_email` ON `account` (`email`)"
        ),
    }
    await db.migrate(migrations)
    await db.verify_migrations(migrations)
    await db.verify([Account])
    async with db.transaction() as tx:
        await tx.execute(insert(Account(email="alice@example.com")))
        account = await tx.fetch_one(
            select(Account).where(Account.email.eq("alice@example.com")),
        )

Use transactions for all work:

async with db.transaction() as tx:
    rows = await tx.fetch_all(select(User).all())
    first_email = await tx.fetch_one(
        select(User.email).all().order_by(User.id.asc()).limit(1)
    )
    await tx.execute(update(User).set(User.status.to("inactive")).all())

Runtime methods:

• fetch_all(select(...)) returns all result rows. It is for bounded result sets: the whole result is loaded into memory and every row is validated synchronously on the event loop. The materialization loop yields a cooperative checkpoint periodically so a large read does not monopolize the loop, but it still holds the connection for its full duration. For large result sets stream with fetch_chunks instead, which fetches incrementally and keeps per-batch materialization small.

• fetch_chunks(select(...), size=N) streams rows in batches of up to N from a server-side cursor, so an arbitrarily large result never has to fit in memory. It returns a ChunkStream — an async context manager and async iterator — that must be consumed inside async with so the cursor is closed and the connection released deterministically:

  async with tx.fetch_chunks(select(User).all(), size=500) as stream:
      async for batch in stream:  # batch: list[User[Fetched]]
          for user in batch:
              ...
  

  The stream holds the transaction's single connection for its whole lifetime: no other query may run on the transaction until the stream is closed, and it must be opened and consumed within one task. On MariaDB this uses an unbuffered SSCursor; a default cursor would buffer the full result client-side and defeat streaming.

• fetch_one(select(...)) returns the single matching row (exactly-one contract); it raises NoResultError for no row and MultipleResultsError for more than one. A None from a single-value fetch_one means SQL NULL.

• fetch_one_or_none(select(...)) returns the row or None for the zero-or-one case (model, tuple, and join selects), still raising on more than one row.

• execute(insert(...)) returns None, including conflict-handled inserts without .returning(...); execute(update/delete) returns the affected-row count. SQLite counts matched rows; MariaDB counts only rows an UPDATE actually changed.

• close() is async and idempotent after a successful close.

Migrations and verification

Initialization does no schema work. A live Database applies the complete migration chain, verifies its recorded head, then checks the schema against the models:

db = await Database.initialize(database=Path("app.db"))
# MIGRATIONS is the complete committed chain shown in the quick start.
result = await db.migrate(MIGRATIONS)
await db.verify_migrations(MIGRATIONS)
await db.verify([User], policy="strict")

• db.migrate(migrations) accepts the complete ordered dict[str, str] chain. It verifies each recorded position and exact-body SHA-256 before applying the pending suffix, then returns an immutable MigrationResult. Migrations are the sole schema-creation authority. Run scaffold(...) during development, review its output, and commit the SQL as literals rather than recomputing old bodies from current model metadata.
• db.verify_migrations(migrations) performs a read-only exact-head check. It neither applies pending SQL nor upgrades legacy history.
• db.verify(models, *, policy=...) is a partial, structural check that reports Schema Drift between the models and the live schema. policy="strict" raises SchemaVerificationError; policy="warn" logs and continues. It compares columns, indexes, foreign keys, and storage options by name and semantics, and cannot see default values, CHECK constraints, triggers, or data.

A deploy step runs initialize -> migrate -> verify_migrations -> verify; app replicas run initialize -> verify_migrations -> verify. See docs/migrations.md and docs/schema-drift.md.

Error model

Every intentional package-originated exception is a SnekqlError subclass. Use SnekqlError to catch all snekql failures, or catch narrower subclasses:

• ModelDeclarationError, ModelValidationError, FrozenModelError
• QueryConstructionError, QueryCompilationError
• DatabaseClosedError, PoolTimeoutError, TransactionClosedError, ExecutionError
• SchemaVerificationError
• MigrationDeclarationError, MigrationHistoryError, MigrationError, MigrationLockError

ExecutionError preserves sql and params for debugging. Structured query logs may also include SQL and params exactly as supplied to the database driver; snekql does not redact secrets.

Further reading

• Adoption and release confidence
• Why snekql is not an ORM
• Typing guide
• Schema startup and drift
• Temporary MariaDB Test Server
• Error handling guide
• MariaDB integration PRD

Runnable examples live in examples/:

uv run python -m examples.basic_app
uv run ty check examples/typed_queries.py

Local validation uses uv run snektest. MariaDB integration tests start a Temporary MariaDB Test Server through snekql.testing.mariadb, so mariadbd, mariadb-install-db, and mariadb must be available on the test machine.

Public API

The backend namespaces are the public import surface. Pick snekql.sqlite or snekql.mariadb and import the whole surface from it -- the dialect-neutral verbs and builders as well as that backend's Model and column constructors. There is no flat snekql.<symbol> surface; the package root only exposes the sqlite and mariadb namespace handles. This keeps SQLite-only and MariaDB-only symbols from colliding and stops auto-imports from landing on the wrong backend.

The supported import surface is snekql.sqlite, snekql.mariadb, and snekql.testing.mariadb, each curated in its own __all__. Underscored modules (snekql._*) and backend submodules (snekql.sqlite.config, snekql.sqlite.verbs, …) are implementation detail and not supported import paths — their public symbols are re-exported through the namespace top level. Use Select[Row] and Write[Result] to annotate queries without depending on their state-specific implementation classes. Use ColumnRef[Owner, T] for a read-only column parameter. Queries are built only through the select/insert/update/delete factory verbs. The catchable error contract is the SnekqlError hierarchy re-exported from each namespace. See docs/typing.md for the full contract.

Agent navigation map:

• snekql/model.py: model metaclass, table metadata, pending/fetched materialization.
• snekql/storage.py: column descriptors, SQLite storage metadata, value codecs.
• snekql/expressions.py: predicates, ordering, update assignments.
• snekql/query.py: query builders and SQL compilation.
• snekql/runtime.py: Database, Transaction, execution methods.
• snekql/sqlite/pool.py: internal async SQLite connection pool.
• snekql/sqlite/schema.py: scaffold DDL generation and schema verification (dialect-blind pipeline in snekql/_schema_*.py).
• snekql/errors.py: public exception hierarchy.
• tests/test_public_typing.py: type-checker prototypes for the public API.
• CONTEXT.md: project language and terminology.