db — Database Module

Desi's db module provides a unified API for PostgreSQL and MySQL, plus a query builder and ORM.

Zero dependencies — pure C wire protocol clients bundled in libdesi.a.

Import

import db

Connecting

# PostgreSQL
db.connect("postgres", "localhost", 5432, "mydb", "user", "password")

# MySQL
db.connect("mysql", "localhost", 3306, "mydb", "root", "password")

# Check connection
if db.is_connected() == 1:
    print(f"Connected to {db.driver()}")

Raw SQL

# SELECT — returns row count
let rows = db.query("SELECT * FROM users WHERE age > 21")

# Access results
for i in range(rows):
    let name = db.get_field(i, "name")     # by column name
    let age = db.get_value(i, 2)           # by column index
    print(f"{name}, age {age}")

# INSERT/UPDATE/DELETE — returns affected row count
db.execute("INSERT INTO users (name) VALUES ('Alice')")
db.execute("UPDATE users SET age = 31 WHERE name = 'Alice'")
db.execute("DELETE FROM users WHERE id = 5")

# Close when done
db.close()

Query Builder

db.find("users")
db.columns("name")
db.columns("email")
db.where("age", ">", "21")
db.order_by("name", 0)
db.limit(10)
let sql = db.build_sql()
# → SELECT name, email FROM users WHERE age > 21 ORDER BY name LIMIT 10

db.insert_into("users")
db.set_field("name", "Alice")
db.set_field("email", "alice@example.com")
let sql = db.build_sql()
# → INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com')

ORM — Model Definition

db.model("users")
db.auto_field("id")
db.char_field("name", 100, 0, 0)
db.char_field("email", 255, 0, 1)           # unique
db.int_field("age", 0, 0)
db.bool_field("active", 1, 0)
db.datetime_field("created_at", 0, 1, 0)    # auto_now_add
db.json_field("metadata", 1)                # nullable
db.uuid_field("public_id", 0, 1)            # UUID
db.decimal_field("balance", 10, 2, 0)       # DECIMAL(10,2)
db.array_field("tags", "TEXT", 1)            # PG: TEXT[], MySQL: JSON
db.inet_field("ip", 1)                      # PG: INET, MySQL: VARCHAR(45)
db.custom_field("extra", "HSTORE", 1)       # any native DB type

let sql = db.create_table_sql("users")
db.execute(sql)

Debugging

db.set_debug(1)            # log protocol messages
db.dump_results()          # print result table to stderr
print(db.connection_info())
print(db.driver())         # "postgres", "mysql", or "none"

ORM QuerySet (Django-style)

# Django-style filter with __ lookups
db.find("users")
db.filter("age__gte", "18")
db.filter("name__contains", "alice")
db.order_by("-created_at")
db.limit(10)
let rows = db.fetch()

# Aggregation
db.find("orders")
db.annotate("total", "SUM", "amount")
db.group_by("customer_id")
let rows = db.fetch()

# Insert
db.find("users")
db.set("name", "Alice")
db.set("email", "alice@example.com")
db.do_insert()

# Bulk insert
db.find("users")
db.bulk_begin(2)
db.bulk_col("name")
db.bulk_col("email")
db.bulk_row2("Alice", "alice@example.com")
db.bulk_row2("Bob", "bob@example.com")
db.bulk_execute()

Supported Lookups

exact, iexact, contains, icontains, startswith, istartswith, endswith, iendswith, gt, gte, lt, lte, ne, in, range, isnull, year, month, day, hour, minute, second, week, quarter, json_has, json_contains

HAVING Clause

Filter aggregate results after group_by() + annotate():

db.objects("orders")
db.group_by("customer_id")
db.annotate("total", "SUM", "amount")
db.having("SUM(amount) > 100")        # raw condition
db.fetch_all()

# Parameterized (safe for user input):
db.objects("orders")
db.group_by("customer_id")
db.annotate("order_count", "COUNT", "*")
db.having_val("COUNT(*) >", "5")       # value is $N-bound
db.fetch_all()

Multi-column Ordering

Append multiple ORDER BY columns (unlike sort_by() which overwrites):

