Package

models

viur-models — SQLModel-backed models for ViUR.

Model/Record (base classes), Field (bone metadata), the bone-typed field types, structure_for_model, and the boot wiring install/setup. Importing has no side effects. Bone mapping tables: docs/bones.md.

VIUR_META_KEY module-attribute

VIUR_META_KEY = 'viur'

Key of the bone metadata in FieldInfo.json_schema_extra.

BONE_TYPE_REGISTRY module-attribute

BONE_TYPE_REGISTRY: dict[type, BoneType] = {}

Python type → BoneType, matched along the MRO; an Annotated marker wins.

Code module-attribute

Code = t.Annotated[str, BoneType('raw.code', replace=True, extras={'indexed': False})]

CodeBone ("raw.code"), not indexed.

Color module-attribute

Color = t.Annotated[str, BoneType('color', replace=True, extras=dict(_COLOR_EXTRAS))]

ColorBone ("color", mode: rgb).

Country module-attribute

Country = CountryAlpha2

SelectCountryBone ("select.country"), validated by pydantic-extra-types.

Credential module-attribute

Credential = t.Annotated[str, BoneType('str.credential', extras={'maxlength': None, 'minlength': None}, write_only=True)]

CredentialBone ("str.credential"), write-only.

Email module-attribute

Email = t.Annotated[str, BoneType('str.email')]

EmailBone ("str.email").

Json module-attribute

Json = t.Annotated[dict, BoneType('raw.json', replace=True, extras={'schema': {}, 'indexed': False})]

JsonBone ("raw.json"); needs an explicit sa_type.

Password module-attribute

Password = t.Annotated[str, BoneType('password', replace=True, emptyvalue='', write_only=True, extras={'maxlength': 254, 'minlength': None, 'tests': PASSWORD_TESTS, 'test_threshold': 4})]

PasswordBone ("password"), write-only. Hashing is NOT automatic — do it in onAdd/onEdit.

Phone module-attribute

Phone = t.Annotated[str, BoneType('str.phone', extras=dict(_PHONE_EXTRAS))]

PhoneBone ("str.phone"); max_length=15 for parity. Real validation: pydantic PhoneNumber.

Raw module-attribute

Raw = t.Annotated[str, BoneType('raw', replace=True)]

RawBone ("raw").

SortIndex module-attribute

SortIndex = t.Annotated[float, BoneType('numeric.sortindex', extras={'clone_behavior': {'strategy': 'set_default'}})]

SortIndexBone ("numeric.sortindex").

Text module-attribute

Text = t.Annotated[str, BoneType('text', extras={'valid_html': None}, replace=True, emptyvalue='')]

TextBone ("text").

Uid module-attribute

Uid = t.Annotated[str, BoneType('uid', replace=True, extras={'fillchar': '*', 'length': 13, 'pattern': '*', 'readonly': True, 'unique': 1, 'clone_behavior': {'strategy': 'set_default'}, 'compute': {'method': 'Once'}})]

UidBone ("uid"): readonly, unique. The module supplies the value.

Uri module-attribute

Uri = t.Annotated[str, BoneType('uri', replace=True, extras=dict(_URI_EXTRAS))]

UriBone ("uri"), default hints. Real validation: pydantic AnyUrl with sa_type=String.

Model

Bases: SQLModel

Base for SQL-backed models. The structure is built at class definition (unmappable types fail fast) and cached per class.

Source code in src/viur/models/base.py

