Skip to content

db

A trusted clauz3 layer for tabular reads and structured DuckDB writes. Read rows from CSV, Parquet, DuckDB, or a folder of CSV/Parquet tables. Create DuckDB tables and insert typed rows, with separate read and write guarantees.

Unlike the layers under examples/, these effects are real, not mocks. The prover never runs the body — it records each call as a fact, materializes a symbolic list[RowT], and proves the contract. The body runs only under clauz3 run, after a program is proved and approved.

Install

clauz3 install stdlib:db

The runtime bodies use DuckDB. Install clauz3[db] (or duckdb>=1.4,<2) in the environment that runs proved programs. Proof-only tests do not require DuckDB because effect bodies are skipped by the prover.

Effects

Import effects from tools.db.trusted.effects:

  • read_rows(schema, path, table, where) -> list[RowT] — read up to 100 rows from path and instantiate each row as schema. schema must be a clauz3.Row subclass. path may be a DuckDB database file, a CSV file, a Parquet file, or a directory containing <table>.csv / <table>.parquet. For a single CSV or Parquet file, table must match the file stem. CSV and Parquet inputs are queried through DuckDB in memory.

  • create_table(schema, path, table) -> None — create a new table with columns from a clauz3.Row schema; create the DuckDB file if needed. Fields are str/int/bool (VARCHAR/BIGINT/BOOLEAN), all NOT NULL. Existing tables cause an error and are never replaced.

  • insert_row(path, table, row) -> None — append one typed Row value. The file and table must already exist. Each call commits atomically and independently; a later failure does not roll back earlier successful calls.

Writes support .duckdb, .ddb and .db files with existing parent folders. There is no arbitrary SQL API: identifiers are quoted and values are bound parameters. Write connections disable external access, extension loading and temporary spill files. DuckDB may write an adjacent .wal journal. Paths are lexical, without tilde expansion or physical symlink confinement.

The schema argument is the type-system hook. The function is generic at runtime, but the prover uses the concrete Row subclass passed at the call site to type row.email, row.id, and other field accesses inside a loop.

Contracts

Import with from tools.db.trusted import contracts as db:

  • db.no_guarantees() — explicit null contract.
  • db.no_db_reads() — no read_rows calls; writes are governed separately.
  • db.reads_at_most(count) — at most count database reads occur.
  • db.only_read_under(root) — every database source path starts with root.
  • db.never_read_under(prefix) — no database source path starts with prefix.
  • db.only_table(table) — every read targets exactly table.
  • db.only_tables(tables) — every read targets one of tables.
  • db.only_where(filter_dict) — every read uses exactly filter_dict.

Write contracts:

  • db.no_db_writes() — no table creation or insertion.
  • db.only_writes_databases(files: list[FilePath]) — exact destination database file names, including their adjacent journals. Import FilePath from clauz3. Use the same list for one or many databases.
  • db.only_writes_tables(tables: list[str]) — creation and insertion target only these table names. Existing only_table / only_tables govern reads only.
  • db.writes_at_most(count) — maximum combined number of create/insert calls.
  • db.rows_inserted_at_most(count) — maximum number of inserted rows.

Database reads are governed by this layer's own path policies (only_read_under, never_read_under). They were once covered by filesystem.only_read_under as well, through a shared read marker; an effect is addressed by name now, so name db.only_read_under for database sources and filesystem.only_read_under for file reads.

Example

import clauz3
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows


class UserRow(clauz3.Row):
    name: str
    email: str
    consented: bool


@clauz3.guarantee(db.only_read_under("/repo/data"))
@clauz3.guarantee(db.only_table("users"))
@clauz3.guarantee(db.reads_at_most(1))
def main() -> None:
    for row in read_rows(
        UserRow,
        "/repo/data",
        "users",
        {"consented": True},
    ):
        _ = row.email

Creating a database

Sessions offer the tested create-duckdb recipe when the installed db library matches the bundled version. It defines a Transaction row, creates a table and inserts fictional transactions. Use integer cents for amounts and strings for ISO dates; Row fields currently support str, int and bool.

Construct a row with all fields as keywords, then pass it to insert_row. Lists of constructed Row values are not supported by the prover yet. To copy query results, iterate read_rows and insert each returned row individually. Write scope is independent of read scope, and filesystem contracts do not cover database writes. There are no update, delete, drop, or SQL execution effects in this interface.

After updating clauz3, refresh an unmodified installed library with clauz3 install --force --skills stdlib:db and restart the running session. Preserve local library customizations before using --force.

Tests

clauz3 test stdlib:db