db.objects("employees")
db.order_by_add("-salary")   # primary: salary DESC
db.order_by_add("name")     # secondary: name ASC
db.fetch_all()
# → ORDER BY salary DESC, name ASC

Multi-column Update

Update multiple fields in a single SQL statement:

db.objects("users")
db.filter_by("id", "1")
db.update_set("name", "Alice")
db.update_set("email", "alice@example.com")
db.update_exec()
# → UPDATE users SET name=$1, email=$2 WHERE id = $3

Pagination

Page-based helpers that calculate LIMIT/OFFSET:

db.objects("products")
db.filter_by("category", "electronics")

# Get total count (ignores LIMIT/OFFSET)
let total = db.total_count()

# Fetch page 2 (25 items per page)
db.paginate(2, 25)    # → LIMIT 25 OFFSET 25
db.fetch_all()

print(f"Showing page 2 of {total} results")

Soft Delete

Mark rows as deleted without removing them from the database:

# Enable soft-delete mode (column defaults to "is_deleted")
db.objects("users")
db.soft_delete_mode("is_deleted")

# All queries auto-filter: WHERE is_deleted = FALSE
db.fetch_all()   # only active users

# Soft-delete a row (UPDATE SET is_deleted = TRUE)
db.objects("users")
db.soft_delete_mode("is_deleted")
db.filter_by("id", "42")
db.soft_delete()

# Include deleted rows (bypass auto-filter)
db.objects("users")
db.soft_delete_mode("is_deleted")
db.with_deleted()
db.fetch_all()   # all users, including deleted

# Restore a soft-deleted row
db.objects("users")
db.soft_delete_mode("is_deleted")
db.with_deleted()
db.filter_by("id", "42")
db.restore()     # SET is_deleted = FALSE

# Permanently delete (real DELETE FROM, ignores soft-delete)
db.objects("users")
db.filter_by("id", "42")
db.hard_delete()

Row Locking (select_for_update)

Lock rows for concurrent-safe reads within a transaction:

import db

# Basic row lock — blocks until lock acquired
db.begin()
db.objects("accounts")
db.filter_by("id", "1")
db.select_for_update("")      # FOR UPDATE
db.fetch_all()
db.update("balance", "500")
db.commit()

# Non-blocking — error if row already locked
db.begin()
db.objects("accounts")
db.filter_by("id", "1")
db.select_for_update("nowait")    # FOR UPDATE NOWAIT
db.fetch_all()
db.commit()

# Skip locked rows — useful for job queues
db.begin()
db.objects("jobs")
db.filter_by("status", "pending")
db.select_for_update("skip_locked")  # FOR UPDATE SKIP LOCKED
db.limit(10)
db.fetch_all()
db.commit()

Modes: "" (blocking), "nowait" (error if locked), "skip_locked" (skip locked rows). Works with both PostgreSQL and MySQL.

Bulk Fetch (in_bulk)

Fetch multiple rows by primary key list in a single query:

import db

db.objects("users")
let n = db.in_bulk("1,5,10,42", "id")
for i in range(n):
    print(db.get_value(i, 0))  # prints each user's id

The second argument is the PK column name (defaults to "id" if empty).

Column Projection (values_list)

Restrict SELECT to specific columns:

import db

db.objects("users")
db.values_list("id, name, email")
let n = db.fetch_all()
# Result set only contains id, name, email columns

Equivalent to Django's values_list("id", "name", "email") or only("id", "name", "email").

Database Functions

SQL expression builders for use in annotate(), filter(), and order_by():

import db

# String functions
db.func_lower("name")      # → LOWER(name)
db.func_upper("name")      # → UPPER(name)
db.func_length("name")     # → LENGTH(name) / CHAR_LENGTH(name)
db.func_concat("a, b")     # → CONCAT(a, b)

# Type conversion
db.func_cast("price", "INTEGER")  # → CAST(price AS INTEGER)
db.func_coalesce("nickname, name, 'Anonymous'")  # → COALESCE(...)

# Numeric
db.func_abs("balance")     # → ABS(balance)
db.func_greatest("a, b")   # → GREATEST(a, b)
db.func_least("a, b")      # → LEAST(a, b)

