loom.core.model

class loom.core.model.BaseModel

Bases: LoomStruct

Base for all loom domain models.

Subclasses must declare __tablename__ and provide typed attributes. Column metadata is optional via ColumnField(...).

class loom.core.model.Privilege(value)

Bases: StrEnum

Row operations subject to row-level security.

TRUNCATE, REFERENCES and TRIGGER are deliberately absent: none of them honours row-level security, so none of them can ever be granted to a database user that is not meant to bypass it.

class loom.core.model.RowScoped

Bases: object

Mark a model as row-scoped; adds no column.

The product names the scope on one of the model’s columns with ColumnField(scope=...). A read-only scoped table narrows __scope_privileges__ to {Privilege.SELECT}.

class loom.core.model.ScopeColumn(scope, column, on, elevable)

Bases: object

One scoped column: which scope, which column, and how far it reaches.

Parameters:

• scope (str)

• column (str)

• on (Literal['read', 'write', 'both'])

• elevable (bool)

class loom.core.model.ScopedTable(schema, name, scopes, privileges, partition_by=None)

Bases: object

A compiled scoped table: its scopes, the privileges it grants and its partition column.

partition_by names the column a range-partitioned table is partitioned by, None for a plain table.

Parameters:

• schema (str | None)

• name (str)

• scopes (tuple[ScopeColumn, ...])

• privileges (frozenset[Privilege])

• partition_by (str | None)

class loom.core.model.PartialUnique(rule, columns, where)

Bases: object

A unique index over columns restricted to the rows matching where.

Parameters:

• rule (str)

• columns (tuple[str, ...])

• where (str)

class loom.core.model.PartitionRange(name, lower, upper)

Bases: object

One partition: its name and the half-open range [lower, upper) it holds.

Parameters:

• name (str)

• lower (str)

• upper (str)

class loom.core.model.RangePartition(column)

Bases: object

A table partitioned by range over column.

Parameters:

column (str)

loom.core.model.declared_checks(cls)

Named CHECK constraints declared with __checks__ as {rule: sql_expression}.

The rule is the constraint name, or its %(constraint_name)s under a naming convention; the expression is SQL that loom passes through verbatim.

Parameters:

cls (type)

Return type:

Mapping[str, str]

loom.core.model.declared_indexes(cls)

Non-unique indexes declared with __indexes__.

Parameters:

cls (type)

Return type:

tuple[tuple[str, …], …]

loom.core.model.declared_partial_unique(cls)

Partial unique indexes declared with __partial_unique__.

Each entry reads {rule: (columns, where)}: columns is a tuple of column names, where the SQL predicate passed through verbatim.

Parameters:

cls (type)

Return type:

tuple[PartialUnique, …]

loom.core.model.declared_partition(cls)

The range partitioning declared with __partition_by__ = ("RANGE", "<column>").

The column must be a DateTime column of the primary key, and every unique key must contain it, as Postgres requires of a partitioned table. Only row-scoped models may be partitioned, so the guard protects every partition; LIST and HASH partitioning are not supported.

Raises:

ValueError – Naming the model and what is wrong with the declaration.

Parameters:

cls (type)

Return type:

RangePartition | None

loom.core.model.declared_privileges(cls)

Group privileges declared with __privileges__ on an unscoped model.

Parameters:

cls (type)

Return type:

Mapping[str, frozenset[Privilege]]

loom.core.model.declared_unique(cls)

Composite UNIQUE constraints declared with __unique__.

Parameters:

cls (type)

Return type:

tuple[tuple[str, …], …]

loom.core.model.is_row_scoped(cls)

Whether cls carries the RowScoped marker.

Parameters:

cls (type)

Return type:

bool

loom.core.model.range_partitions(table, start, end, interval='month')

The partitions of table covering [start, end), from the period holding start.

A datetime bound is read as its date.

Raises:

ValueError – When interval is unknown or a partition name would be longer than Postgres keeps.

Parameters:

• table (str)

• start (date)

• end (date)

• interval (Literal['day', 'month', 'year'])

Return type:

tuple[PartitionRange, …]

loom.core.model.scope_columns(cls)

Resolve the scoped columns of cls, enforcing rules C1 to C5 and C8.

Raises ValueError naming the rule, the model and the column.

Parameters:

cls (type)

Return type:

tuple[ScopeColumn, …]

exception loom.core.model.BoundaryValidationError(message, violations=())

Bases: LoomError, ValueError

A boundary value failed validation.