class Model(SQLModel):
    """Base for SQL-backed models. The structure is built at class definition (unmappable types
    fail fast) and cached per class."""

    #: Fields carried in a referencing model's ``relskel``/``dest`` (``refKeys``); ``key`` always included.
    viur_ref_keys: t.ClassVar[tuple[str, ...]] = ("name",)

    #: Bone-parameter overrides per relation name; the only source for many-to-many relations.
    viur_relation_meta: t.ClassVar[dict[str, dict]] = {}

    #: Bones always included when a client bonelist restricts a response (``"*"``-subskel
    #: analogue); ``key`` always is.
    viur_bones_always: t.ClassVar[tuple[str, ...]] = ()

    #: Engine name (``viur.models.db.configure(..., name=)``); relation targets and link
    #: tables must share it.
    viur_database: t.ClassVar[str] = "default"

    id: int | None = SQLModelField(default=None, primary_key=True)
    creationdate: datetime | None = Field(
        default_factory=_utcnow, readonly=True, visible=False, tags=("technical",),
        descr="created at", compute={"method": "Once"},
    )
    changedate: datetime | None = Field(
        default_factory=_utcnow, readonly=True, visible=False, tags=("technical",),
        descr="updated at", compute={"method": "OnWrite"},
    )

    @property
    def errors(self) -> list:
        """Client-input errors (``SkeletonInstance.errors`` counterpart). Kept in ``__dict__``:
        loaded rows skip ``__init__``, so pydantic private attrs may not exist."""
        return self.__dict__.get("_viur_errors", [])

    @errors.setter
    def errors(self, value: t.Iterable) -> None:
        self.__dict__["_viur_errors"] = list(value)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: t.Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        # fail fast; deferred for relationships (mapper not configured yet) and
        # cross-store references (skeleton registry fills at boot)
        if crossstore := crossstore_fields(cls):
            _crossstore._register_model(cls)
        if not getattr(cls, "__sqlmodel_relationships__", None) and not crossstore:
            cls._viur_structure_shared()

    @classmethod
    def _viur_structure_shared(cls) -> dict:
        """Cached structure, shared by every request — never mutate; for read-only hot paths."""
        if "_viur_structure" not in cls.__dict__:
            cls._viur_structure = structure_for_model(cls)
        return cls._viur_structure

    @classmethod
    def viur_structure(cls) -> dict:
        """Skeleton-compatible structure dict, fresh deep copy (the caller owns it)."""
        return copy.deepcopy(cls._viur_structure_shared())

    @classmethod
    def viur_crossstore(cls) -> dict:
        """``name → SkeletonRefMarker``, cached."""
        if "_viur_crossstore" not in cls.__dict__:
            cls._viur_crossstore = crossstore_fields(cls)
        return cls._viur_crossstore

    @classmethod
    def viur_relations(cls) -> dict:
        """Relations per ``relations_for_model``, cached."""
        if "_viur_relations" not in cls.__dict__:
            relations = relations_for_model(cls)
            _check_relation_databases(cls, relations)
            cls._viur_relations = relations
        return cls._viur_relations

    @classmethod
    def viur_write_only(cls) -> frozenset:
        """Write-only field names, cached."""
        if "_viur_write_only" not in cls.__dict__:
            cls._viur_write_only = write_only_fields(cls)
        return cls._viur_write_only

    def viur_dump(self, *, bones: t.Iterable[str] = ()) -> dict:
        """``SkeletonInstance.dump()``-shaped values: opaque ``key``, ISO datetimes, enum values,
        relations in ``RelationalBone`` shape. ``bones`` restricts the output; without it, a
        client bonelist attached by ``SQLList`` (``_viur_bones``) does."""
        bones = set(bones) if bones else self.__dict__.get("_viur_bones", ())
        structure = self._viur_structure_shared()
        relations = self.viur_relations()
        crossstore = self.viur_crossstore()
        write_only = self.viur_write_only()
        out = {}
        for name in structure:
            if bones and name not in bones:
                continue
            if name == "key":
                out[name] = self.viur_key
            elif name in crossstore:
                out[name] = self._viur_dump_crossstore(name, crossstore[name])
            elif name in relations:
                out[name] = self._viur_dump_relation(name, relations[name])
            elif name in write_only:
                out[name] = structure[name]["emptyvalue"]
            elif languages := structure[name]["languages"]:
                value = getattr(self, name)
                if not isinstance(value, dict):  # unvalidated form
                    value = {}
                out[name] = {lang: value.get(lang) for lang in languages}
            elif self.__dict__.get("_viur_unvalidated") \
                    and name in type(self).model_computed_fields:
                # computed fields over raw values may raise on a rejected form
                try:
                    out[name] = _dump_value(getattr(self, name))
                except Exception:
                    out[name] = structure[name]["emptyvalue"]
            else:
                out[name] = _dump_value(getattr(self, name))
        return out

    @staticmethod
    def _viur_dest(related: "Model") -> dict:
        dest = {"key": related.viur_key}
        for ref in type(related).viur_ref_keys:
            if ref in type(related).model_fields \
                    or ref in type(related).model_computed_fields:
                dest[ref] = _dump_value(getattr(related, ref))
        return dest

    def _viur_dump_crossstore(self, name: str, marker: t.Any) -> dict | list | None:
        """RelationalBone value shape from the stored ``dest`` snapshot."""
        value = getattr(self, name)
        if marker.multiple:
            return [{"dest": dest, "rel": None} for dest in (value or [])]
        return {"dest": value, "rel": None} if value else None

    def _viur_dump_relation(self, rel_name: str, info: dict) -> dict | list | None:
        """``RelationalBone`` value shape. Related objects are read from ``__dict__`` only (no lazy
        IO on detached instances); unloaded to-one → key-only ``dest`` from the FK, unloaded
        many-to-many → parked client input."""
        using_fields = info.get("using_fields")
        pending = self.__dict__.get("_viur_pending_relations", {}).get(rel_name)

        def _rel_payload(row: t.Any) -> dict | None:
            if not using_fields:
                return None
            return {name: _dump_value(getattr(row, name)) for name in using_fields}

        if info.get("crossstore"):
            rows = self.__dict__.get(rel_name) or pending or []
            return [{"dest": row.dest, "rel": _rel_payload(row)} for row in rows]
        if info.get("link"):
            out = []
            for row in self.__dict__.get(rel_name) or pending or []:
                related = row.__dict__.get(info["dest_rel"])
                if related is not None:
                    dest = self._viur_dest(related)
                elif (foreign_key := getattr(row, info["dest_fk"])) is not None:
                    dest = {"key": info["target"].viur_encode_key(foreign_key)}
                else:
                    dest = None
                out.append({"dest": dest, "rel": _rel_payload(row)})
            return out
        if info["multiple"]:
            related_list = self.__dict__.get(rel_name)
            if related_list is None and pending is not None:
                return [  # parked client input (rejected re-render)
                    {"dest": {"key": info["target"].viur_encode_key(pk)}, "rel": None}
                    for pk in pending
                ]
            return [
                {"dest": self._viur_dest(related), "rel": None}
                for related in (related_list or [])
            ]
        related = self.__dict__.get(rel_name)
        if related is not None:
            return {"dest": self._viur_dest(related), "rel": None}
        foreign_key = getattr(self, info["fk"])
        if foreign_key is None:
            return None
        return {"dest": {"key": info["target"].viur_encode_key(foreign_key)}, "rel": None}

    @classmethod
    def viur_from_client(
        cls, data: dict,
    ) -> tuple[t.Self | None, list["ReadFromClientError"]]:
        """``skel.fromClient()`` counterpart. Unknown and read-only fields are dropped. Returns
        ``(instance, [])`` or ``(form, errors)`` — ``form`` is unvalidated (``_viur_form``),
        never persist it."""
        structure = cls._viur_structure_shared()
        cleaned = {
            name: value
            for name, value in data.items()
            if name in structure and not structure[name]["readonly"]
        }

        # dotted sub-keys: title.de=…, pos.lat=…, address.street=…, name.<idx>.<field>=…
        dotted: dict[str, dict] = {}
        for key, value in data.items():
            base_name, separator, suffix = key.partition(".")
            if separator and base_name in structure and not structure[base_name]["readonly"]:
                dotted.setdefault(base_name, {})[suffix] = value
        for base_name, subs in dotted.items():
            if structure[base_name]["languages"]:
                cleaned[base_name] = _merge_sub_keys(cleaned.get(base_name), subs)
            elif structure[base_name]["type"] == "spatial":
                cleaned[base_name] = [subs.get("lat"), subs.get("lng")]
            elif structure[base_name]["type"] == "record" \
                    and not structure[base_name]["multiple"]:
                cleaned[base_name] = _merge_sub_keys(cleaned.get(base_name), subs)
            elif structure[base_name].get("multiple") \
                    and subs and all(key.partition(".")[0].isdigit() for key in subs):
                items: dict[int, dict] = {}
                for key, value in subs.items():
                    index, _, field = key.partition(".")
                    items.setdefault(int(index), {})[field] = value
                ordered = [items[index] for index in sorted(items)]
                if structure[base_name]["type"] == "record":
                    cleaned[base_name] = ordered
                else:
                    cleaned[base_name] = [
                        {"dest": {"key": item.pop("key", None)}, "rel": item}
                        for item in ordered
                    ]

        # Language dicts keep only declared languages.
        for name, bone in structure.items():
            if (languages := bone["languages"]) and isinstance(cleaned.get(name), dict):
                cleaned[name] = {
                    lang: value
                    for lang, value in cleaned[name].items()
                    if lang in languages and value is not None
                }

        _normalize_empty_values(structure, cleaned)

        # empty write-only input keeps the stored value
        for name in cls.viur_write_only():
            if name in cleaned and cleaned[name] in ("", None):
                del cleaned[name]

        relation_errors = []

        # cross-store: read_dest is the existence check; a full stale snapshot survives a
        # vanished target, a bare unknown key rejects
        for name, marker in cls.viur_crossstore().items():
            if name not in cleaned:
                continue
            raw = cleaned.pop(name)
            values = raw if isinstance(raw, (list, tuple)) else [raw]
            dests = []
            for item in values:
                item, stale_snapshot = _split_dest_input(item)
                if item in (None, "", "None"):
                    continue
                dest = _crossstore.read_dest(marker, str(item))
                if dest is not None:
                    dests.append(dest)
                elif stale_snapshot is not None:
                    dests.append(stale_snapshot)
                else:
                    relation_errors.append(relation_error(name, "Unknown key"))
            if marker.multiple:
                cleaned[name] = dests
            else:
                cleaned[name] = dests[0] if dests else None

        # to-one → FK column; many-to-many parked in _viur_pending_relations for SQLList
        # (existence is checked there, only the key format here)
        pending_relations = {}
        for rel_name, info in cls.viur_relations().items():
            if rel_name not in cleaned:
                continue
            raw = cleaned.pop(rel_name)
            if info["multiple"]:
                values = raw if isinstance(raw, (list, tuple)) else [raw]
                using_fields = info.get("using_fields") or ()
                primary_keys = []

                def _validated_link(link_cls: type, data: dict) -> t.Any:
                    """Validate one link row; errors get the ``[name, "rel", field]`` path. Built from
                    the submitted keys only — ``model_validate`` would pin the parent FK to ``None``."""
                    try:
                        validated = link_cls.model_validate(data)
                    except ValidationError as exc:
                        for error in map_validation_error(exc):
                            error.fieldPath = [rel_name, "rel", *error.fieldPath]
                            relation_errors.append(error)
                        return None
                    return link_cls(**{name: getattr(validated, name) for name in data})

                for item in values:
                    payload = {}
                    stale_snapshot = None
                    if isinstance(item, dict):
                        payload = {
                            key: value
                            for key, value in (item.get("rel") or {}).items()
                            if key in using_fields
                        }
                        item, stale_snapshot = _split_dest_input(item)
                    if item in (None, "", "None"):
                        continue
                    if marker := info.get("crossstore"):
                        dest = _crossstore.read_dest(marker, str(item))
                        if dest is None:
                            dest = stale_snapshot
                        if dest is None:
                            relation_errors.append(relation_error(rel_name, "Unknown key"))
                        elif (link := _validated_link(
                            info["target"], {"key": dest["key"], "dest": dest} | payload,
                        )) is not None:
                            primary_keys.append(link)
                        continue
                    primary_key = info["target"].viur_parse_key(str(item))
                    if primary_key is None:
                        relation_errors.append(relation_error(rel_name))
                    elif link_cls := info.get("link"):
                        if (link := _validated_link(
                            link_cls, {info["dest_fk"]: primary_key} | payload,
                        )) is not None:
                            primary_keys.append(link)
                    else:
                        primary_keys.append(primary_key)
                # MultipleConstraints; duplicates forbidden by default (the composite PK
                # would collapse them silently)
                constraints = getattr(cls, "viur_relation_meta", {}) \
                    .get(rel_name, {}).get("multiple")
                if not isinstance(constraints, dict):
                    constraints = {}

                def _identity(item: t.Any) -> t.Any:
                    if isinstance(item, SQLModel):
                        return getattr(item, info.get("dest_fk") or "key")
                    return item

                identity = [_identity(item) for item in primary_keys]
                if not constraints.get("duplicates", False) \
                        and len(set(identity)) != len(identity):
                    relation_errors.append(
                        relation_error(rel_name, "Duplicate entries are not allowed"))
                if (minimum := int(constraints.get("min", 0))) \
                        and len(primary_keys) < minimum:
                    relation_errors.append(
                        relation_error(rel_name, f"Too few entries (min {minimum})"))
                if (maximum := int(constraints.get("max", 0))) \
                        and len(primary_keys) > maximum:
                    relation_errors.append(
                        relation_error(rel_name, f"Too many entries (max {maximum})"))
                pending_relations[rel_name] = primary_keys
                continue
            if isinstance(raw, dict):
                raw = (raw.get("dest") or raw).get("key")
            if raw in (None, "", "None"):
                cleaned[info["fk"]] = None
                continue
            primary_key = info["target"].viur_parse_key(str(raw))
            if primary_key is None:
                relation_errors.append(relation_error(rel_name))
            else:
                cleaned[info["fk"]] = primary_key
        if relation_errors:
            return cls._viur_form(cleaned, pending_relations), relation_errors

        try:
            instance = cls.model_validate(cleaned)
        except ValidationError as exc:
            return cls._viur_form(cleaned, pending_relations), map_validation_error(exc)
        if pending_relations:
            instance.__dict__["_viur_pending_relations"] = pending_relations
        return instance, []

    @classmethod
    def _viur_form(cls, cleaned: dict, pending_relations: dict) -> t.Self:
        """Unvalidated instance carrying the submitted values for a rejected re-render. Never persist."""
        if getattr(cls, "__table__", None) is not None:
            form = cls(**cleaned)
        else:
            form = cls.model_construct(**cleaned)
        form.__dict__["_viur_unvalidated"] = True
        if pending_relations:
            form.__dict__["_viur_pending_relations"] = pending_relations
        return form

    # --- Renderable protocol (SkeletonInstance aliases) ---------------------

    def dump(self, *, bones: t.Iterable[str] = ()) -> dict:
        return self.viur_dump(bones=bones)

    def structure(self) -> dict:
        full = self.viur_structure()
        if bones := self.__dict__.get("_viur_bones"):  # client bonelist → subskel-shaped
            return {name: bone for name, bone in full.items() if name in bones}
        return full

    @classmethod
    def _viur_kind(cls) -> str:
        table_name = getattr(cls, "__tablename__", None)
        # non-table models: __tablename__ is a declared_attr descriptor
        return table_name if isinstance(table_name, str) else cls.__name__.lower()

    @classmethod
    def viur_encode_key(cls, primary_key: int | str) -> str:
        """Primary key → opaque key string."""
        raw = f"{cls._viur_kind()}{KEY_SEPARATOR}{primary_key}".encode()
        return base64.urlsafe_b64encode(raw).decode().rstrip("=")

    @property
    def viur_key(self) -> str | None:
        """Opaque urlsafe key; ``None`` before insert."""
        if self.id is None:
            return None
        return type(self).viur_encode_key(self.id)

    @classmethod
    def viur_parse_key(cls, key: str) -> int | str | None:
        """Primary key from a ``viur_key``; ``None`` for malformed or foreign keys."""
        try:
            raw = base64.urlsafe_b64decode(key + "=" * (-len(key) % 4)).decode()
        except (ValueError, UnicodeDecodeError):
            return None
        table_name, separator, primary_key = raw.partition(KEY_SEPARATOR)
        if not separator or table_name != cls._viur_kind() or not primary_key:
            return None
        return int(primary_key) if primary_key.isdigit() else primary_key