# Timestamps
db.func_now()              # → NOW()

Use with annotate():

db.objects("users")
db.annotate("lname", "IDENTITY", db.func_lower("name"))
db.fetch_all()

All functions are dialect-aware (PG vs MySQL).

Case/When Conditional Expressions

Build SQL CASE expressions incrementally:

import db

# Categorize users by age
db.case_when("age < 18", "'minor'")
db.case_when("age >= 18 AND age < 65", "'adult'")
db.case_when("age >= 65", "'senior'")
db.case_else("'unknown'")
let expr = db.case_end("age_group")

# Use with annotate
db.objects("users")
db.annotate("age_group", "IDENTITY", expr)
db.fetch_all()

Generates: CASE WHEN age < 18 THEN 'minor' WHEN age >= 18 AND age < 65 THEN 'adult' WHEN age >= 65 THEN 'senior' ELSE 'unknown' END AS age_group

The builder is thread-safe and resets after case_end().

Model Instance Hydration + Save

After querying, load a result row into a model instance cache, modify fields, and persist:

import db

# Fetch a user
db.objects("users")
db.filter("id", "1")
db.fetch_all()

# Hydrate row 0 into instance cache
db.hydrate("users", 0)

# Read fields
let name = db.instance_get("name")
let pk = db.instance_pk()  # > 0 for existing rows

# Modify and save (UPDATE since PK > 0)
db.instance_set("name", "Alice Updated")
db.instance_save()  # Executes: UPDATE users SET name='Alice Updated' WHERE id=1

# Clear when done
db.instance_clear()

For new instances (PK == 0), instance_save() performs an INSERT instead.

Field Validation

Register constraints per (table, field) and validate before saving:

import db

# Register validators (min_len, max_len, min_val, max_val — use 0 for unused)
db.add_validator("users", "name", 2, 100, 0, 0)     # name: 2-100 chars
db.add_validator("users", "age", 0, 0, 0, 150)      # age: 0-150

# Validate before save
db.hydrate("users", 0)
db.instance_set("name", "")
let err = db.validate_instance()
if err != "":
    print("Validation failed: " + err)

Signals

Fire lifecycle hooks (pre/post save/delete):

# Signal types: 0=pre_save, 1=post_save, 2=pre_delete, 3=post_delete
db.fire_signal("users", 0)  # fire pre_save handlers
db.instance_save()
db.fire_signal("users", 1)  # fire post_save handlers

Signal handlers are registered via C function pointers in the runtime and can veto operations by returning -1.

Many-to-Many Relationships

Register and manage M2M relationships with automatic junction tables:

import db

# Register M2M between articles and tags
db.m2m("articles", "tags", "article_tags")

# Generate junction table SQL
let sql = db.m2m_create_sql("article_tags")
# Creates: article_tags (id, articles_id, tags_id, UNIQUE(articles_id, tags_id))

# Manage links
db.m2m_add("article_tags", "1", "5")     # article 1 ↔ tag 5
db.m2m_add("article_tags", "1", "3")     # article 1 ↔ tag 3
db.m2m_remove("article_tags", "1", "5")  # remove link
db.m2m_clear("article_tags", "1")        # remove all tags for article 1

# Query all related tags for an article
db.m2m_all("article_tags", "1")          # SELECT tags_id FROM article_tags WHERE articles_id = 1

Add operations are idempotent (ON CONFLICT DO NOTHING / INSERT IGNORE). Junction tables include cascading foreign keys.

Abstract Model Inheritance

Share fields across models using abstract base classes:

import db

# Define abstract mixin (won't create a table)
db.model("timestamp_mixin")
db.datetime_field("created_at", 0, 1, 0)  # auto_now_add
db.datetime_field("updated_at", 1, 0, 0)  # auto_now
db.set_abstract("timestamp_mixin")

# Child inherits all parent fields
db.model("articles")
db.auto_field("id")
db.char_field("title", 200, 0, 0)
db.inherit("articles", "timestamp_mixin")
# articles now has: id, title, created_at, updated_at

Inheritance copies fields, unique constraints, and indexes from parent to child. The parent model is skipped during table creation.