Raised by every LoomType implementation with the library error as __cause__. The pydantic message is built field by field and omits the input and any documentation URL; the msgspec message is forwarded verbatim, and is one line and free of the input already. It is also a ValueError, so an except ValueError that used to catch msgspec.ValidationError or pydantic.ValidationError keeps catching it.

Parameters:

• message (str) – One-line description of the failure.

• violations (tuple[tuple[str, str], ...]) – (field, message) pairs, one per failed field. Empty when the caller has no per-field breakdown to offer.

Return type:

None

class loom.core.model.LoomFrozenStruct

Bases: Struct

Base immutable struct for Loom value contracts.

Use this for configuration, DSL declarations, wire envelopes, and other value objects that should not expose mutable assignment after creation.

classmethod loom_message_type()

Return the stable logical message type for this model class.

Return type:

str

class loom.core.model.LoomStruct

Bases: Struct

Base struct for Loom logical data types.

LoomStruct is intentionally neutral. Specialised framework types such as responses or persistence models define their own serialisation behaviour.

classmethod loom_message_type()

Return the stable logical message type for this model class.

Return type:

str

class loom.core.model.TimestampedModel(*, created_at=UNSET, updated_at=UNSET)

Bases: BaseModel

Opt-in base for domain models that need audit timestamps.

Adds created_at and updated_at as nullable server-default columns. Both fields are omitted from serialisation output when None.

Design note: trace_id is intentionally absent. Trace identifiers are an observability concern belonging to the transport layer, not to the domain model.

Parameters:

• created_at (datetime | None)

• updated_at (datetime | None)

created_at

UTC timestamp set by the database at INSERT time. None until the first flush/commit.

Type:

datetime.datetime | None

updated_at

UTC timestamp updated by the database on every UPDATE. None until the first flush/commit.

Type:

datetime.datetime | None

class loom.core.model.Cardinality(value)

Bases: StrEnum

class loom.core.model.ColumnFieldInfo(name, python_type, column_type, field)

Bases: object

Resolved metadata for a single column field.

Parameters:

• name (str)

• python_type (type)

• column_type (ColumnType)

• field (Field)

loom.core.model.ColumnField(column_type=None, *, primary_key=False, unique=False, index=False, nullable=False, autoincrement=False, server_default=None, server_onupdate=None, foreign_key=None, on_delete=None, default=msgspec.UNSET, length=None, db_type=None)

Declare column metadata without using explicit Annotated syntax.

Parameters:

• column_type (ColumnType | None)

• primary_key (bool)

• unique (bool)

• index (bool)

• nullable (bool)

• autoincrement (bool)

• server_default (ServerDefault | None)

• server_onupdate (ServerOnUpdate | str | None)

• foreign_key (str | None)

• on_delete (OnDelete | None)

• default (Any)

• length (int | None)

• db_type (ColumnType | None)

Return type:

Any

loom.core.model.ScopedField(column_type=None, *, scope, on='both', elevable=False, primary_key=False, nullable=False, foreign_key=None, on_delete=None, length=None, db_type=None)

Declare the column that scopes the rows of a RowScoped model.

scope names the session value the column is compared with; on limits the comparison to reads or writes, and elevable lets an authorized execution lift a write-only scope within the table’s boundary.

Parameters:

• column_type (ColumnType | None)

• scope (str)

• on (Literal['read', 'write', 'both'])

• elevable (bool)

• primary_key (bool)

• nullable (bool)

• foreign_key (str | None)

• on_delete (OnDelete | None)

• length (int | None)

• db_type (ColumnType | None)

Return type:

Any

class loom.core.model.ColumnType(type_name, args=(), kwargs=<factory>)

Bases: object

Backend column type descriptor stored inside Annotated[...].

Parameters:

• type_name (str)

• args (tuple[Any, ...])

• kwargs (dict[str, Any])

class loom.core.model.Field(primary_key=False, unique=False, index=False, nullable=False, autoincrement=False, server_default=None, server_onupdate=None, foreign_key=None, on_delete=None, default=UNSET, length=None, scope=None, on='both', elevable=False)

Bases: object

Column constraint metadata stored inside Annotated[...].

Parameters:

• primary_key (bool)

• unique (bool)

• index (bool)

• nullable (bool)

• autoincrement (bool)

• server_default (ServerDefault | None)

• server_onupdate (ServerOnUpdate | str | None)

• foreign_key (str | None)

• on_delete (OnDelete | None)