errors property writable

errors: list

Client-input errors (SkeletonInstance.errors counterpart). Kept in __dict__: loaded rows skip __init__, so pydantic private attrs may not exist.

viur_key property

viur_key: str | None

Opaque urlsafe key; None before insert.

viur_structure classmethod

viur_structure() -> dict

Skeleton-compatible structure dict, fresh deep copy (the caller owns it).

Source code in src/viur/models/base.py

@classmethod
def viur_structure(cls) -> dict:
    """Skeleton-compatible structure dict, fresh deep copy (the caller owns it)."""
    return copy.deepcopy(cls._viur_structure_shared())

viur_crossstore classmethod

viur_crossstore() -> dict

name → SkeletonRefMarker, cached.

Source code in src/viur/models/base.py

@classmethod
def viur_crossstore(cls) -> dict:
    """``name → SkeletonRefMarker``, cached."""
    if "_viur_crossstore" not in cls.__dict__:
        cls._viur_crossstore = crossstore_fields(cls)
    return cls._viur_crossstore

viur_relations classmethod

viur_relations() -> dict

Relations per relations_for_model, cached.

Source code in src/viur/models/base.py

@classmethod
def viur_relations(cls) -> dict:
    """Relations per ``relations_for_model``, cached."""
    if "_viur_relations" not in cls.__dict__:
        relations = relations_for_model(cls)
        _check_relation_databases(cls, relations)
        cls._viur_relations = relations
    return cls._viur_relations

viur_write_only classmethod

viur_write_only() -> frozenset

Write-only field names, cached.

Source code in src/viur/models/base.py

@classmethod
def viur_write_only(cls) -> frozenset:
    """Write-only field names, cached."""
    if "_viur_write_only" not in cls.__dict__:
        cls._viur_write_only = write_only_fields(cls)
    return cls._viur_write_only

viur_dump

viur_dump(*, bones: Iterable[str] = ()) -> dict

SkeletonInstance.dump()-shaped values: opaque key, ISO datetimes, enum values, relations in RelationalBone shape. bones restricts the output; without it, a client bonelist attached by SQLList (_viur_bones) does.

Source code in src/viur/models/base.py

def viur_dump(self, *, bones: t.Iterable[str] = ()) -> dict:
    """``SkeletonInstance.dump()``-shaped values: opaque ``key``, ISO datetimes, enum values,
    relations in ``RelationalBone`` shape. ``bones`` restricts the output; without it, a
    client bonelist attached by ``SQLList`` (``_viur_bones``) does."""
    bones = set(bones) if bones else self.__dict__.get("_viur_bones", ())
    structure = self._viur_structure_shared()
    relations = self.viur_relations()
    crossstore = self.viur_crossstore()
    write_only = self.viur_write_only()
    out = {}
    for name in structure:
        if bones and name not in bones:
            continue
        if name == "key":
            out[name] = self.viur_key
        elif name in crossstore:
            out[name] = self._viur_dump_crossstore(name, crossstore[name])
        elif name in relations:
            out[name] = self._viur_dump_relation(name, relations[name])
        elif name in write_only:
            out[name] = structure[name]["emptyvalue"]
        elif languages := structure[name]["languages"]:
            value = getattr(self, name)
            if not isinstance(value, dict):  # unvalidated form
                value = {}
            out[name] = {lang: value.get(lang) for lang in languages}
        elif self.__dict__.get("_viur_unvalidated") \
                and name in type(self).model_computed_fields:
            # computed fields over raw values may raise on a rejected form
            try:
                out[name] = _dump_value(getattr(self, name))
            except Exception:
                out[name] = structure[name]["emptyvalue"]
        else:
            out[name] = _dump_value(getattr(self, name))
    return out

