database - Chirp

data.database

module

Typed async database access.

Supports SQLite (via stdlibsqlite3 + anyio) and PostgreSQL (via in-tree pelt). SQL in, frozen dataclasses out.

Connection URL format::

sqlite:///path/to/db.sqlite # SQLite file sqlite:///:memory: #…

Typed async database access.

Supports SQLite (via stdlibsqlite3 + anyio) and PostgreSQL (via in-tree pelt). SQL in, frozen dataclasses out.

Connection URL format::

sqlite:///path/to/db.sqlite    # SQLite file
sqlite:///:memory:             # In-memory SQLite
postgresql://user:pass@host/db # PostgreSQL

Free-threading safety:

Connection pool usesanyio.Lockfor async-safe initialization
Connections are per-task (ContextVar), never shared between tasks
All public methods are async — no sync I/O on the calling thread

Concurrency model:

SQLite uses a small bounded pool (sized bypool_size) of WAL-mode

connections. Readers acquire any free connection and run concurrently;

write transactions serialize behind ``_sqlite_lock`` (single-writer).

PostgreSQL uses the in-tree pelt pool with transaction-level isolation.

data.database

Name	Type	Default	Description
`type`		—
`qualified_name`		—
`element_type`		—
`description`		—
`source_file`		—
`line_number`		—
`is_autodoc`		—
`autodoc_element`		—
`_autodoc_template`		—
`_autodoc_url_path`		—
`_autodoc_page_type`		—
`title`		—
`doc_content_hash`		—

Symbols on this page

function _postgresql_sql

Translate the facade's portable qmark parameters to PostgreSQL$N.

Question marks inside quoted strings, identifiers, dollar-quoted bodies, and comments are inert. Native$N…

function get_db

Return the app-level database instance.

Available when aDatabase is configured on the App::

app = App(db="sqlite:///app.db") @app.route("/users") async def users(): db =…

function _in_transaction

Check if the current task is inside a managed transaction.

class Database

Typed async database access.

SQL queries return frozen dataclasses. Streaming queries return async iterators. Both modes use the same SQL — the difference is…

function _detect_driver

Detect the database driver from the URL scheme.

_postgresql_sql

function

def _postgresql_sql(sql: str, param_count: int) -> str

Translate the facade's portable qmark parameters to PostgreSQL$N.

Question marks inside quoted strings, identifiers, dollar-quoted bodies, and comments are inert. Native$NSQL remains accepted when it contains no qmark parameters; mixing both styles is rejected as ambiguous.

Parameters

Name	Type	Default	Description
`sql`	`str`	—
`param_count`	`int`	—

get_db

function

def get_db() -> Database

Return the app-level database instance.

Available when aDatabase is configured on the App::

app = App(db="sqlite:///app.db")

@app.route("/users")
async def users():
    db = get_db()
    return await db.fetch(User, "SELECT * FROM users")

RaisesLookupErrorif no database is configured or the app has not started yet.

No parameters.

_in_transaction

function

def _in_transaction() -> bool

Check if the current task is inside a managed transaction.

No parameters.

Database

class

Typed async database access.

SQL queries return frozen dataclasses. Streaming queries return async iterators. Both modes use the same SQL — the difference is whether you want all results at once or incrementally.

Usage::

db = Database("sqlite:///app.db")

@dataclass(frozen=True, slots=True)
class User:
    id: int
    name: str
    email: str

# Fetch all
users = await db.fetch(User, "SELECT * FROM users")

# Fetch one
user = await db.fetch_one(User, "SELECT * FROM users WHERE id = ?", 42)

# Stream (cursor-based)
async for user in db.stream(User, "SELECT * FROM users"):
    process(user)

# Execute (INSERT/UPDATE/DELETE)
await db.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                 "Alice", "alice@example.com")

# Raw scalar
count = await db.fetch_val("SELECT COUNT(*) FROM users")

# Transaction (atomic multi-statement)
async with db.transaction():
    await db.execute("INSERT INTO users ...", name, email)
    await db.execute("INSERT INTO profiles ...", user_id)

_detect_driver

function

def _detect_driver(url: str) -> str

Detect the database driver from the URL scheme.

Parameters

Name	Type	Default	Description
`url`	`str`	—

View source · /home/runner/work/chirp/chirp/site/../src/chirp/data/database.py:1