API reference

Examples use imports from grugdb and the row classes shown below.

Row

Row is the base class for immutable records. Annotate each field with Column[T] or Unique[T] and pass values by keyword.

class User(Row):
    id: Column[int]
    name: Column[str]


user = User(id=1, name="Ada")

Source

@dataclass_transform(kw_only_default=True, frozen_default=True, field_specifiers=(column,))
class Row:
    __slots__ = ("_values",)
    _values: tuple[object, ...]
    _schema: ClassVar[tuple[FieldSpec, ...]] = ()

    def __init_subclass__(cls) -> None:
        cls._schema = build_schema(cls)

    @classmethod
    def fields(cls) -> tuple[FieldSpec, ...]:
        return cls._schema

    def __init__(self, /, **values: object) -> None:
        if type(self) is Row:
            raise TypeError("define a model by subclassing Row")
        unknown = values.keys() - {spec.name for spec in self._schema}
        if unknown:
            raise TypeError(f"unknown fields: {', '.join(sorted(unknown))}")
        packed: list[object] = []
        for spec in self._schema:
            value = values.get(spec.name, spec.default)
            if value is MISSING:
                raise TypeError(f"missing field: {spec.name}")
            if not matches(value, spec.hint):
                raise TypeError(f"invalid value for {spec.name}: {value!r}")
            packed.append(value)
        object.__setattr__(self, "_values", tuple(packed))

    def replace(self, /, **changes: object) -> Self:
        values = {spec.name: self._values[spec.position] for spec in self._schema}
        return type(self)(**(values | changes))

    def __setattr__(self, name: str, value: object) -> None:
        raise AttributeError("rows are immutable; use replace()")

    def __delattr__(self, name: str) -> None:
        raise AttributeError("rows are immutable")

    def __repr__(self) -> str:
        fields = ", ".join(f"{s.name}={self._values[s.position]!r}" for s in self._schema)
        return f"{type(self).__name__}({fields})"

    def __eq__(self, other: object) -> bool:
        return (
            isinstance(other, Row) and type(self) is type(other) and self._values == other._values
        )

    def __hash__(self) -> int:
        return hash((type(self), self._values))

    def __reduce__(self) -> tuple[object, tuple[type[Row], tuple[object, ...]]]:
        return restore_row, (type(self), self._values)

RowId

An internal row identifier, represented at runtime as an integer.

The type checker distinguishes RowId from ordinary integers. Tables assign increasing IDs, separate from any application primary key. A primary key is a declared unique, non-null key.

row_id = users.insert(user)
users.get(row_id)

Source

RowId = NewType("RowId", int)

Column

Column[T] declares a field holding a value of type T.

On a row, user.name is the value. On the class, User.name is a Field that names the column, for use in queries.

class User(Row):
    name: Column[str]
    age: Column[int | None] = column(default=None)


User(name="Ada").name  # "Ada"
users.scan().where(User.age >= 18)

Source

class Column[K]:
    __slots__ = ("_field", "default")

    def __init__(self, default: object = MISSING) -> None:
        self.default = default
        self._field: Field[Any, K] | None = None

    def bind(self, model: type[Row], spec: FieldSpec) -> None:
        self._field = Field(model, spec)

    @property
    def field(self) -> Field[Any, K]:
        if self._field is None:
            raise TypeError("field is not part of a Row class")
        return self._field

    @overload
    def __get__[T: Row](self, instance: None, owner: type[T]) -> Field[T, K]: ...

    @overload
    def __get__(self, instance: Row, owner: type[Row] | None = None) -> K: ...

    def __get__(self, instance: Row | None, owner: type[Row] | None = None) -> object:
        if instance is None:
            return self.field
        return instance._values[self.field.spec.position]

    def __set__(self, instance: Row, value: K) -> None:
        raise AttributeError("rows are immutable; use replace()")

column

column(default=...) gives a field a default value.

class User(Row):
    age: Column[int | None] = column(default=None)

Source

def column[K](*, default: K) -> Column[K]:
    return Column[K](default)

Unique

Unique[T] declares a field whose values must be unique across a table.

On the class, Contact.email is a UniqueField you can pass to Table.lookup. Creating a table with unique fields raises NotImplementedError until you build indexes.

class Contact(Row):
    email: Unique[str]