viur_from_client classmethod

viur_from_client(data: dict) -> tuple[Self | None, list[ReadFromClientError]]

skel.fromClient() counterpart. Unknown and read-only fields are dropped. Returns (instance, []) or (form, errors) — form is unvalidated (_viur_form), never persist it.

Source code in src/viur/models/base.py

@classmethod
def viur_from_client(
    cls, data: dict,
) -> tuple[t.Self | None, list["ReadFromClientError"]]:
    """``skel.fromClient()`` counterpart. Unknown and read-only fields are dropped. Returns
    ``(instance, [])`` or ``(form, errors)`` — ``form`` is unvalidated (``_viur_form``),
    never persist it."""
    structure = cls._viur_structure_shared()
    cleaned = {
        name: value
        for name, value in data.items()
        if name in structure and not structure[name]["readonly"]
    }

    # dotted sub-keys: title.de=…, pos.lat=…, address.street=…, name.<idx>.<field>=…
    dotted: dict[str, dict] = {}
    for key, value in data.items():
        base_name, separator, suffix = key.partition(".")
        if separator and base_name in structure and not structure[base_name]["readonly"]:
            dotted.setdefault(base_name, {})[suffix] = value
    for base_name, subs in dotted.items():
        if structure[base_name]["languages"]:
            cleaned[base_name] = _merge_sub_keys(cleaned.get(base_name), subs)
        elif structure[base_name]["type"] == "spatial":
            cleaned[base_name] = [subs.get("lat"), subs.get("lng")]
        elif structure[base_name]["type"] == "record" \
                and not structure[base_name]["multiple"]:
            cleaned[base_name] = _merge_sub_keys(cleaned.get(base_name), subs)
        elif structure[base_name].get("multiple") \
                and subs and all(key.partition(".")[0].isdigit() for key in subs):
            items: dict[int, dict] = {}
            for key, value in subs.items():
                index, _, field = key.partition(".")
                items.setdefault(int(index), {})[field] = value
            ordered = [items[index] for index in sorted(items)]
            if structure[base_name]["type"] == "record":
                cleaned[base_name] = ordered
            else:
                cleaned[base_name] = [
                    {"dest": {"key": item.pop("key", None)}, "rel": item}
                    for item in ordered
                ]

    # Language dicts keep only declared languages.
    for name, bone in structure.items():
        if (languages := bone["languages"]) and isinstance(cleaned.get(name), dict):
            cleaned[name] = {
                lang: value
                for lang, value in cleaned[name].items()
                if lang in languages and value is not None
            }

    _normalize_empty_values(structure, cleaned)

    # empty write-only input keeps the stored value
    for name in cls.viur_write_only():
        if name in cleaned and cleaned[name] in ("", None):
            del cleaned[name]

    relation_errors = []

    # cross-store: read_dest is the existence check; a full stale snapshot survives a
    # vanished target, a bare unknown key rejects
    for name, marker in cls.viur_crossstore().items():
        if name not in cleaned:
            continue
        raw = cleaned.pop(name)
        values = raw if isinstance(raw, (list, tuple)) else [raw]
        dests = []
        for item in values:
            item, stale_snapshot = _split_dest_input(item)
            if item in (None, "", "None"):
                continue
            dest = _crossstore.read_dest(marker, str(item))
            if dest is not None:
                dests.append(dest)
            elif stale_snapshot is not None:
                dests.append(stale_snapshot)
            else:
                relation_errors.append(relation_error(name, "Unknown key"))
        if marker.multiple:
            cleaned[name] = dests
        else:
            cleaned[name] = dests[0] if dests else None

    # to-one → FK column; many-to-many parked in _viur_pending_relations for SQLList
    # (existence is checked there, only the key format here)
    pending_relations = {}
    for rel_name, info in cls.viur_relations().items():
        if rel_name not in cleaned:
            continue
        raw = cleaned.pop(rel_name)
        if info["multiple"]:
            values = raw if isinstance(raw, (list, tuple)) else [raw]
            using_fields = info.get("using_fields") or ()
            primary_keys = []

            def _validated_link(link_cls: type, data: dict) -> t.Any:
                """Validate one link row; errors get the ``[name, "rel", field]`` path. Built from
                the submitted keys only — ``model_validate`` would pin the parent FK to ``None``."""
                try:
                    validated = link_cls.model_validate(data)
                except ValidationError as exc:
                    for error in map_validation_error(exc):
                        error.fieldPath = [rel_name, "rel", *error.fieldPath]
                        relation_errors.append(error)
                    return None
                return link_cls(**{name: getattr(validated, name) for name in data})

            for item in values:
                payload = {}
                stale_snapshot = None
                if isinstance(item, dict):
                    payload = {
                        key: value
                        for key, value in (item.get("rel") or {}).items()
                        if key in using_fields
                    }
                    item, stale_snapshot = _split_dest_input(item)
                if item in (None, "", "None"):
                    continue
                if marker := info.get("crossstore"):
                    dest = _crossstore.read_dest(marker, str(item))
                    if dest is None:
                        dest = stale_snapshot
                    if dest is None:
                        relation_errors.append(relation_error(rel_name, "Unknown key"))
                    elif (link := _validated_link(
                        info["target"], {"key": dest["key"], "dest": dest} | payload,
                    )) is not None:
                        primary_keys.append(link)
                    continue
                primary_key = info["target"].viur_parse_key(str(item))
                if primary_key is None:
                    relation_errors.append(relation_error(rel_name))
                elif link_cls := info.get("link"):
                    if (link := _validated_link(
                        link_cls, {info["dest_fk"]: primary_key} | payload,
                    )) is not None:
                        primary_keys.append(link)
                else:
                    primary_keys.append(primary_key)
            # MultipleConstraints; duplicates forbidden by default (the composite PK
            # would collapse them silently)
            constraints = getattr(cls, "viur_relation_meta", {}) \
                .get(rel_name, {}).get("multiple")
            if not isinstance(constraints, dict):
                constraints = {}

            def _identity(item: t.Any) -> t.Any:
                if isinstance(item, SQLModel):
                    return getattr(item, info.get("dest_fk") or "key")
                return item

            identity = [_identity(item) for item in primary_keys]
            if not constraints.get("duplicates", False) \
                    and len(set(identity)) != len(identity):
                relation_errors.append(
                    relation_error(rel_name, "Duplicate entries are not allowed"))
            if (minimum := int(constraints.get("min", 0))) \
                    and len(primary_keys) < minimum:
                relation_errors.append(
                    relation_error(rel_name, f"Too few entries (min {minimum})"))
            if (maximum := int(constraints.get("max", 0))) \
                    and len(primary_keys) > maximum:
                relation_errors.append(
                    relation_error(rel_name, f"Too many entries (max {maximum})"))
            pending_relations[rel_name] = primary_keys
            continue
        if isinstance(raw, dict):
            raw = (raw.get("dest") or raw).get("key")
        if raw in (None, "", "None"):
            cleaned[info["fk"]] = None
            continue
        primary_key = info["target"].viur_parse_key(str(raw))
        if primary_key is None:
            relation_errors.append(relation_error(rel_name))
        else:
            cleaned[info["fk"]] = primary_key
    if relation_errors:
        return cls._viur_form(cleaned, pending_relations), relation_errors

    try:
        instance = cls.model_validate(cleaned)
    except ValidationError as exc:
        return cls._viur_form(cleaned, pending_relations), map_validation_error(exc)
    if pending_relations:
        instance.__dict__["_viur_pending_relations"] = pending_relations
    return instance, []

viur_encode_key classmethod

viur_encode_key(primary_key: int | str) -> str

Primary key → opaque key string.

Source code in src/viur/models/base.py

@classmethod
def viur_encode_key(cls, primary_key: int | str) -> str:
    """Primary key → opaque key string."""
    raw = f"{cls._viur_kind()}{KEY_SEPARATOR}{primary_key}".encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")