• default (Any)

• length (int | None)

• scope (str | None)

• on (Literal['read', 'write', 'both'])

• elevable (bool)

loom.core.model.ProjectionField(*, loader, profiles=('default',), depends_on=(), default=PROJECTION_DEFAULT_MISSING)

Declare a projection field without triggering assignment type errors.

Parameters:

• loader (Any) – Loader descriptor or instance responsible for computing the value.

• profiles (tuple[str, ...]) – Profile names in which this projection is active.

• depends_on (tuple[str, ...]) – Names of other projections or relation events this depends on.

• default (Any) – Fallback value when the loader returns no result for an entity.

Returns:

A Projection instance cast to Any for clean class-body assignment.

Return type:

Any

Example:

count_reviews: int = ProjectionField(
    loader=CountLoader(model=ProductReview),
    profiles=("with_details",),
    default=0,
)

class loom.core.model.LoomStructMeta(name, bases, namespace, **kwargs)

Bases: StructMeta

Metaclass that intercepts Relation and Projection assignments before StructMeta processes the class body.

Each intercepted attribute is replaced with UNSET as its default, and the type annotation is widened to T | UnsetType so that omit_defaults=True strips unloaded fields from serialisation output.

class loom.core.model.LoomType(*args, **kwargs)

Bases: Protocol

Boundary contract a compiled type exposes to its consumers.

type

The declared type or a compiler-generated annotation, kept for introspection.

Type:

Any

library

Closed tag naming the implementation behind the value.

Type:

Literal[‘msgspec’, ‘pydantic’]

schema()

Return the JSON Schema of the wrapped type.

Return type:

Mapping[str, Any]

decode_json(data, /)

Validate data strictly and build the typed value in one pass.

Raises:

BoundaryValidationError – data is malformed JSON or does not satisfy the type.

Parameters:

data (bytes | str)

Return type:

Any

to_builtins(obj, /)

Turn a typed value into JSON-mode builtins.

Parameters:

obj (Any)

Return type:

Any

from_builtins(payload, /)

Validate builtins already produced by a trusted source and build the typed value.

Unlike decode_json() this takes Python builtins rather than JSON text, and pydantic validates them in its lax mode: it is the entry point for a value that already round-tripped through a cache backend or an MCP client rather than through raw JSON.

Raises:

BoundaryValidationError – payload does not satisfy the type.

Parameters:

payload (Any)

Return type:

Any

class loom.core.model.OnDelete(value)

Bases: StrEnum

class loom.core.model.OnUpdate(value)

Bases: StrEnum

class loom.core.model.Projection(loader, profiles=('default', ), depends_on=(), default=<object object>)

Bases: object

Derived-field metadata assigned as a class attribute on a BaseModel.

Parameters:

• loader (Any)

• profiles (tuple[str, ...])

• depends_on (tuple[str, ...])

• default (Any)

class loom.core.model.Relation(foreign_key, cardinality, secondary=None, back_populates=None, on_delete=None, on_update=None, profiles=('default',), depends_on=(), default=None)

Bases: object

Relationship metadata assigned as a class attribute on a BaseModel.

Parameters:

• foreign_key (str)

• cardinality (Cardinality)

• secondary (str | None)

• back_populates (str | None)

• on_delete (OnDelete | None)

• on_update (OnUpdate | None)

• profiles (tuple[str, ...])

• depends_on (tuple[str, ...])

• default (Any)

loom.core.model.RelationField(*, foreign_key, cardinality, secondary=None, back_populates=None, on_delete=None, on_update=None, profiles=('default',), depends_on=(), default=None)

Declare a relation field with normal typing (without assignment type errors).

Parameters:

• foreign_key (str)

• cardinality (Cardinality)

• secondary (str | None)

• back_populates (str | None)

• on_delete (OnDelete | None)

• on_update (OnUpdate | None)

• profiles (tuple[str, ...])

• depends_on (tuple[str, ...])

• default (Any)

Return type:

Any

class loom.core.model.ServerDefault(value)

Bases: StrEnum

class loom.core.model.ServerOnUpdate(value)

Bases: StrEnum

classmethod is_now(value)

Return whether value requests the server clock, as enum or string.

Parameters:

value (ServerOnUpdate | str | None) – A server_onupdate declaration; the string form is matched case-insensitively.

Returns:

True for ServerOnUpdate.NOW or "now" in any case.

Return type:

bool

exception loom.core.model.UnsupportedBoundaryType(reason)