Source

class Unique[K](Column[K]):
    __slots__ = ()

    def bind(self, model: type[Row], spec: FieldSpec) -> None:
        self._field = UniqueField(model, spec)

    @overload
    def __get__[T: Row](self, instance: None, owner: type[T]) -> UniqueField[T, K]: ...

    @overload
    def __get__(self, instance: Row, owner: type[Row] | None = None) -> K: ...

    def __get__(self, instance: Row | None, owner: type[Row] | None = None) -> object:
        if instance is None:
            return self.field
        return instance._values[self.field.spec.position]

class Column[K]:
    __slots__ = ("_field", "default")

    def __init__(self, default: object = MISSING) -> None:
        self.default = default
        self._field: Field[Any, K] | None = None

    def bind(self, model: type[Row], spec: FieldSpec) -> None:
        self._field = Field(model, spec)

    @property
    def field(self) -> Field[Any, K]:
        if self._field is None:
            raise TypeError("field is not part of a Row class")
        return self._field

    @overload
    def __get__[T: Row](self, instance: None, owner: type[T]) -> Field[T, K]: ...

    @overload
    def __get__(self, instance: Row, owner: type[Row] | None = None) -> K: ...

    def __get__(self, instance: Row | None, owner: type[Row] | None = None) -> object:
        if instance is None:
            return self.field
        return instance._values[self.field.spec.position]

    def __set__(self, instance: Row, value: K) -> None:
        raise AttributeError("rows are immutable; use replace()")

Field

User.age on the class is a Field[User, int | None]. It names one column of a model. Comparing it with a value builds a condition instead of a bool.

User.age >= 18  # Compare(User.age, ">=", 18)
User.id == Order.user_id  # JoinOn(User.id, Order.user_id)

Source

@dataclass(frozen=True, slots=True, eq=False)
class Field[T, K]:
    model: type[T]
    spec: FieldSpec

    @property
    def name(self) -> str:
        return self.spec.name

    def read(self, row: Value) -> Value:
        if type(row) is self.model:
            return row._values[self.spec.position]
        if type(row) is tuple:
            sides = [side for side in row if type(side) is self.model]
            if len(sides) == 1:
                return sides[0]._values[self.spec.position]
            if sides:
                raise TypeError(f"{self} is ambiguous: the row joins {self.model.__name__} twice")
        raise TypeError(f"{self} cannot be read from {type(row).__name__}")

    def check(self, value: K) -> K:
        from grugdb.schema import matches

        if not matches(value, self.spec.hint):
            raise TypeError(f"invalid value for {self}: {value!r}")
        return value

    @overload
    def __eq__[U](self, other: Field[U, Any]) -> JoinOn[T, U]: ...

    @overload
    def __eq__(self, other: K) -> Compare[T]: ...

    def __eq__(self, other: Any) -> Any:  # ty: ignore[invalid-method-override]
        if isinstance(other, Field):
            return JoinOn(self, other)
        return Compare(self, Op.EQ, self.check(other))

    def __ne__(self, other: K) -> Compare[T]:  # ty: ignore[invalid-method-override]
        return Compare(self, Op.NE, self.check(other))

    def __lt__(self, other: K) -> Compare[T]:
        return Compare(self, Op.LT, self.check(other))

    def __le__(self, other: K) -> Compare[T]:
        return Compare(self, Op.LE, self.check(other))

    def __gt__(self, other: K) -> Compare[T]:
        return Compare(self, Op.GT, self.check(other))

    def __ge__(self, other: K) -> Compare[T]:
        return Compare(self, Op.GE, self.check(other))

    def __repr__(self) -> str:
        return f"{self.model.__name__}.{self.name}"

    def is_none(self) -> IsNone[T]:
        return IsNone(self)

    def in_(self, values: Iterable[K]) -> In[T]:
        return In(self, frozenset(self.check(value) for value in values))

    def desc(self) -> Desc[T]:
        return Desc(self)

Expr

Expr[T] is a condition on rows of model T. Combine conditions with &, |, and ~. Python's and, or, and not can't be overloaded, so they raise.

adults = users.scan().where((User.age >= 18) & ~User.name.in_(["root"]))

Source

