pyxle-db

The official database plugin for Pyxle. One explicit-SQL API over SQLite, PostgreSQL, and MySQL — no ORM, no query builder, Django-grade ergonomics for people who like writing SQL.

• Write SQL once, in portable qmark style (? placeholders); pyxle-db translates per backend with a literal-aware rewriter.
• Every backend returns the same Row type, raises the same pyxle_db error types, and hands back timezone-aware UTC datetimes — application code never imports sqlite3, asyncpg, or asyncmy.
• Filesystem migrations with SHA-256 checksum tracking, applied atomically.
• First-class Pyxle plugin: one entry in pyxle.config.json and your app has an open database at startup.

Install

pip install pyxle-db               # SQLite — zero extra dependencies
pip install 'pyxle-db[postgres]'   # + asyncpg
pip install 'pyxle-db[mysql]'      # + asyncmy

Quickstart

Same code on every backend — only the connection string changes.

from pyxle_db import connect

# SQLite (a bare path, exactly like 0.1)
db = await connect("./data/app.db", migrations_dir="migrations")

# PostgreSQL
db = await connect("postgresql://app:s3cret@localhost/app")

# MySQL
db = await connect("mysql://app:s3cret@localhost/app")

await db.execute(
    "INSERT INTO users (id, email) VALUES (?, ?)", (user_id, email)
)

row = await db.fetchone("SELECT email, created_at FROM users WHERE id = ?", (user_id,))
row["email"]        # name access
row[0]              # index access
row["created_at"]   # timezone-aware UTC datetime, on every backend

user = await db.get("SELECT * FROM users WHERE id = ?", (user_id,))  # raises NotFoundError if absent

await db.aclose()

Constraint violations raise pyxle_db.IntegrityError; unreachable/timed-out databases raise pyxle_db.OperationalError (the retryable family); everything else is a pyxle_db.DatabaseError subclass. Driver exceptions never leak.

Database URLs

Database(...), connect(...), and the plugin's url setting all accept:

Ports default to 5432/3306. Query-string options (e.g. sslmode) pass through to the driver. Credentials are percent-decoded, so encode special characters (p@ss → p%40ss).

Placeholders

Always write ?. pyxle-db rewrites it to the backend's native style — ? for SQLite, $1/$2 for PostgreSQL, %s for MySQL.

When you need a literal question mark — PostgreSQL's JSON operators — escape it as ??:

await db.fetchall(
    "SELECT id FROM events WHERE payload ?? 'user_id' AND kind = ?",
    ("signup",),
)
# asyncpg receives: SELECT id FROM events WHERE payload ? 'user_id' AND kind = $1

The rewriter is literal-aware: ? inside string literals, quoted identifiers, comments, and dollar-quoted bodies is never touched, so user data can't become SQL structure during translation.

Writing portable schemas

Rules proven against real PostgreSQL and MySQL servers (the live suites in tests/ enforce them):

• VARCHAR(n) for every key or indexed column. MySQL cannot index bare TEXT (error 1170). SQLite and PostgreSQL treat VARCHAR as TEXT, so nothing is lost. Keep TEXT for payloads.
• Datetimes: bind anything, read aware UTC. Columns store UTC wall time. You may bind naive datetimes (assumed UTC) or aware ones (the backend converts to UTC before binding); every read comes back as an aware-UTC datetime on all three engines.
• TIMESTAMP columns are fine on SQLite and PostgreSQL. On MySQL prefer DATETIME(6) via a per-dialect migration override (0002-x.mysql.sql): MySQL's TIMESTAMP is capped at 2038 and rounds to whole seconds. (The backend pins each session to UTC, so even TIMESTAMP columns read back correct instants.)
• MySQL has no CREATE INDEX IF NOT EXISTS. Create indexes in migrations (they run exactly once, checksum-tracked) or probe information_schema.statistics before creating.
• Spell out inserted values instead of relying on column DEFAULTs, which drift subtly between engines.

Transactions

async with db.transaction() as tx:
    await tx.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, src))
    await tx.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, dst))

Commits on exit, rolls back on exception. Concurrent transactions never interleave statements on the same connection.

SQLite only: synchronous scripts and tests can use with db.sync_transaction() as tx: ... and db.close(). Server backends hold async pools, so both raise UnsupportedOperationError there — use async with db.transaction() and await db.aclose().

Migrations

Each migration is a .sql file named <NNNN>-<slug>.sql in your migrations directory:

migrations/
  0001-initial-schema.sql
  0002-add-sessions.sql
  0003-fulltext-index.sql
  0003-fulltext-index.postgresql.sql   # backend-specific override

• The numeric prefix is the canonical order; two files may not share one.
• A dialect-suffixed file (.sqlite.sql, .postgresql.sql, .mysql.sql) replaces the base file on that backend — for the rare migration that genuinely needs backend-specific DDL.
• Each migration runs in its own transaction and is applied exactly once per database, recorded in schema_migrations with the file's SHA-256.
• Applying is concurrency-safe: when several processes race the same pending migration (each worker of a multi-worker server migrates on startup), they serialize on a database lock — SQLite BEGIN IMMEDIATE, PostgreSQL pg_advisory_xact_lock, MySQL GET_LOCK. One process applies it; the rest verify the recorded checksum and continue. On MySQL this needs pool_max >= 2 (the default is 10).
• Checksum policy: every startup re-hashes applied migrations. An edited file raises MigrationChecksumMismatch; a deleted one raises MigrationError. Once applied anywhere, a migration is immutable — write a follow-up migration instead.