Bases: LoomError

A symbol cannot be compiled into a LoomType.

Parameters:

reason (str) – Why the symbol is rejected, shared by every boundary’s issue.

Return type:

None

loom.core.model.get_column_fields(cls)

Extract column fields from a model class.

Parameters:

cls (type)

Return type:

dict[str, ColumnFieldInfo]

loom.core.model.get_id_attribute(cls)

Return the name of the first primary-key field.

Parameters:

cls (type)

Return type:

str

loom.core.model.get_id_attributes(cls)

Return the names of the primary-key fields, in declaration order.

Raises:

ValueError – If the model declares no primary-key field.

Parameters:

cls (type)

Return type:

tuple[str, …]

loom.core.model.get_projections(cls)

Return projections registered by LoomStructMeta.

Parameters:

cls (type)

Return type:

dict[str, Projection]

loom.core.model.get_relations(cls)

Return relations registered by LoomStructMeta.

Parameters:

cls (type)

Return type:

dict[str, Relation]

loom.core.model.get_table_name(cls)

Return the __tablename__ declared on the model.

Parameters:

cls (type)

Return type:

str

loom.core.model.is_pydantic_model(symbol)

Report whether symbol is a pydantic.BaseModel subclass.

Total over any symbol: a non-class value answers False rather than raising, so a caller can probe an annotation argument without first checking it is a class.

Parameters:

symbol (Any) – Value to test.

Returns:

True when symbol is a pydantic.BaseModel subclass and pydantic is already imported.

Return type:

bool

loom.core.model.loom_type(symbol)

Compile an authored class into a LoomType.

Strictness is a compile-time property of an authored symbol only: a msgspec.Struct must declare forbid_unknown_fields=True and a pydantic.BaseModel must resolve model_config["extra"] == "forbid". The check happens once, here, before the symbol is handed to loom_type_of(); a compiler-generated annotation built from an authored symbol never repeats it.

Parameters:

symbol (type) – A strict msgspec.Struct or strict pydantic.BaseModel subclass.

Returns:

The boundary type wrapping symbol.

Raises:

UnsupportedBoundaryType – symbol is neither library’s model class or does not reject unknown fields.

Return type:

LoomType

loom.core.model.loom_type_of(annotation)

Compile any annotation, authored or compiler-generated, into a LoomType.

The single dispatcher every boundary uses once loom_type has already checked the strictness of an authored root: a msgspec.Struct class resolves to msgspec_type(), a pydantic.BaseModel class or an annotation that mentions one anywhere inside its typing.get_args walk resolves to pydantic_type(), an annotation that mentions both libraries is rejected, and anything else resolves to msgspec_type().

The result is memoised by hash/equality of the annotation (typing generics compare structurally); the memo lives for the process, because it is a pure factory: the same annotation always builds the same kind of LoomType, and building one is not free — it compiles a decoder or a TypeAdapter. An annotation that is not hashable (for example Annotated metadata carrying a dict) simply skips the cache.

Parameters:

annotation (Any) – Any annotation, authored or compiler-generated.

Returns:

The boundary type wrapping annotation.

Raises:

UnsupportedBoundaryType – annotation mentions both a msgspec.Struct and a pydantic.BaseModel.

Return type:

LoomType

loom.core.model.msgspec_type(annotation)

Wrap a compiler-generated annotation in a msgspec-backed LoomType.

Accepts whatever msgspec.json.Decoder accepts: a defstruct result, dict[str, Any], list[...] or a scalar. Strictness is not checked because the annotation is not authored by a user.

Parameters:

annotation (Any) – Any annotation msgspec.json.Decoder can build.

Returns:

The boundary type wrapping annotation.

Raises:

UnsupportedBoundaryType – msgspec.json.Decoder cannot build annotation.

Return type:

LoomType

loom.core.model.pydantic_type(annotation)

Wrap any annotation pydantic can describe in a pydantic-backed LoomType.

Covers a BaseModel, a RootModel, a parameterised generic model, a pydantic.dataclasses type and list/tuple/union of those — the same TypeAdapter-backed implementation the authored BaseModel case in loom_type() uses. Strictness is not checked because the annotation is not necessarily an authored root symbol.

Parameters:

annotation (Any) – Any annotation pydantic.TypeAdapter can build.

Returns:

The boundary type wrapping annotation.

Raises:

UnsupportedBoundaryType – pydantic.TypeAdapter cannot build annotation.

Return type:

LoomType