tests/cases/*_pass.py must prove; tests/cases/*_fail.py must not. The Justfile also recreates tests/runtime-data/ on each run. It always writes a small CSV folder fixture and, when DuckDB is installed, also writes Parquet and DuckDB fixtures before running a runtime smoke test. Without DuckDB, the smoke test skips after fixture setup.

Source

src/clauz3/stdlib/db/tools/db/trusted/effects.py

"""Trusted database reads and structured DuckDB writes.

The prover records calls to ``read_rows`` and materializes the result as a
symbolic ``list[RowT]``. The body runs only under ``clauz3 run`` and uses
DuckDB as the execution engine for DuckDB files, CSV files, Parquet files, or
directories containing ``<table>.csv`` / ``<table>.parquet`` files.
"""

from __future__ import annotations

from pathlib import Path
from typing import TypeVar

import deal

import clauz3
from clauz3._path_safety import (
    has_no_parent_segments,
    is_plain_path,
    is_single_path_component,
)

RowT = TypeVar("RowT", bound=clauz3.Row)

MAX_ROWS = 100
_DUCKDB_SUFFIXES = {".duckdb", ".ddb", ".db"}
_CSV_SUFFIXES = {".csv"}
_PARQUET_SUFFIXES = {".parquet"}
_SUPPORTED_VALUE_TYPES = (str, int, bool)


@clauz3.requires(
    lambda schema, path, table, where: has_no_parent_segments(path),
    message="path must not contain parent-directory segments",
)
@clauz3.requires(
    lambda schema, path, table, where: is_single_path_component(table),
    message="table must be a single path component",
)
@clauz3.requires(
    lambda schema, path, table, where: len(path) > 0 and len(table) > 0,
    message="path and table must be non-empty",
)
@clauz3.ensures(lambda result: len(result) <= 100)
@deal.has("read")
@clauz3.trusted
def read_rows(
    schema: type[RowT],
    path: str,
    table: str,
    where: dict[str, object],
) -> list[RowT]:
    """Read up to 100 rows from ``path``/``table`` as instances of ``schema``.

    ``path`` may point to a DuckDB database, one CSV or Parquet file, or a
    directory with files named ``<table>.csv`` / ``<table>.parquet``. CSV and
    Parquet sources are queried through an in-memory DuckDB connection.
    """
    if not isinstance(schema, type) or not issubclass(schema, clauz3.Row):
        raise TypeError("schema must be a clauz3.Row subclass")

    duckdb = _duckdb()
    connection, relation_sql = _connect(duckdb, Path(path).expanduser(), table)
    try:
        columns = tuple(schema.__annotations__)
        if not columns:
            raise ValueError(f"{schema.__name__} declares no columns")
        where_sql, params = _where_clause(where)
        sql = (
            f"SELECT {_select_list(columns)} FROM {relation_sql}"
            f"{where_sql} LIMIT {MAX_ROWS}"
        )
        rows = connection.execute(sql, params).fetchall()
    finally:
        connection.close()

    return [_row_from_tuple(schema, columns, row) for row in rows]


@clauz3.requires(
    lambda schema, path, table: is_plain_path(path),
    message="database path must have plain path segments",
)
@clauz3.requires(
    lambda schema, path, table: is_single_path_component(table),
    message="table must be a single path component",
)
@clauz3.requires(
    lambda schema, path, table: len(path) > 0 and len(table) > 0,
    message="path and table must be non-empty",
)
@deal.has("read", "write")
@clauz3.trusted
def create_table(schema: type[RowT], path: str, table: str) -> None:
    """Create a new table from a Row schema, creating the DuckDB file if needed.

    Supports str, int and bool columns, all NOT NULL. An existing table is an
    error, never replaced. The parent directory must already exist. Writes
    accept .duckdb, .ddb and .db files only, with no tilde expansion. DuckDB
    may also write its adjacent .wal journal. No arbitrary SQL is accepted.
    """
    columns = _write_columns(schema)
    sql_types = {str: "VARCHAR", int: "BIGINT", bool: "BOOLEAN"}
    definitions = ", ".join(
        f"{_quote_identifier(name)} {sql_types[kind]} NOT NULL"
        for name, kind in columns.items()
    )
    _validate_write_target(path, table)
    connection = _write_connection(path)
    try:
        connection.execute(f"CREATE TABLE {_quote_identifier(table)} ({definitions})")
    finally:
        connection.close()


@clauz3.requires(
    lambda path, table, row: is_plain_path(path),
    message="database path must have plain path segments",
)
@clauz3.requires(
    lambda path, table, row: is_single_path_component(table),
    message="table must be a single path component",
)
@clauz3.requires(
    lambda path, table, row: len(path) > 0 and len(table) > 0,
    message="path and table must be non-empty",
)
@deal.has("read", "write")
@clauz3.trusted
def insert_row(path: str, table: str, row: RowT) -> None:
    """Append one typed Row to an existing DuckDB table atomically.

    Supports str, int and bool fields. Values are bound parameters, never SQL
    fragments. Existing rows are preserved. The file and table must already
    exist; writes may include the adjacent .wal journal. Each call commits
    independently, so a later failure does not undo earlier successful calls.
    """
    _validate_write_target(path, table)
    columns = _write_columns(type(row))
    values = []
    for name, kind in columns.items():
        value = getattr(row, name)
        if type(value) is not kind:
            raise TypeError(f"column {name!r} must contain {kind.__name__}")
        values.append(value)
    if not Path(path).is_file():
        raise FileNotFoundError(f"database does not exist: {path}")
    names = _select_list(tuple(columns))
    placeholders = ", ".join("?" for _ in columns)
    connection = _write_connection(path)
    try:
        connection.execute(
            f"INSERT INTO {_quote_identifier(table)} ({names}) VALUES ({placeholders})",
            values,
        )
    finally:
        connection.close()


def _write_columns(schema: type[clauz3.Row]) -> dict[str, type]:
    if not isinstance(schema, type) or not issubclass(schema, clauz3.Row):
        raise TypeError("schema must be a clauz3.Row subclass")
    columns = dict(schema.__annotations__)
    if not columns or any(
        kind not in _SUPPORTED_VALUE_TYPES for kind in columns.values()
    ):
        raise TypeError("a write schema needs str, int or bool columns")
    return columns


def _validate_write_target(path: str, table: str) -> None:
    clauz3.FilePath(path)
    if Path(path).suffix.lower() not in _DUCKDB_SUFFIXES:
        raise ValueError("writes require a .duckdb, .ddb or .db file")
    if not table or not is_single_path_component(table):
        raise ValueError("table must be a nonempty single path component")


def _write_connection(path: str):
    # No spill files, extension downloads, replacement scans or external
    # file/network access through expressions stored in an existing database.
    return _duckdb().connect(
        path,
        config={
            "enable_external_access": "false",
            "autoload_known_extensions": "false",
            "autoinstall_known_extensions": "false",
            "temp_directory": "",
        },
    )


def _duckdb():
    try:
        import duckdb
    except ImportError as exc:
        raise RuntimeError(
            "stdlib:db requires the optional 'duckdb' package at run time. "
            "Install it in the environment that runs proved programs."
        ) from exc
    return duckdb


def _connect(duckdb, path: Path, table: str):
    if path.is_dir():
        connection = duckdb.connect(":memory:")
        return connection, _file_relation(_table_file(path, table))

    suffix = path.suffix.lower()
    if suffix in _DUCKDB_SUFFIXES:
        return duckdb.connect(str(path), read_only=True), _quote_identifier(table)
    if suffix in _CSV_SUFFIXES | _PARQUET_SUFFIXES:
        if table != path.stem:
            raise ValueError(
                f"table {table!r} must match file stem {path.stem!r} for "
                "single-file CSV/Parquet sources"
            )
        connection = duckdb.connect(":memory:")
        return connection, _file_relation(path)
    raise ValueError(
        f"unsupported database source {path!s}; expected DuckDB, CSV, Parquet, "
        "or a directory containing CSV/Parquet tables"
    )


def _table_file(root: Path, table: str) -> Path:
    candidates = [
        root / f"{table}.parquet",
        root / f"{table}.csv",
    ]
    existing = [candidate for candidate in candidates if candidate.is_file()]
    if len(existing) == 1:
        return existing[0]
    if existing:
        raise ValueError(f"ambiguous table {table!r}: both CSV and Parquet exist")
    raise FileNotFoundError(f"no CSV or Parquet file for table {table!r} in {root!s}")


def _file_relation(path: Path) -> str:
    suffix = path.suffix.lower()
    literal = _quote_literal(str(path))
    if suffix in _CSV_SUFFIXES:
        return f"read_csv_auto({literal})"
    if suffix in _PARQUET_SUFFIXES:
        return f"read_parquet({literal})"
    raise ValueError(f"unsupported table file {path!s}")


def _select_list(columns: tuple[str, ...]) -> str:
    return ", ".join(_quote_identifier(column) for column in columns)


def _where_clause(where: dict[str, object]) -> tuple[str, list[object]]:
    if not where:
        return "", []

    clauses: list[str] = []
    params: list[object] = []
    for key, value in where.items():
        if not isinstance(key, str) or not key:
            raise ValueError("where keys must be non-empty column names")
        if not isinstance(value, _SUPPORTED_VALUE_TYPES):
            raise TypeError("where values must be str, int, or bool")
        clauses.append(f"{_quote_identifier(key)} = ?")
        params.append(value)
    return " WHERE " + " AND ".join(clauses), params


def _row_from_tuple(
    schema: type[RowT],
    columns: tuple[str, ...],
    values: tuple[object, ...],
) -> RowT:
    data = {}
    for column, value in zip(columns, values, strict=True):
        if value is None:
            raise ValueError(f"column {column!r} was NULL")
        data[column] = _coerce_value(value, schema.__annotations__[column])
    return schema(**data)


def _coerce_value(value: object, expected: type) -> object:
    if expected is str:
        return str(value)
    if expected is int:
        return int(value)
    if expected is bool:
        return bool(value)
    raise TypeError(f"unsupported Row field type {expected!r}")


def _quote_identifier(value: str) -> str:
    if not value:
        raise ValueError("identifier must be non-empty")
    return '"' + value.replace('"', '""') + '"'


def _quote_literal(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"

src/clauz3/stdlib/db/tools/db/trusted/contracts.py

"""Database read and write policies over this library's structured calls."""

from tools.db.trusted.effects import create_table, insert_row, read_rows

from clauz3 import FilePath
from clauz3.spec import ContractSpec, contract, effect
from clauz3.spec import no_guarantees as core_no_guarantees

DbRead = effect(read_rows)
DbWrite = effect(create_table, insert_row)
DbInsert = effect(insert_row)


@contract
def no_guarantees() -> ContractSpec:
    """Make no guarantees about database effects."""
    return core_no_guarantees()


@contract
def no_db_reads() -> ContractSpec:
    """Forbid read_rows calls; creation and insertion are governed separately."""
    return DbRead.empty()


@contract
def reads_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` database reads occur."""
    return DbRead.count() <= count


@contract
def only_read_under(root: str) -> ContractSpec:
    """Guarantee every database source path starts with ``root``."""
    return DbRead.all(lambda e: e.path.startswith(root))


@contract
def never_read_under(prefix: str) -> ContractSpec:
    """Guarantee no database source path starts with ``prefix``."""
    return DbRead.all(lambda e: not e.path.startswith(prefix))


@contract
def only_table(table: str) -> ContractSpec:
    """Guarantee every database read targets exactly ``table``."""
    return DbRead.all(lambda e: e.table == table)


@contract
def only_tables(tables: list[str]) -> ContractSpec:
    """Guarantee every database read targets one of ``tables``."""
    return DbRead.all(lambda e: e.table in tables)


@contract
def only_where(filter_dict: dict[str, object]) -> ContractSpec:
    """Guarantee every database read uses exactly ``filter_dict``."""
    return DbRead.all(lambda e: e.where == filter_dict)


@contract(description="No database tool creates tables or inserts rows.")
def no_db_writes() -> ContractSpec:
    """Forbid database write calls; reads are allowed."""
    return DbWrite.empty()


@contract(
    description=(
        "Database tools write only to databases {files} and their adjacent "
        ".wal journals (paths compared lexically)."
    )
)
def only_writes_databases(files: list[FilePath]) -> ContractSpec:
    """Restrict table creation and insertion to these exact database files.

    One or many FilePath literals; an empty list forbids all write calls.
    Leading ./ is optional for relative paths. DuckDB journal writes are
    included; temporary spill files are disabled. No symlink confinement is
    implied, and filesystem or other libraries' writes are not covered.
    """
    if type(files) is not list or any(type(path) is not FilePath for path in files):
        raise TypeError("files must be a list of FilePath literals")
    names = []
    for path in files:
        bare = path.value.removeprefix("./")
        names.append(bare)
        if not bare.startswith("/"):
            names.append("./" + bare)
    return DbWrite.all(lambda e: e.path in names)


@contract(
    description="Database tools create or insert into only these tables: {tables}."
)
def only_writes_tables(tables: list[str]) -> ContractSpec:
    """Restrict write targets to these table names; read contracts are separate."""
    return DbWrite.all(lambda e: e.table in tables)


@contract(description="Database tools make at most {count} create or insert calls.")
def writes_at_most(count: int) -> ContractSpec:
    """Limit the combined number of create_table and insert_row calls."""
    return DbWrite.count() <= count


@contract(description="Database tools insert at most {count} rows in total.")
def rows_inserted_at_most(count: int) -> ContractSpec:
    """Limit the number of rows appended by insert_row calls."""
    return DbInsert.count() <= count

All cases

Browse the full set on the all-cases page.