viur_parse_key classmethod

viur_parse_key(key: str) -> int | str | None

Primary key from a viur_key; None for malformed or foreign keys.

Source code in src/viur/models/base.py

@classmethod
def viur_parse_key(cls, key: str) -> int | str | None:
    """Primary key from a ``viur_key``; ``None`` for malformed or foreign keys."""
    try:
        raw = base64.urlsafe_b64decode(key + "=" * (-len(key) % 4)).decode()
    except (ValueError, UnicodeDecodeError):
        return None
    table_name, separator, primary_key = raw.partition(KEY_SEPARATOR)
    if not separator or table_name != cls._viur_kind() or not primary_key:
        return None
    return int(primary_key) if primary_key.isdigit() else primary_key

Record

Bases: SQLModel

Base for nested record values (RelSkel analogue): no table, no key, no system fields.

Source code in src/viur/models/base.py

class Record(SQLModel):
    """Base for nested record values (``RelSkel`` analogue): no table, no key, no system fields."""

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: t.Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        cls._viur_structure_shared()  # fail fast on unmappable fields

    @classmethod
    def _viur_structure_shared(cls) -> dict:
        """Cached structure, shared — read-only."""
        if "_viur_structure" not in cls.__dict__:
            cls._viur_structure = structure_for_model(cls)
        return cls._viur_structure

    @classmethod
    def viur_structure(cls) -> dict:
        """``using`` structure, fresh copy."""
        return copy.deepcopy(cls._viur_structure_shared())

viur_structure classmethod

viur_structure() -> dict

using structure, fresh copy.

Source code in src/viur/models/base.py

@classmethod
def viur_structure(cls) -> dict:
    """``using`` structure, fresh copy."""
    return copy.deepcopy(cls._viur_structure_shared())

ModelsConfig

conf.models.* namespace owned by viur-models.

Source code in src/viur/models/config.py

class ModelsConfig:
    """``conf.models.*`` namespace owned by viur-models."""

    def __init__(self) -> None:
        #: Databases by name (``Model.viur_database``, ``"default"`` required). Entry keys:
        #: ``engine`` (``"memory"`` | ``"sqlite"`` | ``"postgres"`` | ``"bigquery"``),
        #: ``sqlite_file``, ``postgres_dsn``, ``bigquery_dsn``, ``engine_options``, or ``url``.
        self.databases: dict[str, dict] = {}

CrossStoreIndex

Bases: SQLModel

Reverse index target key → (table, row, field) of JSON-column references; maintained by SQLList on write. SkeletonLink tables index themselves via key.

Source code in src/viur/models/crossstore.py

class CrossStoreIndex(SQLModel, table=True):
    """Reverse index target key → (table, row, field) of JSON-column references; maintained by
    ``SQLList`` on write. ``SkeletonLink`` tables index themselves via ``key``."""

    __tablename__ = "viur_models_relations"

    id: int | None = SQLModelField(default=None, primary_key=True)
    target_key: str = SQLModelField(index=True)
    model_table: str = SQLModelField(index=True)
    row_id: int
    field: str

SkeletonLink

Bases: SQLModel

Base for link-table-backed multiple cross-store references (one row per target).

Carries key (datastore key, part of the PK) and the dest snapshot; subclasses add the parent FK and set viur_kind (required) plus the viur_link_* ClassVars. The parent relationship needs cascade="all, delete-orphan".

Source code in src/viur/models/crossstore.py

class SkeletonLink(SQLModel):
    """Base for link-table-backed multiple cross-store references (one row per target).

    Carries ``key`` (datastore key, part of the PK) and the ``dest`` snapshot; subclasses add
    the parent FK and set ``viur_kind`` (required) plus the ``viur_link_*`` ClassVars. The
    parent relationship needs ``cascade="all, delete-orphan"``.
    """

    key: str = SQLModelField(primary_key=True)
    dest: dict = SQLModelField(default_factory=dict, sa_type=JSON)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: t.Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        _register_link(cls)

    viur_kind: t.ClassVar[str | None] = None
    viur_link_ref_keys: t.ClassVar[tuple[str, ...]] = ("name",)
    viur_link_module: t.ClassVar[str | None] = None
    viur_link_type_suffix: t.ClassVar[str | None] = None
    viur_link_format: t.ClassVar[str | None] = None
    viur_link_extras: t.ClassVar[dict | None] = None

    @classmethod
    def viur_marker(cls) -> SkeletonRefMarker:
        if not cls.viur_kind:
            raise TypeError(
                f"{cls.__name__} must set the ``viur_kind`` ClassVar to the "
                "referenced skeleton kind"
            )
        return SkeletonRefMarker(
            kind=cls.viur_kind,
            ref_keys=tuple(cls.viur_link_ref_keys),
            module=cls.viur_link_module or cls.viur_kind,
            type_suffix=cls.viur_link_type_suffix,
            multiple=True,
            format=cls.viur_link_format,
            extras=dict(cls.viur_link_extras) if cls.viur_link_extras else None,
        )

RecordJSON

Bases: TypeDecorator

JSON column for nested records: record_cls instances (or lists) dump on write, validate on read.

Source code in src/viur/models/db.py

class RecordJSON(TypeDecorator):
    """JSON column for nested records: ``record_cls`` instances (or lists) dump on write, validate on read."""

    impl = JSON
    cache_ok = True

    def __init__(self, record_cls: type[SQLModel]):
        super().__init__()
        self.record_cls = record_cls

    def process_bind_param(self, value: t.Any, dialect: t.Any) -> t.Any:
        if value is None:
            return None
        if isinstance(value, (list, tuple)):
            return [self._to_plain(item) for item in value]
        return self._to_plain(value)

    def process_result_value(self, value: t.Any, dialect: t.Any) -> t.Any:
        if value is None:
            return None
        if isinstance(value, list):
            return [self.record_cls.model_validate(item) for item in value]
        return self.record_cls.model_validate(value)

    @staticmethod
    def _to_plain(value: t.Any) -> t.Any:
        return value.model_dump(mode="json") if isinstance(value, SQLModel) else value

RelationLink

Bases: SQLModel

Marker base for association-object link tables.

Source code in src/viur/models/links.py

class RelationLink(SQLModel):
    """Marker base for association-object link tables."""

    #: Same database as both related models.
    viur_database: t.ClassVar[str] = "default"

BoneType dataclass

Annotation marker setting a field's bone type.

Parameters:

Name	Type	Description	Default
name	str	Structure type string.	required
extras	dict | None	Additional structure keys.	None
replace	bool	Drop the Python type's structure; emptyvalue/extras define the bone.	False
emptyvalue	Any	Emitted emptyvalue (with replace).	None
write_only	bool	Dumps emit the emptyvalue instead of the value.	False

Source code in src/viur/models/types.py

@dataclasses.dataclass(frozen=True)
class BoneType:
    """Annotation marker setting a field's bone type.

    :param name: Structure ``type`` string.
    :param extras: Additional structure keys.
    :param replace: Drop the Python type's structure; ``emptyvalue``/``extras`` define the bone.
    :param emptyvalue: Emitted ``emptyvalue`` (with ``replace``).
    :param write_only: Dumps emit the ``emptyvalue`` instead of the value.
    """

    name: str
    extras: dict | None = None
    replace: bool = False
    emptyvalue: t.Any = None
    write_only: bool = False

Language

Language[X]: {lang: value} dict in a JSON column; languages from Field(languages=…) or set_default_languages.

Source code in src/viur/models/types.py

class Language:
    """``Language[X]``: ``{lang: value}`` dict in a JSON column; languages from
    ``Field(languages=…)`` or ``set_default_languages``."""

    def __class_getitem__(cls, inner: t.Any) -> t.Any:
        return t.Annotated[dict[str, str | None], LanguageWrapper(inner)]

install