Dirty Field Tracking

Only update columns that actually changed:

import db

db.hydrate("users", 0)            # load from query result
db.instance_set("name", "Bob")    # mark "name" as dirty

if db.is_dirty() == 1:
    let changed = db.dirty_fields()  # "name"
    db.instance_save()               # UPDATE users SET name='Bob' WHERE id=1
    # Only the dirty field is included in the UPDATE

Instance Delete & Refresh

# Delete the hydrated instance
db.instance_delete()               # DELETE FROM users WHERE id=1
                                    # Fires pre_delete/post_delete signals

# Re-fetch a stale instance
db.refresh_from_db()               # SELECT * FROM users WHERE id=1
                                    # Reloads all fields, clears dirty flags

OneToOneField

ForeignKey with UNIQUE constraint:

db.model("user_profiles")
db.auto_field("id")
db.one_to_one_field("user_id", "users", "id", "CASCADE", 0)
db.char_field("bio", 500, 0, 0)

QuerySet Reverse

db.objects("users")
db.order_by("name")    # name ASC
db.reverse()           # name DESC (flips all ORDER BY directions)

Additional Database Functions

# String functions
db.func_substr("name", 1, 5)              # SUBSTR(name, 1, 5)
db.func_trim("name")                       # TRIM(name)
db.func_left("name", 3)                    # LEFT(name, 3)
db.func_right("name", 3)                   # RIGHT(name, 3)
db.func_replace("name", "old", "new")      # REPLACE(name, 'old', 'new')

# Numeric functions
db.func_round("price", 2)                  # ROUND(price, 2)
db.func_ceil("price")                       # CEIL(price)
db.func_floor("price")                      # FLOOR(price)

Custom Managers

Reusable named query scopes:

# Register a manager with a pre-defined filter
db.register_manager("users", "active", "is_active__exact=true")
db.register_manager("posts", "published", "status__exact=published")

# Use the manager — auto-applies filter
db.use_manager("users", "active")
let count = db.fetch_all()  # SELECT * FROM users WHERE is_active = true

Reverse Relations

Query from FK target back to source:

# Django equivalent: user.posts.all()
let posts = db.reverse_query("posts", "author_id", "1")

# Count reverse related objects
let n = db.reverse_count("posts", "author_id", "1")

Multi-Column Update

db.objects("users")
db.filter_by("id", "1")
db.update_pairs("name=Bob,age=30")   # UPDATE users SET name='Bob', age=30 WHERE id=1

Multi-Table Inheritance

Create parent-child table relationships where the child table references the parent via a OneToOne FK:

import db

# Parent model
db.model("places")
db.auto_field("id")
db.char_field("name", 100, 0, 0)
db.char_field("address", 200, 0, 0)

# Child model inherits from parent
db.model("restaurants")
db.auto_field("id")
db.bool_field("serves_pizza", 0, 0)
db.multi_table_inherit("restaurants", "places")
# Auto-adds: places_ptr_id INTEGER NOT NULL UNIQUE REFERENCES places(id) ON DELETE CASCADE

let sql1 = db.create_table_sql("places")
let sql2 = db.create_table_sql("restaurants")
db.execute(sql1)
db.execute(sql2)

Django equivalent: class Restaurant(Place): ... — creates both tables with an automatic FK pointer.

Proxy Models

Create an alias for an existing table — same data, different name. Useful for attaching different managers:

import db

db.model("users")
db.auto_field("id")
db.char_field("name", 100, 0, 0)
db.bool_field("is_active", 0, 0)

# Register a proxy — no new table created
db.proxy_model("active_users", "users")
db.register_manager("active_users", "default", "is_active__exact=true")

# Use the proxy with its manager
db.use_manager("active_users", "default")
let n = db.fetch_all()  # queries "users" table with is_active filter

# Resolve proxy back to base table
let base = db.resolve_proxy("active_users")  # → "users"

Django equivalent: class Meta: proxy = True.

Built-in Audit Trail

Track all changes to a model automatically — no external packages needed:

import db

# Enable auditing for a model
db.enable_audit("users")

# Create the history table
let sql = db.create_audit_table("users")
db.execute(sql)
# Creates: users_history (history_id, action, record_id, changed_at, changed_by, + all model fields)