class Expr[T](ABC):
    @abstractmethod
    def test(self, row: object) -> bool: ...

    def __and__[U](self, other: Expr[U]) -> Expr[T | U]:
        return And[T | U](self, other)

    def __or__[U](self, other: Expr[U]) -> Expr[T | U]:
        return Or[T | U](self, other)

    def __invert__(self) -> Expr[T]:
        return Not(self)

    def __bool__(self) -> bool:
        raise TypeError("use &, |, and ~ to combine conditions, not and, or, and not")

Table

Table[T] stores rows of one declared type. insert() returns an internal row ID for retrieving the stored row, separate from any declared id field.

users = Table(User)
row_id = users.insert(User(name="Ada"))
users.get(row_id)  # User(name="Ada")

Use scan() to start a query rather than iterating over the table itself.

Source

class Table[T: Row]:
    def __init__(self, model: type[T]) -> None:
        if not issubclass(model, Row) or model is Row:
            raise TypeError("expected a model inheriting from Row")
        if any(spec.unique for spec in model.fields()):
            raise NotImplementedError("unique fields need indexes, which you build in chapter 6")
        self.model = model
        self._rows: dict[RowId, T] = {}
        self._next_id = 1

    def scan(self) -> Query[T]:
        raise NotImplementedError('Exercise: Table.scan')

    def __len__(self) -> int:
        raise NotImplementedError('Exercise: Table.__len__')

    def insert(self, row: T) -> RowId:
        raise NotImplementedError('Exercise: Table.insert')

    def get(self, row_id: RowId) -> T:
        raise NotImplementedError('Exercise: Table.get')

    def update(self, row_id: RowId, /, **changes: object) -> T:
        raise NotImplementedError('Exercise: Table.update')

    def delete(self, row_id: RowId) -> None:
        raise NotImplementedError('Exercise: Table.delete')

    def update_where(self, condition: Expr[T], /, **changes: object) -> int:
        raise NotImplementedError('Exercise: Table.update_where')

    def delete_where(self, condition: Expr[T], /) -> int:
        raise NotImplementedError('Exercise: Table.delete_where')

    def lookup[K](self, field: UniqueField[T, K], key: K, /) -> T | None:
        if field.model is not self.model:
            raise TypeError(f"{field.name} is not a field of {self.model.__name__}")
        field.check(key)
        raise NotImplementedError("index lookup is the chapter 6 exercise; use scan() for now")

Database

Database keeps one table for each row class. Repeated calls to table() with the same class return the same table.

db = Database()
users = db.table(User)

Source

class Database:
    def __init__(self) -> None:
        self._tables: dict[type, Table[Any]] = {}

    def table[T: Row](self, cls: type[T]) -> Table[T]:
        if cls not in self._tables:
            self._tables[cls] = Table(cls)
        return cast("Table[T]", self._tables[cls])

Query

Query[T] describes how to produce rows of type T. Building a query reads nothing. Iterating it, or calling all(), runs it.

adults = users.scan().where(User.age >= 18).select(User.name)
adults.all()  # ["Ada", ...]

Source