Apply on startup via connect(..., migrations_dir="migrations"), the plugin's migrationsDir setting, or explicitly:

from pathlib import Path
from pyxle_db import Migrator

await Migrator(db, Path("migrations")).apply_all()

Using it as a Pyxle plugin

{
  "plugins": [
    {
      "name": "pyxle-db",
      "settings": {
        "url": "env:DATABASE_URL",
        "migrationsDir": "migrations"
      }
    }
  ]
}

Settings (all optional):

Registered services: db.database (the open Database), db.url (password-redacted connection string for logging), db.auto_transactions, and — SQLite only — db.path. With orm configured it also registers db.orm.engine and db.orm.session_factory.

In loaders and actions, skip the service registry boilerplate:

from pyxle_db import get_database

@server
async def load(request):
    db = get_database()
    return {"posts": await db.fetchall("SELECT * FROM posts ORDER BY id DESC")}

Request-scoped access & auto-transactions

With the plugin installed, every loader and action gets a database handle on request.state.db — no import, no service lookup:

@server
async def load(request):
    rows = await request.state.db.fetchall("SELECT * FROM posts ORDER BY id DESC")
    return {"posts": rows}

@action
async def create_post(request):
    body = await request.json()
    await request.state.db.execute("INSERT INTO posts (title) VALUES (?)", (body["title"],))
    return {"ok": True}

The handle is lazy: a request that never queries opens no connection.

Auto-transactions. On an unsafe method (POST/PUT/PATCH/DELETE) the request's writes run inside one transaction that commits when the action succeeds and rolls back when it fails — where "fails" means the action raised ActionError (or any exception), which Pyxle turns into a non-2xx response. You don't write commit()/rollback(); a failed action never leaves a partial write behind. GET/HEAD requests run read-only (no held write transaction).

Opt out when you need to manage transactions yourself:

from pyxle_db import no_auto_transaction

@action
@no_auto_transaction
async def transfer(request):
    async with request.state.db.transaction() as tx:
        await tx.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amt, src))
        await tx.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amt, dst))

Set "autoTransactions": false to disable it app-wide.

The ORM path (SQLAlchemy)

Prefer an ORM? Install the extra and turn it on — both paths are first-class, and the base install stays SQLAlchemy-free:

pip install 'pyxle-db[sqlalchemy]'

{ "name": "pyxle-db", "settings": { "url": "env:DATABASE_URL",
  "orm": { "metadata": "app.models:Base" } } }

# app/models.py
from sqlalchemy.orm import Mapped, mapped_column
from pyxle_db.orm import Base

class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(primary_key=True)
    body: Mapped[str]

from sqlalchemy import select

@server
async def load(request):
    notes = (await request.state.session.scalars(select(Note))).all()
    return {"notes": [n.body for n in notes]}

@action
async def add_note(request):
    request.state.session.add(Note(body=(await request.json())["body"]))
    return {"ok": True}  # committed automatically on success

request.state.session is a request-scoped AsyncSession under the same auto-transaction rules as request.state.db. SQLAlchemy errors surface as the same pyxle_db error types (IntegrityError, OperationalError, …) on both paths. Pool tuning lives under orm.pool (poolSize, maxOverflow, poolTimeout, poolRecycle, poolPrePing — pre-ping defaults on).

The pyxle-db CLI

pyxle-db migrate              # apply pending checksum migrations
pyxle-db migrate --dry-run    # show what would be applied
pyxle-db status               # applied vs pending

# ORM migrations via Alembic (needs the [sqlalchemy] extra):
pyxle-db alembic-init                         # scaffold alembic.ini + alembic/
pyxle-db revision -m "add notes" --autogenerate
pyxle-db upgrade head
pyxle-db downgrade -1
pyxle-db current      # / history

The CLI reads the same pyxle.config.json + .env your app does, so it always targets the identical database. Pick one migration tool per app: the checksum migrator (migrate) for the explicit-SQL path, Alembic for the ORM path.

Upgrading from 0.1

Breaking changes in 0.2

• Transaction methods are now coroutines. async with db.transaction() as tx: then await tx.execute(...) / await tx.fetchone(...) — 0.1's sync calls inside the async block no longer work.
• db.close() and db.sync_transaction() are now SQLite-only and raise UnsupportedOperationError on PostgreSQL/MySQL. Prefer await db.aclose() everywhere — it works on every backend.
• Mapping (named) parameters are SQLite-only; portable SQL uses positional ? parameters.

Everything else is additive: bare SQLite paths, the migration format, and the plugin settings from 0.1 keep working unchanged.

Roadmap

Short list, subject to change: streaming fetches for large result sets, read-replica routing, and slow-query/pool-stat observability hooks. Issues and PRs welcome.

License

MIT.