derp.orm.engine – Database Engine

Async query engine (asyncpg wrapper) for Derp ORM.

class derp.orm.engine.Transaction

Bases: _QueryBase

Transaction context manager with query builder support.

Queries created via this transaction’s select, insert, update, and delete methods reuse the transaction’s connection instead of acquiring a new one from the pool.

Example:

async with db.transaction() as txn:
    user = await txn.insert(User).values(name="Alice").returning(User).execute()
    await txn.update(Profile).set(user_id=user.id).execute()

__init__(connection)

Parameters:

connection (Connection)

class derp.orm.engine.DatabaseEngine

Bases: _QueryBase

Main async database engine for Derp ORM.

Example

db = DatabaseEngine(“postgresql://user:pass@localhost:5432/mydb”)

async with db:

users = await db.select(User).where(User.name == “Alice”).execute()

# Or manual lifecycle await db.connect() users = await db.select(User).execute() await db.disconnect()

__init__(dsn, *, min_size=2, max_size=10, statement_cache_size=None)

Initialize Derp engine.

Parameters:

• dsn (str) – PostgreSQL connection string

• min_size (int) – Minimum connection pool size

• max_size (int) – Maximum connection pool size

• statement_cache_size (int | None) – Size of the prepared statement cache per connection. Set to 0 to disable, which is required when connecting through PgBouncer in transaction mode. None uses asyncpg’s default.

async connect()

Establish connection pool.

Return type:

None

async disconnect()

Close connection pool.

Return type:

None

set_cache(store)

Set the KV store for query result caching.

Parameters:

store (KVClient | None)

Return type:

None

set_router(router)

Set the replica router for automatic read routing.

Parameters:

router (ReplicaRouter | None)

Return type:

None

property pool: Pool

Get the connection pool.

table(table_name)

Start a non ORM query from a table name or Table class.

Parameters:

table_name (Table | str)

Return type:

TableRef

async execute(query, params=None)

Execute a raw SQL query.

Parameters:

• query (str) – SQL query string with $N placeholders

• params (list[Any] | None) – Parameter values

Returns:

List of row dicts

Return type:

list[dict[str, Any]]

Example

result = await db.execute(“SELECT * FROM users WHERE id = $1”, [user_id])

async execute_many(query, params_list)

Execute a query with multiple parameter sets.

Parameters:

• query (str) – SQL query string

• params_list (list[list[Any]]) – List of parameter lists

Return type:

None

transaction()

Start a transaction.

Example

async with db.transaction():

await db.insert(User).values(…).execute() await db.update(Post).set(…).execute()

Return type:

AsyncIterator[Transaction]

acquire()

Acquire a connection from the pool.

Example

async with db.acquire() as conn:

await conn.execute(”…”)

Return type:

AsyncIterator[Connection]