class Query[T](Iterable[T]):
    @overload
    def where[L, R](
        self: Query[tuple[L, R]], condition: Expr[L] | Expr[R] | Expr[L | R], /
    ) -> Query[tuple[L, R]]: ...

    @overload
    def where(self, condition: Expr[T], /) -> Query[T]: ...

    def where(self, condition: Expr[Any], /) -> Query[Any]:
        return Where(self, condition)

    @overload
    def select[L, R, A](self: Query[tuple[L, R]], a: Field[L, A] | Field[R, A], /) -> Query[A]: ...

    @overload
    def select[L, R, A, B](
        self: Query[tuple[L, R]], a: Field[L, A] | Field[R, A], b: Field[L, B] | Field[R, B], /
    ) -> Query[tuple[A, B]]: ...

    @overload
    def select[A](self, a: Field[T, A], /) -> Query[A]: ...

    @overload
    def select[A, B](self, a: Field[T, A], b: Field[T, B], /) -> Query[tuple[A, B]]: ...

    @overload
    def select[A, B, C](
        self, a: Field[T, A], b: Field[T, B], c: Field[T, C], /
    ) -> Query[tuple[A, B, C]]: ...

    def select(self, *fields: Field[Any, Any]) -> Query[Any]:
        if not fields:
            raise TypeError("select() needs at least one field")
        return Select(self, fields)

    def join[U](self, other: Query[U], on: JoinOn[T, U], /) -> Query[tuple[T, U]]:
        return HashJoin(self, other, on)

    @overload
    def order_by[L, R](
        self: Query[tuple[L, R]], *keys: SortKey[L] | SortKey[R]
    ) -> Query[tuple[L, R]]: ...

    @overload
    def order_by(self, *keys: SortKey[T]) -> Query[T]: ...

    def order_by(self, *keys: SortKey[Any]) -> Query[Any]:
        return OrderBy(self, keys)

    @overload
    def group_by[L, R, K](
        self: Query[tuple[L, R]], key: Field[L, K] | Field[R, K], /
    ) -> Query[tuple[K]]: ...

    @overload
    def group_by[L, R, K, A](
        self: Query[tuple[L, R]],
        key: Field[L, K] | Field[R, K],
        a: Aggregate[L, A] | Aggregate[R, A],
        /,
    ) -> Query[tuple[K, A]]: ...

    @overload
    def group_by[L, R, K, A, B](
        self: Query[tuple[L, R]],
        key: Field[L, K] | Field[R, K],
        a: Aggregate[L, A] | Aggregate[R, A],
        b: Aggregate[L, B] | Aggregate[R, B],
        /,
    ) -> Query[tuple[K, A, B]]: ...

    @overload
    def group_by[K](self, key: Field[T, K], /) -> Query[tuple[K]]: ...

    @overload
    def group_by[K, A](self, key: Field[T, K], a: Aggregate[T, A], /) -> Query[tuple[K, A]]: ...

    @overload
    def group_by[K, A, B](
        self, key: Field[T, K], a: Aggregate[T, A], b: Aggregate[T, B], /
    ) -> Query[tuple[K, A, B]]: ...

    def group_by(self, key: Field[Any, Any], *aggregates: Aggregate[Any, Any]) -> Query[Any]:
        return GroupBy(self, key, aggregates)

    def limit(self, n: int, offset: int = 0) -> Query[T]:
        return Limit(self, n, offset)

    def all(self) -> list[T]:
        return list(self)

    def first(self) -> T | None:
        for row in self:
            return row
        return None

    def count(self) -> int:
        return sum(1 for _ in self)

Count

The number of rows in the group.

Source

@dataclass(frozen=True, slots=True)
class Count(Aggregate[Any, int]):
    def start(self) -> Value:
        return 0

    def add(self, total: Value, row: object) -> Value:
        return total + 1

class Aggregate[T, V](ABC):
    @abstractmethod
    def start(self) -> Value: ...

    @abstractmethod
    def add(self, total: Value, row: object) -> Value: ...

Sum

The sum of a numeric field over the group. None values are skipped, and a group with no other values sums to 0. SQL would return NULL for that group.

Source

@dataclass(frozen=True, slots=True)
class Sum[T, N: (int, float, Decimal)](Aggregate[T, N]):
    field: Field[T, N] | Field[T, N | None]

    def start(self) -> Value:
        return 0

    def add(self, total: Value, row: object) -> Value:
        value = self.field.read(row)
        return total if value is None else total + value

class Aggregate[T, V](ABC):
    @abstractmethod
    def start(self) -> Value: ...

    @abstractmethod
    def add(self, total: Value, row: object) -> Value: ...

lookup

table.lookup(Model.field, key) finds the row whose unique field equals key, using an index rather than a scan. It returns None when no row matches.

# After building indexes in chapter 6:
contacts.lookup(Contact.email, "[email protected]")  # Contact | None

Source

def lookup[K](self, field: UniqueField[T, K], key: K, /) -> T | None:
    if field.model is not self.model:
        raise TypeError(f"{field.name} is not a field of {self.model.__name__}")
    field.check(key)
    raise NotImplementedError("index lookup is the chapter 6 exercise; use scan() for now")

scan

Return a lazy Query over the stored rows in insertion order.

Each execution reads current rows, including updates and duplicates. An empty table yields nothing. This is a live view, not a snapshot. Changing the table's size during iteration raises RuntimeError.

Source

def scan(self) -> Query[T]:
    raise NotImplementedError('Exercise: Table.scan')