`derp.orm` – ORM¶

Derp ORM - A strongly-typed async Python ORM for PostgreSQL.

Example usage:

from derp.orm import Table, Serial, Text, Varchar, Timestamp, Field

class User(Table, table="users"):
    id: Serial = Field(primary=True)
    name: Text = Field()
    email: Varchar[255] = Field(unique=True)
    created_at: Timestamp = Field(default="now()")

async with DatabaseEngine("postgresql://...") as db:
    users = await db.select(User).where(User.name == "Alice").execute()

class derp.orm.DatabaseConfig[source]

Bases: _StrictModel

Database configuration.

db_url: str

replica_url: str | None

schema_path: str

migrations_dir: str

introspect_schemas: Sequence[str]

introspect_exclude_tables: Sequence[str]

ignore_rls: bool

pool_min_size: int

pool_max_size: int

statement_cache_size: int

replica_pool_min_size: int | None

replica_pool_max_size: int | None

replica_statement_cache_size: int | None

replica_max_lag_bytes: int

replica_write_fence_seconds: float

replica_lag_check_interval_seconds: float

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.orm.DatabaseEngine[source]

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)[source]

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()[source]

Establish connection pool.

Return type:: None

async disconnect()[source]

Close connection pool.

Return type:: None

set_cache(store)[source]

Set the KV store for query result caching.

Parameters:: store (KVClient | None)
Return type:: None

set_router(router)[source]

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)[source]

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)[source]

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)[source]

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()[source]

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()[source]

Acquire a connection from the pool.

Example

async with db.acquire() as conn:: await conn.execute(”…”)

Return type:: AsyncIterator[Connection]

class derp.orm.Table[source]

Bases: object

Base class for all Derp table definitions.

Example:

class User(Table, table="users"):
    id: Serial = Field(primary=True)
    name: Text = Field()
    email: Varchar[255] = Field(unique=True)

# Query building — direct class access:
db.select(User).where(User.name == "Alice")

classmethod indexes()[source]

Override to define indexes for this table.

Return type:: Sequence[Index]

__init__(**kwargs)[source]

Parameters:: kwargs (Any)
Return type:: None

classmethod get_table_name()[source]

Get the SQL table name.

Return type:: str

classmethod get_columns()[source]

Get all column definitions.

Return type:: dict[str, Column[Any]]

classmethod get_primary_key()[source]

Get the primary key column if any.

Return type:: tuple[str, Column[Any]] | None

to_dict()[source]

Serialize instance to a dict.

Return type:: dict[str, Any]

to_json()[source]

Serialize instance to a JSON string.

Return type:: str

classmethod from_dict(data)[source]

Construct an instance from a dict (ignores unknown keys).

Parameters:: data (dict[str, Any])
Return type:: Self

classmethod from_json(data)[source]

Construct an instance from a JSON string.

Parameters:: data (str | bytes)
Return type:: Self

classmethod to_ddl()[source]

Generate CREATE TABLE DDL statement.

Return type:: str

class derp.orm.Column[source]

Bases: Expression, Generic

Base descriptor for all table columns.

Extends Expression so columns can be used directly in query building. Implements the descriptor protocol for typed class/instance access.

Subclasses set _sql_type as a class variable. Parameterized types (e.g., Varchar[255]) override sql_type() to include parameters.

__init__(spec)[source]

Parameters:: spec (FieldSpec)
Return type:: None

property primary_key: bool

property unique: bool

property nullable: bool

property default: Any

property has_default: bool

property generated: str | None

property foreign_key: str | Column[Any] | None

property on_delete: FK | None

property on_update: FK | None

foreign_key_sql()[source]

Generate the REFERENCES clause, or None if no FK.

Return type:: str | None

sql_type()[source]

Return type:: str

is_auto_increment()[source]

Return type:: bool

to_sql(params)[source]

Generate SQL string with parameterized values.

Parameters:: params (list[Any])
Return type:: str

count()[source]

Return type:: Any

sum()[source]

Return type:: Any

avg()[source]

Return type:: Any

min()[source]

Return type:: Any

max()[source]

Return type:: Any

case(mapping, *, else_=None)[source]

Parameters:

mapping (dict[Any, Any])
else_ (Any | None)

Return type:

Any

derp.orm.Field(*, primary=False, unique=False, default=<dataclasses._MISSING_TYPE object>, generated=None, foreign_key=None, on_delete=None, on_update=None)[source]

Declare column constraints.

Foreign keys:

Field(foreign_key=User.id, on_delete="cascade")
Field(foreign_key="users.id")

Generated columns:

Field(generated="price * quantity")

Parameters:

primary (bool)
unique (bool)
default (Any)
generated (str | None)
foreign_key (str | Column[Any] | None)
on_delete (FK | Literal['cascade', 'set null', 'set default', 'restrict'] | None)
on_update (FK | Literal['cascade', 'set null', 'set default', 'restrict'] | None)

Return type:

Any

class derp.orm.FieldSpec[source]

Bases: object

Column constraints returned by Field().

This is a placeholder that Table.__init_subclass__ replaces with a real Column descriptor after resolving the type annotation.

__init__(*, primary=False, unique=False, default=<dataclasses._MISSING_TYPE object>, generated=None, foreign_key=None, on_delete=None, on_update=None)[source]

Parameters:

primary (bool)
unique (bool)
default (Any)
generated (str | None)
foreign_key (str | Column[Any] | None)
on_delete (FK | None)
on_update (FK | None)

Return type:

None

primary

unique

default

generated

foreign_key

on_delete

on_update

class derp.orm.Fn[source]

Bases: object

Predefined SQL functions for use as column defaults.

static gen_random_uuid()[source]

Return type:: str

static now()[source]

Return type:: str

static current_timestamp()[source]

Return type:: str

static to_tsvector(config, *columns)[source]

Build a to_tsvector(config, col1 || ' ' || col2) expression.

Parameters:

config (str)
columns (str)

Return type:

str

class derp.orm.FK[source]

Bases: StrEnum

Actions for foreign key ON DELETE / ON UPDATE clauses.

CASCADE = 'CASCADE'

SET_NULL = 'SET NULL'

SET_DEFAULT = 'SET DEFAULT'

RESTRICT = 'RESTRICT'

__new__(value)

class derp.orm.Serial[source]

Bases: Column[int]

Auto-incrementing 4-byte integer (SERIAL).

class derp.orm.BigSerial[source]

Bases: Column[int]

Auto-incrementing 8-byte integer (BIGSERIAL).

class derp.orm.SmallInt[source]

Bases: Column[int]

2-byte signed integer (SMALLINT).

class derp.orm.Integer[source]

Bases: Column[int]

4-byte signed integer (INTEGER).

class derp.orm.BigInt[source]

Bases: Column[int]

8-byte signed integer (BIGINT).

class derp.orm.Varchar[source]

Bases: Column[str], Generic

Variable-length string with limit (VARCHAR).

Accepts both Varchar[255] and Varchar[Literal[255]].

sql_type()[source]

Return type:: str

class derp.orm.Char[source]

Bases: Column[str]

Fixed-length string (CHAR).

Accepts both Char[10] and Char[Literal[10]].

sql_type()[source]

Return type:: str

class derp.orm.Text[source]

Bases: Column[str]

Variable unlimited length string (TEXT).

class derp.orm.Boolean[source]

Bases: Column[bool]

Boolean type (BOOLEAN).

class derp.orm.Timestamp[source]

Bases: Column[datetime]

Timestamp without timezone (TIMESTAMP).

class derp.orm.TimestampTZ[source]

Bases: Column[datetime]

Timestamp with timezone (TIMESTAMP WITH TIME ZONE).

class derp.orm.Date[source]

Bases: Column[date]

Date type (DATE).

class derp.orm.Time[source]

Bases: Column[time]

Time without timezone (TIME).

class derp.orm.TimeTZ[source]

Bases: Column[time]

Time with timezone (TIME WITH TIME ZONE).

class derp.orm.Interval[source]

Bases: Column[timedelta]

Time interval (INTERVAL).

class derp.orm.UUID[source]

Bases: Column[UUID]

UUID type.

class derp.orm.Numeric[source]

Bases: Column[Decimal]

Exact numeric with precision and scale (NUMERIC).