# Log changes (called automatically by save/delete when auditing is enabled)
db.audit_record("users", "INSERT", "1")
db.audit_record("users", "UPDATE", "1")

# Query history for a specific record
let n = db.audit_log("users", "1")  # newest first
for i in range(n):
    let action = db.get_field(i, "action")
    let when = db.get_field(i, "changed_at")
    print(f"{action} at {when}")

# Check if auditing is enabled
if db.is_audited("users") == 1:
    print("users table is audited")

Django equivalent: django-simple-history — but built-in with zero dependencies.

Time-Travel Queries

Query the state of a record at a specific point in time (requires audit trail):

import db

db.enable_audit("users")

# What did user #1 look like on Jan 1, 2024?
db.as_of("users", "1", "2024-01-01 00:00:00")
let name = db.get_field(0, "name")
print(f"Name on Jan 1: {name}")

Django equivalent: user.history.as_of(datetime(2024, 1, 1)).

Runtime Field Validation

Check if a field name exists on a model with "did you mean?" suggestions for typos:

import db

db.model("users")
db.auto_field("id")
db.char_field("name", 100, 0, 0)
db.char_field("email", 255, 0, 0)

# Validate field names at runtime
let ok = db.check_field("users", "name")    # → 1 (valid)
let bad = db.check_field("users", "nmae")   # → 0, prints: WARNING: did you mean 'name'?
let unk = db.check_field("users", "xyz")    # → 0, prints: WARNING: unknown field 'xyz'

Useful for debugging and building safer query builders.

File-Based Migrations

For real projects, generate version-controlled migration files with portable operations:

db.makemigrations("migrations")          # Generate from ORM diff
db.migrate_dir("migrations")             # Apply pending
db.rollback_dir("migrations")            # Undo last
db.migration_status_dir("migrations")    # Show applied/pending

# Multi-app: apply across apps in dependency order
db.migrate_all(["accounts/migrations", "orders/migrations"])

Migration Squashing

Consolidate many migration files into one:

# Squash all migrations in a directory into a single file
db.squash_migrations("migrations", "0001_squashed")

Django equivalent: python manage.py squashmigrations.

Data Migrations

Run arbitrary code during migrations:

# Register a data migration callback, then run it
db.op_run_code("uppercase_names")

Django equivalent: RunPython(forwards, reverse).

Dry Run

Preview what migrations would generate without writing files:

let preview = db.makemigrations_dryrun("migrations")
print(preview)

Django equivalent: python manage.py makemigrations --dry-run.

Inspect Database

Reverse-engineer model definitions from an existing database:

let code = db.inspectdb()
print(code)
# Outputs Desi code with db.model/field calls matching existing tables

Django equivalent: python manage.py inspectdb.

See Migrations for the full guide.

Internals

Under the hood, each query chain (db.objects() → db.filter_by() → db.fetch_all()) creates a QuerySet handle — a heap-allocated struct that holds all query state (table, WHERE clause, parameters, etc.).

The compiler emits explicit lifecycle management for each chain:

1. Allocate — __qs_handle_new("table") creates a fresh handle
2. Bind — __qs_handle_bind(qs) installs it as active for the chain
3. Use — intermediate calls (filter, order_by, etc.) operate on the active handle
4. Free — __qs_handle_free(qs) deallocates after the terminal call

This means:

• Concurrent queries are safe — each spawn-ed task gets its own isolated query state via thread-local binding.
• No parameter limits — parameters, INSERT fields, and UPDATE fields grow dynamically as needed.
• Automatic cleanup — the handle is freed after the terminal call (fetch_all, update_exec, etc.).
• Deterministic lifetime — the compiler controls handle allocation and deallocation, preventing leaks.

You don't need to manage handles manually — the compiler and runtime do it for you.

See QuerySet Handles for the full architecture guide.

See Also

• Migrations — File-based migrations, op-based DSL, multi-app support
• ORM Models — @model decorator and field types
• PostgreSQL — PG-specific types, auth, raw SQL examples
• MySQL — MySQL-specific types, auth, raw SQL examples