install(*, engine: str | None = None, sqlite_file: str | None = None, postgres_dsn: str | None = None, bigquery_dsn: str | None = None, engine_options: dict | None = None, databases: dict[str, dict] | None = None, refresh_hooks: bool = True, **refresh_hook_kwargs: Any) -> Engine

Attach conf.models, apply the settings, build the engines, install the refresh hooks. Once, before core.setup(). None leaves a setting untouched.

Parameters:

Name	Type	Description	Default
engine	str | None	Preset of the default database: "memory" / "sqlite" / "postgres" / "bigquery". The flat arguments are the databases["default"] entry.	None
engine_options	dict | None	Extra create_engine kwargs.	None
databases	dict[str, dict] | None	Entries for conf.models.databases (merged by name).	None
refresh_hooks	bool	False skips install_refresh_hooks.	True
refresh_hook_kwargs	Any	Passed to install_refresh_hooks.	{}

Source code in src/viur/models/boot.py

def install(
    *,
    engine: str | None = None,
    sqlite_file: str | None = None,
    postgres_dsn: str | None = None,
    bigquery_dsn: str | None = None,
    engine_options: dict | None = None,
    databases: dict[str, dict] | None = None,
    refresh_hooks: bool = True,
    **refresh_hook_kwargs: t.Any,
) -> "Engine":
    """Attach ``conf.models``, apply the settings, build the engines, install the
    refresh hooks. Once, before ``core.setup()``. ``None`` leaves a setting untouched.

    :param engine: Preset of the default database: ``"memory"`` / ``"sqlite"`` / ``"postgres"``
        / ``"bigquery"``. The flat arguments are the ``databases["default"]`` entry.
    :param engine_options: Extra ``create_engine`` kwargs.
    :param databases: Entries for ``conf.models.databases`` (merged by name).
    :param refresh_hooks: ``False`` skips ``install_refresh_hooks``.
    :param refresh_hook_kwargs: Passed to ``install_refresh_hooks``.
    """
    from .crossstore import install_refresh_hooks

    config = install_config()
    for name, entry in (databases or {}).items():
        config.databases.setdefault(name, {}).update(entry)
    default = {
        key: value for key, value in (
            ("engine", engine), ("sqlite_file", sqlite_file), ("postgres_dsn", postgres_dsn),
            ("bigquery_dsn", bigquery_dsn), ("engine_options", engine_options),
        ) if value is not None
    }
    if default:
        config.databases.setdefault(db.DEFAULT, {}).update(default)

    sql_engine = configure_from_conf()

    if refresh_hooks:
        install_refresh_hooks(**refresh_hook_kwargs)
    return sql_engine

setup

setup(engine: Engine | None = None, *, migrations: str | PathLike | None = None, initial_revision: bool = True, **scaffold_kwargs: Any) -> str | None

Report the schema revision of every configured database; after core.setup(). Memory preset: create_all of that database's tables. migrations generates the missing Alembic scaffold (dev server only) and, if it generated any, the first revision.

Parameters:

Name	Type	Description	Default
engine	Engine | None	Engine to inspect instead of the configured ones.	None
migrations	str | PathLike | None	Directory for alembic.ini + migrations/; None generates nothing.	None
initial_revision	bool	False leaves the first revision to the developer.	True
scaffold_kwargs	Any	Passed to scaffold.generate.	{}

Returns:

Type	Description
str | None	Stamped revision of the default database, or None (memory preset / unmigrated).

Source code in src/viur/models/boot.py

def setup(
    engine: "Engine | None" = None,
    *,
    migrations: str | os.PathLike | None = None,
    initial_revision: bool = True,
    **scaffold_kwargs: t.Any,
) -> str | None:
    """Report the schema revision of every configured database; after ``core.setup()``.
    Memory preset: ``create_all`` of that database's tables. ``migrations`` generates the
    missing Alembic scaffold (dev server only) and, if it generated any, the first revision.

    :param engine: Engine to inspect instead of the configured ones.
    :param migrations: Directory for ``alembic.ini`` + ``migrations/``; ``None`` generates nothing.
    :param initial_revision: ``False`` leaves the first revision to the developer.
    :param scaffold_kwargs: Passed to ``scaffold.generate``.
    :returns: Stamped revision of the default database, or ``None`` (memory preset / unmigrated).
    """
    if migrations is not None and _ensure_scaffold(migrations, **scaffold_kwargs) \
            and initial_revision:
        _bootstrap_revision(migrations)

    if engine is not None:
        return _setup_database(db.DEFAULT, engine)
    if not SQLModel.metadata.tables:
        logger.warning(
            "viur-models: setup() found an empty SQLModel.metadata — no "
            "model was imported yet, so the in-memory database stays "
            "empty. Call setup() after core.setup()."
        )
    revisions = {name: _setup_database(name, get_engine(name)) for name in db.engine_names()}
    return revisions.get(db.DEFAULT)

map_validation_error

map_validation_error(exc: ValidationError) -> list[ReadFromClientError]

missing → NotSet, anything else → Invalid; fieldPath is pydantic's loc.

Source code in src/viur/models/client.py

def map_validation_error(exc: "ValidationError") -> list["ReadFromClientError"]:
    """``missing`` → ``NotSet``, anything else → ``Invalid``; ``fieldPath`` is pydantic's ``loc``."""
    from viur.core.bones.base import ReadFromClientError, ReadFromClientErrorSeverity

    return [
        ReadFromClientError(
            severity=(
                ReadFromClientErrorSeverity.NotSet
                if error["type"] == "missing"
                else ReadFromClientErrorSeverity.Invalid
            ),
            errorMessage=error["msg"],
            fieldPath=[str(loc) for loc in error["loc"]],
        )
        for error in exc.errors()
    ]

install_config

install_config() -> ModelsConfig

Attach ModelsConfig to conf.models; idempotent.

Source code in src/viur/models/config.py

def install_config() -> ModelsConfig:
    """Attach ``ModelsConfig`` to ``conf.models``; idempotent."""
    from viur.core import conf

    existing = getattr(conf, "models", None)
    if isinstance(existing, ModelsConfig):
        return existing
    models = ModelsConfig()
    conf.models = models
    return models

FileRef

FileRef(ref_keys: Sequence[str] = ('name', 'mimetype', 'size', 'width', 'height', 'dlkey', 'serving_url', 'derived', 'public'), **kwargs: Any) -> Any

FileBone analogue (relational.tree.leaf.file.file); valid_mime_types/public via extras.

Source code in src/viur/models/crossstore.py

def FileRef(
    ref_keys: t.Sequence[str] = (
        "name", "mimetype", "size", "width", "height",
        "dlkey", "serving_url", "derived", "public",
    ),
    **kwargs: t.Any,
) -> t.Any:
    """``FileBone`` analogue (``relational.tree.leaf.file.file``); ``valid_mime_types``/``public`` via ``extras``."""
    kwargs.setdefault("type_suffix", "tree.leaf.file")
    return SkeletonRef("file", ref_keys, **kwargs)

SkeletonRef

SkeletonRef(kind: str, ref_keys: Sequence[str] = ('name',), *, module: str | None = None, type_suffix: str | None = None, multiple: bool = False, format: str | None = None, extras: dict | None = None) -> Any

Cross-store reference type for skeleton kind.

Parameters:

Name	Type	Description	Default
ref_keys	Sequence[str]	Target bones in dest/relskel (key, shortkey always included).	('name',)
module	str | None	Serving module (default: kind).	None
type_suffix	str | None	relational.<suffix>.<kind> (FileBone style).	None
format	str | None	Display format default; Field(format=…) wins.	None
extras	dict | None	Extra structure keys, emitted verbatim.	None

Source code in src/viur/models/crossstore.py