sql_type()[source]

Return type:: str

class derp.orm.Real[source]

Bases: Column[float]

4-byte floating point (REAL).

class derp.orm.DoublePrecision[source]

Bases: Column[float]

8-byte floating point (DOUBLE PRECISION).

class derp.orm.JSON[source]

Bases: Column[Any]

JSON type.

class derp.orm.JSONB[source]

Bases: Column[Any]

Binary JSON type (JSONB).

class derp.orm.Vector[source]

Bases: Column[list[float]]

Vector type for embeddings (pgvector).

Distance methods for use in ORDER BY or WHERE clauses:

db.select(Doc).order_by(Doc.embedding.cosine_distance(vec)).limit(10)

sql_type()[source]

Return type:: str

cosine_distance(other)[source]

Cosine distance (<=>). Sort ASC for most similar.

Parameters:: other (Any)
Return type:: Any

l2_distance(other)[source]

L2 / Euclidean distance (<->). Sort ASC for most similar.

Parameters:: other (Any)
Return type:: Any

inner_product(other)[source]

Negative inner product (<#>). Sort ASC for most similar.

Parameters:: other (Any)
Return type:: Any

class derp.orm.Nullable[source]

Bases: Column[Any], Generic

Nullable column wrapper.

Use as a type annotation to indicate a column that allows NULL:

age: Nullable[Integer] = Field()

At the class level User.age is a Column (supports query operators). At the instance level user.age is int | None.

Nullable[Varchar[255]] also works for parameterized types.

class derp.orm.Enum[source]

Bases: Column, Generic

PostgreSQL enum column.

Usage:

class Status(StrEnum):
    ACTIVE = "active"
    INACTIVE = "inactive"

status: Enum[Status] = Field(default="active")

sql_type()[source]

Return type:: str

class derp.orm.JoinType[source]

Bases: StrEnum

SQL JOIN types.

INNER = 'INNER'

LEFT = 'LEFT'

RIGHT = 'RIGHT'

FULL = 'FULL OUTER'

CROSS = 'CROSS'

__new__(value)

class derp.orm.LockMode[source]

Bases: StrEnum

SQL row-level locking modes.

UPDATE = 'FOR UPDATE'

NO_KEY_UPDATE = 'FOR NO KEY UPDATE'

SHARE = 'FOR SHARE'

KEY_SHARE = 'FOR KEY SHARE'

__new__(value)

class derp.orm.SortOrder[source]

Bases: StrEnum

Column sort order within an index.

ASC = 'ASC'

DESC = 'DESC'

__new__(value)

class derp.orm.LogicalOperator[source]

Bases: StrEnum

SQL logical operators.

AND = 'AND'

OR = 'OR'

__new__(value)

class derp.orm.ComparisonOperator[source]

Bases: StrEnum

SQL comparison operators.

EQ = '='

NE = '<>'

GT = '>'

GTE = '>='

LT = '<'

LTE = '<='

__new__(value)

class derp.orm.Expression[source]

Bases: ABC

Base class for SQL expressions.

abstractmethod to_sql(params)[source]

Generate SQL string with parameterized values.

Parameters:: params (list[Any])
Return type:: str

in_(values)[source]

Parameters:: values (Any)
Return type:: Any

not_in(values)[source]

Parameters:: values (Any)
Return type:: Any

like(pattern)[source]

Parameters:: pattern (str)
Return type:: Any

ilike(pattern)[source]

Parameters:: pattern (str)
Return type:: Any

is_null()[source]

Return type:: Any

is_not_null()[source]

Return type:: Any

between(low, high)[source]

Parameters:

low (Any)
high (Any)

Return type:

Any

matches(query, *, language='english', style='websearch', stored=False)[source]

Full-text search match using @@.

stored: set to True when the column is a pre-computed tsvector (skips the to_tsvector() wrapper). style: "websearch" (default), "plain", or "phrase".

Parameters:

query (str)
language (str)
style (Literal['websearch', 'plain', 'phrase'])
stored (bool)

Return type:

Any

ts_rank(query, *, language='english', style='websearch', stored=False)[source]

Full-text search rank for ORDER BY.

stored: set to True when the column is a pre-computed tsvector (skips the to_tsvector() wrapper).

Parameters:

query (str)
language (str)
style (Literal['websearch', 'plain', 'phrase'])
stored (bool)

Return type:

Any

ts_headline(query, *, language='english', style='websearch', max_words=None, min_words=None, max_fragments=None, start_sel=None, stop_sel=None, fragment_delimiter=None, highlight_all=None, short_word=None)[source]

Highlighted search snippet for display in results.

Parameters:

query (str)
language (str)
style (Literal['websearch', 'plain', 'phrase'])
max_words (int | None)
min_words (int | None)
max_fragments (int | None)
start_sel (str | None)
stop_sel (str | None)
fragment_delimiter (str | None)
highlight_all (bool | None)
short_word (int | None)

Return type:

Any

derp.orm.sql(template, *values)[source]

Create a raw SQL expression fragment.

Use {} as placeholder for parameterized values:

sql("NOW()")
sql("age > {}", 18)
sql("CONCAT({}, name)", "Dr. ")

Parameters:

template (str)
values (Any)

Return type:

RawSQL

class derp.orm.Index[source]

Bases: object

Full PostgreSQL index definition.

Accepts flexible column input:

Index("email")
Index("created_at", "name")
Index(IndexColumn("email", order=SortOrder.DESC))

Index-level options:

Index("embedding", method=IndexMethod.HNSW, opclass="vector_cosine_ops")
Index("status", where=MyTable.is_active == True)
Index("id", include=("name",))
Index(expression="lower(email)", unique=True)

__init__(*columns, method=IndexMethod.BTREE, unique=False, where=None, include=(), nulls_distinct=True, with_params=None, tablespace=None, concurrently=False, name=None, opclass=None, order=None, nulls=None, collation=None, expression=None)[source]

Parameters:

columns (str | IndexColumn | Any)
method (IndexMethod | str)
unique (bool)
where (Any | None)
include (tuple[str, ...] | list[str])
nulls_distinct (bool)
with_params (dict[str, str] | None)
tablespace (str | None)
concurrently (bool)
name (str | None)
opclass (str | None)
order (SortOrder | str | None)
nulls (NullsPosition | str | None)
collation (str | None)
expression (str | None)

Return type:

None

method

unique

where

include

nulls_distinct

with_params

tablespace

concurrently

name

opclass

property columns: tuple[IndexColumn, ...]

property column_names: list[str]: Plain column names for snapshot compat.

auto_name(table_name)[source]

Generate a conventional index name.

Parameters:: table_name (str)
Return type:: str

to_ddl(table_name)[source]

Generate the full CREATE INDEX statement.

Parameters:: table_name (str)
Return type:: str

class derp.orm.IndexColumn[source]

Bases: object

Per-column configuration within an index.

Either name or expression must be provided:

IndexColumn("email")
IndexColumn("email", order=SortOrder.DESC, nulls=NullsPosition.LAST)
IndexColumn(expression="lower(email)")
IndexColumn("embedding", opclass="vector_cosine_ops")

name: str | Any | None = None

expression: str | None = None

opclass: str | None = None

order: SortOrder | None = None

nulls: NullsPosition | None = None

collation: str | None = None

to_ddl()[source]

Generate the DDL fragment for this column.

Return type:: str

__init__(name=None, expression=None, opclass=None, order=None, nulls=None, collation=None)

Parameters:

name (str | Any | None)
expression (str | None)
opclass (str | None)
order (SortOrder | None)
nulls (NullsPosition | None)
collation (str | None)

Return type:

None

class derp.orm.IndexMethod[source]

Bases: StrEnum

PostgreSQL index access methods.

BTREE = 'btree'

HASH = 'hash'

GIN = 'gin'

GIST = 'gist'

SPGIST = 'spgist'

BRIN = 'brin'

HNSW = 'hnsw'

IVFFLAT = 'ivfflat'

__new__(value)

class derp.orm.NullsPosition[source]

Bases: StrEnum

NULLS FIRST / LAST within an index column.

FIRST = 'FIRST'

LAST = 'LAST'

__new__(value)

derp.orm – ORM¶

`derp.orm` – ORM¶