Skip to content

db — all cases

Every case file under src/clauz3/stdlib/db/tests/cases/, inlined. _pass cases must prove; _fail cases must not.

See the curated walk-through for the trusted module and contract vocabulary.

src/clauz3/stdlib/db/tests/cases/create_and_insert_pass.py

from tools.db.trusted import contracts as db
from tools.db.trusted.effects import create_table, insert_row

import clauz3


class Transaction(clauz3.Row):
    id: int
    merchant: str
    amount_cents: int


@clauz3.guarantee(
    db.only_writes_databases([clauz3.FilePath("/tmp/transactions.duckdb")])
)
@clauz3.guarantee(db.only_writes_tables(["transactions"]))
@clauz3.guarantee(db.writes_at_most(3))
@clauz3.guarantee(db.rows_inserted_at_most(2))
def main() -> None:
    create_table(Transaction, "/tmp/transactions.duckdb", "transactions")
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=1, merchant="Fictional Coffee", amount_cents=450),
    )
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=2, merchant="Fictional Books", amount_cents=1800),
    )

src/clauz3/stdlib/db/tests/cases/insert_limit_fail.py

from tools.db.trusted import contracts as db
from tools.db.trusted.effects import create_table, insert_row

import clauz3


class Transaction(clauz3.Row):
    id: int
    merchant: str
    amount_cents: int


@clauz3.guarantee(
    db.only_writes_databases([clauz3.FilePath("/tmp/transactions.duckdb")])
)
@clauz3.guarantee(db.only_writes_tables(["transactions"]))
@clauz3.guarantee(db.writes_at_most(3))
@clauz3.guarantee(db.rows_inserted_at_most(1))
def main() -> None:
    create_table(Transaction, "/tmp/transactions.duckdb", "transactions")
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=1, merchant="Fictional Coffee", amount_cents=450),
    )
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=2, merchant="Fictional Books", amount_cents=1800),
    )

src/clauz3/stdlib/db/tests/cases/loop_rows_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


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


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

src/clauz3/stdlib/db/tests/cases/no_db_reads_fail.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.no_db_reads())
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {})

src/clauz3/stdlib/db/tests/cases/no_db_reads_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db

import clauz3


@clauz3.guarantee(db.no_db_reads())
def main() -> None:
    pass

src/clauz3/stdlib/db/tests/cases/only_read_under_fail.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.only_read_under("/repo/data"))
def main() -> None:
    read_rows(UserRow, "/tmp/users.csv", "users", {})

src/clauz3/stdlib/db/tests/cases/only_read_under_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.only_read_under("/repo/data"))
def main() -> None:
    read_rows(UserRow, "/repo/data/users.csv", "users", {})

src/clauz3/stdlib/db/tests/cases/only_table_fail.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.only_table("users"))
def main() -> None:
    read_rows(UserRow, "/repo/data", "admins", {})

src/clauz3/stdlib/db/tests/cases/only_table_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.only_table("users"))
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {})

src/clauz3/stdlib/db/tests/cases/only_where_fail.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


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


@clauz3.guarantee(db.only_where({"consented": True}))
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {"consented": False})

src/clauz3/stdlib/db/tests/cases/only_where_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


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


@clauz3.guarantee(db.only_where({"consented": True}))
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {"consented": True})

src/clauz3/stdlib/db/tests/cases/reads_at_most_fail.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.reads_at_most(1))
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {})
    read_rows(UserRow, "/repo/data", "admins", {})

src/clauz3/stdlib/db/tests/cases/reads_at_most_pass.py

# ruff: noqa: F821
from tools.db.trusted import contracts as db
from tools.db.trusted.effects import read_rows

import clauz3


class UserRow(clauz3.Row):
    email: str


@clauz3.guarantee(db.reads_at_most(2))
def main() -> None:
    read_rows(UserRow, "/repo/data", "users", {})
    read_rows(UserRow, "/repo/data", "admins", {})

src/clauz3/stdlib/db/tests/cases/write_other_database_fail.py

from tools.db.trusted import contracts as db
from tools.db.trusted.effects import create_table, insert_row

import clauz3


class Transaction(clauz3.Row):
    id: int
    merchant: str
    amount_cents: int


@clauz3.guarantee(
    db.only_writes_databases([clauz3.FilePath("/tmp/transactions.duckdb")])
)
@clauz3.guarantee(db.only_writes_tables(["transactions"]))
@clauz3.guarantee(db.writes_at_most(3))
@clauz3.guarantee(db.rows_inserted_at_most(2))
def main() -> None:
    create_table(Transaction, "/tmp/other.duckdb", "transactions")
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=1, merchant="Fictional Coffee", amount_cents=450),
    )
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=2, merchant="Fictional Books", amount_cents=1800),
    )

src/clauz3/stdlib/db/tests/cases/write_other_table_fail.py

from tools.db.trusted import contracts as db
from tools.db.trusted.effects import create_table, insert_row

import clauz3


class Transaction(clauz3.Row):
    id: int
    merchant: str
    amount_cents: int


@clauz3.guarantee(
    db.only_writes_databases([clauz3.FilePath("/tmp/transactions.duckdb")])
)
@clauz3.guarantee(db.only_writes_tables(["transactions"]))
@clauz3.guarantee(db.writes_at_most(3))
@clauz3.guarantee(db.rows_inserted_at_most(2))
def main() -> None:
    create_table(Transaction, "/tmp/transactions.duckdb", "other")
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=1, merchant="Fictional Coffee", amount_cents=450),
    )
    insert_row(
        "/tmp/transactions.duckdb",
        "transactions",
        Transaction(id=2, merchant="Fictional Books", amount_cents=1800),
    )