def SkeletonRef(
    kind: str,
    ref_keys: t.Sequence[str] = ("name",),
    *,
    module: str | None = None,
    type_suffix: str | None = None,
    multiple: bool = False,
    format: str | None = None,
    extras: dict | None = None,
) -> t.Any:
    """Cross-store reference type for skeleton ``kind``.

    :param ref_keys: Target bones in ``dest``/``relskel`` (``key``, ``shortkey`` always included).
    :param module: Serving module (default: ``kind``).
    :param type_suffix: ``relational.<suffix>.<kind>`` (``FileBone`` style).
    :param format: Display format default; ``Field(format=…)`` wins.
    :param extras: Extra structure keys, emitted verbatim.
    """
    marker = SkeletonRefMarker(
        kind=kind,
        ref_keys=tuple(ref_keys),
        module=module or kind,
        type_suffix=type_suffix,
        multiple=multiple,
        format=format,
        extras=dict(extras) if extras else None,
    )
    if multiple:
        return t.Annotated[list[dict], marker]
    return t.Annotated[dict, marker]

UserRef

UserRef(ref_keys: Sequence[str] = ('name', 'firstname', 'lastname'), **kwargs: Any) -> Any

UserBone analogue (relational.user).

Source code in src/viur/models/crossstore.py

def UserRef(ref_keys: t.Sequence[str] = ("name", "firstname", "lastname"), **kwargs: t.Any) -> t.Any:
    """``UserBone`` analogue (``relational.user``)."""
    kwargs.setdefault("format", "$(dest.lastname), $(dest.firstname) ($(dest.name))")
    return SkeletonRef("user", ref_keys, **kwargs)

install_refresh_hooks

install_refresh_hooks(*, missing_on_delete: str = 'set_null', countdown: int = 10) -> None

Wrap Skeleton.postSavedHandler/postDeletedHandler to defer refresh_for_target for referenced kinds. Once at boot; idempotent. Skeletons overriding the handlers without super() must call it from their own onEdited/onDeleted.

Parameters:

Name	Type	Description	Default
missing_on_delete	str	"set_null" clears references on delete, "keep" leaves them.	'set_null'
countdown	int	Task delay in seconds.	10

Source code in src/viur/models/crossstore.py

def install_refresh_hooks(*, missing_on_delete: str = "set_null", countdown: int = 10) -> None:
    """Wrap ``Skeleton.postSavedHandler``/``postDeletedHandler`` to defer ``refresh_for_target``
    for referenced kinds. Once at boot; idempotent. Skeletons overriding the handlers without
    ``super()`` must call it from their own ``onEdited``/``onDeleted``.

    :param missing_on_delete: ``"set_null"`` clears references on delete, ``"keep"`` leaves them.
    :param countdown: Task delay in seconds.
    """
    from viur.core import tasks
    from viur.core.skeleton import Skeleton

    if missing_on_delete not in ("keep", "set_null"):
        raise ValueError('missing_on_delete must be "keep" or "set_null"')
    if getattr(Skeleton, "_viur_models_refresh_hooks", False):
        return  # idempotent

    original_saved = Skeleton.postSavedHandler.__func__
    original_deleted = Skeleton.postDeletedHandler.__func__

    @tasks.CallDeferred
    def _deferred_refresh(key: str, missing: str) -> None:
        refresh_for_target(key, missing=missing)

    def _is_referenced(kind: t.Any) -> bool:
        return kind in referenced_kinds()

    @classmethod
    def postSavedHandler(cls, skel, key, dbObj):  # noqa: ANN001
        original_saved(cls, skel, key, dbObj)
        if _is_referenced(getattr(skel, "kindName", None)):
            _deferred_refresh(str(key), missing="keep", _countdown=countdown)

    @classmethod
    def postDeletedHandler(cls, skel, key):  # noqa: ANN001
        original_deleted(cls, skel, key)
        if _is_referenced(getattr(skel, "kindName", None)):
            _deferred_refresh(str(key), missing=missing_on_delete, _countdown=countdown)

    Skeleton.postSavedHandler = postSavedHandler
    Skeleton.postDeletedHandler = postDeletedHandler
    Skeleton._viur_models_refresh_hooks = True

refresh_crossstore

refresh_crossstore(*model_classes: type, missing: str = 'keep') -> dict

Re-read every referenced target and rewrite stale snapshots (full scan per model; SkeletonLink tables via key). missing="set_null" clears vanished targets. Returns {"checked", "refreshed", "cleared"}.

Source code in src/viur/models/crossstore.py

def refresh_crossstore(*model_classes: type, missing: str = "keep") -> dict:
    """Re-read every referenced target and rewrite stale snapshots (full scan per model;
    ``SkeletonLink`` tables via ``key``). ``missing="set_null"`` clears vanished targets.
    Returns ``{"checked", "refreshed", "cleared"}``."""
    from sqlmodel import select

    from .db import get_session

    if missing not in ("keep", "set_null"):
        raise ValueError('missing must be "keep" or "set_null"')

    stats = {"checked": 0, "refreshed": 0, "cleared": 0}
    seen_link_tables: set[type] = set()
    read = _dest_reader()

    def _fresh(marker: SkeletonRefMarker, dest: dict) -> dict | None:
        stats["checked"] += 1  # rows, not reads
        return read(marker, dest.get("key"))

    for model_cls in model_classes:
        json_fields = model_cls.viur_crossstore()
        link_relations = {
            name: info for name, info in model_cls.viur_relations().items()
            if info.get("crossstore") and info["target"] not in seen_link_tables
        }
        if not json_fields and not link_relations:
            continue

        with get_session(model_cls) as session:
            if json_fields:
                for row in session.exec(select(model_cls)).all():
                    changed = False
                    for name, marker in json_fields.items():
                        value = getattr(row, name)
                        if not value:
                            continue
                        if marker.multiple:
                            fresh_list = []
                            for dest in value:
                                fresh = _fresh(marker, dest)
                                if fresh is None:
                                    if missing == "set_null":
                                        stats["cleared"] += 1
                                        changed = True
                                        continue
                                    fresh = dest
                                elif fresh != dest:
                                    stats["refreshed"] += 1
                                    changed = True
                                fresh_list.append(fresh)
                            if changed:
                                setattr(row, name, fresh_list)
                        else:
                            fresh = _fresh(marker, value)
                            if fresh is None:
                                if missing == "set_null":
                                    setattr(row, name, None)
                                    stats["cleared"] += 1
                                    changed = True
                            elif fresh != value:
                                setattr(row, name, fresh)
                                stats["refreshed"] += 1
                                changed = True
                    if changed:
                        session.add(row)

            for info in link_relations.values():
                link_cls = info["target"]
                seen_link_tables.add(link_cls)
                marker = info["crossstore"]
                for link in session.exec(select(link_cls)).all():
                    fresh = _fresh(marker, {"key": link.key})
                    if fresh is None:
                        if missing == "set_null":
                            session.delete(link)
                            stats["cleared"] += 1
                    elif fresh != link.dest:
                        link.dest = fresh
                        session.add(link)
                        stats["refreshed"] += 1

    return stats

refresh_for_target

refresh_for_target(key: str, *, missing: str = 'keep') -> dict

Update every snapshot referencing key: JSON columns via the index, SkeletonLink rows via key. missing as in refresh_crossstore. Called by the refresh hooks.

Source code in src/viur/models/crossstore.py

def refresh_for_target(key: str, *, missing: str = "keep") -> dict:
    """Update every snapshot referencing ``key``: JSON columns via the index, ``SkeletonLink`` rows
    via ``key``. ``missing`` as in ``refresh_crossstore``. Called by the refresh hooks."""
    from sqlmodel import select

    from .db import engine_names

    if missing not in ("keep", "set_null"):
        raise ValueError('missing must be "keep" or "set_null"')

    stats = {"checked": 0, "refreshed": 0, "cleared": 0}
    for database in engine_names():
        _refresh_in(database, key, missing, stats)
    return stats

Field

Field(default: Any = PydanticUndefined, *, descr: str | None = None, required: bool | None = None, visible: bool = True, readonly: bool = False, params: dict | None = None, values: dict | None = None, compute: dict | None = None, languages: Sequence[str] | None = None, format: str | None = None, tags: str | Sequence[str] | None = None, schema_extra: dict | None = None, **kwargs: Any) -> Any

sqlmodel.Field() plus bone metadata. Constraints pydantic/SQL express (max_length, ge/le, nullability, defaults) are derived, not repeated.

Parameters:

Name	Type	Description	Default
descr	str | None	Display name (default: title-cased field name).	None
required	bool | None	Override of the pydantic derivation.	None
visible	bool	Shown in client UIs.	True
readonly	bool	Read-only bone; forces required: false.	False
params	dict | None	BaseBone.params.	None
values	dict | None	{value: label} overrides for select fields.	None
compute	dict | None	BaseBone.compute info, emitted verbatim.	None
languages	Sequence[str] | None	Language codes of a Language[X] field (JSON column).	None
format	str | None	Display format of record/relational bones, emitted verbatim.	None
tags	str | Sequence[str] | None	Classification tags (BaseBone.tags), emitted as a list.	None
schema_extra	dict | None	Extra FieldInfo kwargs, see ALLOWED_SCHEMA_EXTRA.	None
kwargs	Any	Passed to sqlmodel.Field unchanged.	{}

Source code in src/viur/models/fields.py

def Field(
    default: t.Any = PydanticUndefined,
    *,
    descr: str | None = None,
    required: bool | None = None,
    visible: bool = True,
    readonly: bool = False,
    params: dict | None = None,
    values: dict | None = None,
    compute: dict | None = None,
    languages: t.Sequence[str] | None = None,
    format: str | None = None,
    tags: str | t.Sequence[str] | None = None,
    schema_extra: dict | None = None,
    **kwargs: t.Any,
) -> t.Any:
    """``sqlmodel.Field()`` plus bone metadata. Constraints pydantic/SQL express
    (``max_length``, ``ge``/``le``, nullability, defaults) are derived, not repeated.

    :param descr: Display name (default: title-cased field name).
    :param required: Override of the pydantic derivation.
    :param visible: Shown in client UIs.
    :param readonly: Read-only bone; forces ``required: false``.
    :param params: ``BaseBone.params``.
    :param values: ``{value: label}`` overrides for select fields.
    :param compute: ``BaseBone.compute`` info, emitted verbatim.
    :param languages: Language codes of a ``Language[X]`` field (JSON column).
    :param format: Display format of record/relational bones, emitted verbatim.
    :param tags: Classification tags (``BaseBone.tags``), emitted as a list.
    :param schema_extra: Extra ``FieldInfo`` kwargs, see ``ALLOWED_SCHEMA_EXTRA``.
    :param kwargs: Passed to ``sqlmodel.Field`` unchanged.
    """
    viur_meta = {
        key: value
        for key, value in {
            "descr": descr,
            "required": required,
            "visible": visible,
            "readonly": readonly,
            "params": params,
            "values": values,
            "compute": compute,
            "languages": list(languages) if languages else None,
            "format": format,
            "tags": ([tags] if isinstance(tags, str) else list(tags)) if tags else None,
        }.items()
        if value is not None
    }

    schema_extra = dict(schema_extra or {})
    if unknown := set(schema_extra) - ALLOWED_SCHEMA_EXTRA:
        raise TypeError(
            f"Unknown schema_extra key(s) {sorted(unknown)} — pydantic would "
            f"silently drop them. Allowed: {sorted(ALLOWED_SCHEMA_EXTRA)}"
        )
    schema_extra["json_schema_extra"] = {
        **(schema_extra.get("json_schema_extra") or {}),
        VIUR_META_KEY: viur_meta,
    }

    return SQLModelField(default, schema_extra=schema_extra, **kwargs)

structure_for_model

structure_for_model(cls: type, *, include_relations: bool = True, resolve_refs: bool = True) -> dict

Skeleton-compatible structure dict. Primary key → key bone; an FK consumed by a to-one relation → one relational.<kind> bone under the relation's name. include_relations=False: scalar-only (relskel); resolve_refs=False: no datastore lookup for cross-store relskel.

Source code in src/viur/models/structure.py

def structure_for_model(
    cls: type, *, include_relations: bool = True, resolve_refs: bool = True,
) -> dict:
    """Skeleton-compatible structure dict. Primary key → ``key`` bone; an FK consumed by a
    to-one relation → one ``relational.<kind>`` bone under the relation's name.
    ``include_relations=False``: scalar-only (``relskel``); ``resolve_refs=False``: no
    datastore lookup for cross-store ``relskel``."""
    relations = relations_for_model(cls) if include_relations else {}
    consumed = {
        info["fk"]: rel_name
        for rel_name, info in relations.items()
        if info["fk"] is not None
    }

    structure = {}
    for sortindex, (name, field_info) in enumerate(cls.model_fields.items()):
        if getattr(field_info, "primary_key", False) is True:
            structure["key"] = _key_bone() | {"sortindex": sortindex}
        elif name in consumed:
            rel_name = consumed[name]
            structure[rel_name] = (
                _relational_structure(
                    rel_name, field_info, relations[rel_name], cls,
                    resolve_refs=resolve_refs,
                )
                | {"sortindex": sortindex}
            )
        else:
            structure[name] = _bone_structure(
                name, field_info, resolve_refs=resolve_refs,
            ) | {"sortindex": sortindex}

    # relations without an FK field: appended in declaration order
    sortindex = len(cls.model_fields)
    for rel_name, info in relations.items():
        if info["fk"] is not None:
            continue
        meta = getattr(cls, "viur_relation_meta", {}).get(rel_name, {})
        if marker := info.get("crossstore"):
            entry = _crossstore_structure(marker, meta, resolve_refs=resolve_refs)
            entry["descr"] = meta.get("descr") or rel_name.replace("_", " ").title()
            if info.get("using_fields"):
                entry["using"] = _using_structure(
                    info["target"], info["using_fields"], resolve_refs=resolve_refs,
                )
            entry = {
                "params": meta.get("params") or {},
                "tags": _tags(meta),
                "required": False,
                "visible": bool(meta.get("visible", True)),
                "readonly": bool(meta.get("readonly", False)),
                "unique": False,
                "languages": None,
                "indexed": True,
                "clone_behavior": dict(CLONE_BEHAVIOR),
            } | entry
        else:
            entry = _relational_structure(
                rel_name, None, info, cls, resolve_refs=resolve_refs,
            )
        structure[rel_name] = entry | {"sortindex": sortindex}
        sortindex += 1

    # computed fields: appended last
    for name, computed_info in cls.model_computed_fields.items():
        structure[name] = _computed_bone_structure(name, computed_info) \
            | {"sortindex": sortindex}
        sortindex += 1
    return structure

Spatial

Spatial(bounds_lat: tuple[float, float], bounds_lng: tuple[float, float]) -> Any

SpatialBone ("spatial") type factory; values are (lat, lng), stored as JSON.

Source code in src/viur/models/types.py

def Spatial(
    bounds_lat: tuple[float, float], bounds_lng: tuple[float, float],
) -> t.Any:
    """``SpatialBone`` (``"spatial"``) type factory; values are ``(lat, lng)``, stored as JSON."""
    return t.Annotated[tuple[float, float], BoneType(
        "spatial",
        replace=True,
        emptyvalue=[0.0, 0.0],
        extras={"boundslat": list(bounds_lat), "boundslng": list(bounds_lng)},
    )]

register_bone_type

register_bone_type(python_type: type, marker: BoneType) -> None

Map a Python type to a bone type.

Source code in src/viur/models/types.py

def register_bone_type(python_type: type, marker: BoneType) -> None:
    """Map a Python type to a bone type."""
    BONE_TYPE_REGISTRY[python_type] = marker

set_default_languages

set_default_languages(*codes: str) -> None

Project-wide default language list.

Source code in src/viur/models/types.py

def set_default_languages(*codes: str) -> None:
    """Project-wide default language list."""
    global DEFAULT_LANGUAGES
    DEFAULT_LANGUAGES = tuple(codes) or None