# SPDX-FileCopyrightText: Chris Wilson <[email protected]>

#

# SPDX-License-Identifier: MIT

"""The TinydanticModel base class and query helpers."""

from __future__ import annotations

import inspect

import sys

from importlib import metadata

from typing import (

TYPE_CHECKING,

Any,

ClassVar,

Literal,

cast,

overload,

)

from uuid import UUID, uuid4

from pydantic import BaseModel, ConfigDict, Field, TypeAdapter

from pydantic.alias_generators import to_snake

from pydantic.json_schema import SkipJsonSchema

from tinydb.queries import Query, QueryInstance

from tinydb.table import Document, Table

from tinydantic._config import (

_CONFIG_KEYS,

CONFIG_ATTR,

TinydanticConfig,

check_config_ambiguity,

get_config_value,

)

from tinydantic._errors import (

ConstraintFieldError,

DatabaseNotBoundError,

DocumentAlreadyExistsError,

DocumentIDRequiredError,

DocumentIDUpdateError,

DocumentNotFoundError,

QueryFieldError,

RevisionFieldError,

RevisionUpdateError,

SelectorError,

ShadowedFieldError,

StaleDocumentError,

TinydanticError,

UniqueConstraintError,

UnknownUpdateFieldError,

)

from tinydantic._fields import Unique, UniqueConstraint

from tinydantic._find import FindQuery

from tinydantic._query import (

_COND_NOT_GIVEN,

DocIdCondition,

DocIdQuery,

GuardedQuery,

_checked_cond,

_require_condition,

has_id_condition,

id_from_condition,

)

from tinydantic.tinydb.operations import replace

if TYPE_CHECKING:

from collections.abc import (

Callable,

Iterable,

Mapping,

MutableMapping,

Sequence,

)

from collections.abc import (

Set as AbstractSet,

)

# pydantic's ModelMetaclass lives in a private module. We

# type-check against the real class but resolve it at runtime as

# type(BaseModel), so tinydantic never imports pydantic internals

# at runtime and survives pydantic moving the module.

from pydantic._internal._model_construction import ModelMetaclass

from tinydb import TinyDB

from tinydb.queries import QueryLike

# Self is typing_extensions-only on Python 3.10 (typing.Self is

# 3.11+). This import is TYPE_CHECKING-only, so typing_extensions

# is not a runtime dependency.

from typing_extensions import Self

else:

ModelMetaclass = type(BaseModel)

# Name of the per-class attribute caching per-field TypeAdapters

# (built lazily by _field_adapter for update() serialization).

_FIELD_ADAPTERS_ATTR = "__tinydantic_field_adapters__"

# Fields tinydantic manages itself, so they are never offered to

# before_write() and never accepted back from it: `id` is the

# TinyDB document id, `revision_id` rotates on every write.

_MANAGED_FIELDS = frozenset({"id", "revision_id"})

# Name of the per-class attribute caching unique field names

# (computed once in __pydantic_init_subclass__).

_UNIQUE_MARKERS_ATTR = "__tinydantic_unique_markers__"

if sys.version_info >= (3, 14):

from annotationlib import Format, call_annotate_function

# Class-namespace keys that may hold a PEP 649 deferred-annotations

# function on Python 3.14+. CPython's compiler emits

# ``__annotate_func__`` (``type.__new__`` renames it to the class's

# ``__annotate__`` afterwards); the PEP-spelled ``__annotate__`` is

# kept as a fallback in case that implementation detail changes.

_ANNOTATE_KEYS = ("__annotate_func__", "__annotate__")

def _deferred_annotate_key(namespace: dict[str, Any]) -> str | None:

"""Return the deferred-annotations key present, if any."""

for key in _ANNOTATE_KEYS:

if key in namespace:

return key

return None

def _declared_annotations(namespace: dict[str, Any]) -> dict[str, Any]:

"""Return a class body's own annotations, eager or deferred.

Class bodies produce an eager ``__annotations__`` dict on

Python 3.13 and earlier — and on any version in modules using

``from __future__ import annotations`` — but under PEP 649

(3.14+ default semantics) they instead emit a deferred

annotate function. Both shapes are read here; the deferred

form is materialized with ``Format.FORWARDREF`` so

unresolvable names cannot raise during class creation.

"""

if "__annotations__" in namespace:

return dict(namespace["__annotations__"])

if sys.version_info >= (3, 14):

key = _deferred_annotate_key(namespace)

if key is not None:

return dict(

call_annotate_function(namespace[key], Format.FORWARDREF),

)

return {}

def _inject_revision_annotation(namespace: dict[str, Any]) -> None:

"""Add the ``revision_id`` annotation to a class-body namespace.

On eager-annotation class bodies this extends the

``__annotations__`` dict. Under PEP 649 deferred annotations

(3.14+) it wraps the compiler-emitted annotate function

in place instead — assigning an eager ``__annotations__`` dict

there would *shadow* the deferred annotations and silently

erase the user's own field annotations (pydantic would then

reject their defaulted, "non-annotated" attributes).

"""

key = (

_deferred_annotate_key(namespace)

if sys.version_info >= (3, 14)

else None

)

if sys.version_info >= (3, 14) and key is not None:

prior = namespace[key]

def annotate(format_value: int, /) -> dict[str, Any]:

"""Merge ``revision_id`` into the deferred annotations."""

fmt = Format(format_value)

merged = dict(call_annotate_function(prior, fmt))

merged["revision_id"] = (

"SkipJsonSchema[UUID | None]"

if fmt is Format.STRING

else SkipJsonSchema[UUID | None] # type: ignore[misc]

)

return merged

namespace[key] = annotate

return

annotations = dict(namespace.get("__annotations__", {}))

# SkipJsonSchema is a class at runtime and an Annotated alias

# to the type checkers, so the documented subscript form is the

# only spelling both accept — mypy just needs telling that this

# value position is deliberate.

annotations["revision_id"] = SkipJsonSchema[ # type: ignore[misc]

UUID | None

]

namespace["__annotations__"] = annotations

class _NothingMatchedError(Exception):

"""Internal control-flow marker for id-condition writes.

Raised inside the updater handed to ``Table._update_table``

when no document matched, aborting the cycle before its

storage write — a no-match write then costs one read and zero

writes, matching the public no-match paths.

"""

def _q_hint(expr: Any) -> str:

"""Return the fix for the way ``q()`` was misused.

Each wrong argument type implies a different mistake, so the

generic "expected a Query" line is followed by advice aimed at

the one that was actually made. Two of those types carry the

name of what was passed, so their advice can be the exact call

to make rather than a description of it.

"""

if isinstance(expr, str):

return (

"A string is a value, not a field: for a field whose "

"name is shadowed by a method, use field(Model, "

"'name'); for a raw document key, use "

"tinydb.where('key')."

)

if isinstance(expr, QueryInstance):

return (

"That is a built condition, not a field. Wrap the "

"field, not the comparison: q(Model.field) == value."

)

if isinstance(expr, property):

# Computed fields resolve to a Query and never arrive here,

# so a property is necessarily the kind that is never stored.

name = getattr(expr.fget, "__name__", None)

subject = repr(name) if name else "it"

shorthand = f"Model.{name}" if name else "the shorthand"

return (

f"A plain property is not serialized, so no document "

f"key exists for {subject} to match. Decorate it with "

f"pydantic's @computed_field to store it, after which "

f"{shorthand} queries it like any other field."

)

if inspect.isroutine(expr):

owner = getattr(expr, "__self__", None)

model = owner.__name__ if isinstance(owner, type) else "Model"

name = getattr(expr, "__name__", "name")

return (

f"That resolves to a method, not a field, so the "

f"comparison is a plain False matching nothing. Reach "

f"the field by name: field({model}, {name!r})."

)

return (

"Reach for the value on an instance (user.name); "

"Model.field is a query builder."

)

def q(expr: Any) -> Query:

"""Re-type a class-level field expression as a TinyDB Query.

At runtime, class-level field access like ``User.name`` already

returns a [Query][tinydb.queries.Query] (courtesy of the model

metaclass), but static type checkers see the field annotation

instead, so

``User.name == "Alice"`` types as ``bool``. Wrapping the field in

``q()`` gives the type checker the runtime truth:

```python

User.search(q(User.name) == "Alice")

```

``q()`` is a *cast*, not a constructor: it hands back the object

it was given and changes nothing at runtime. It does not accept

a field name — a string is indistinguishable from an instance

attribute that happens to hold one, so accepting either would

make ``q(user.name)`` silently build a query on the *value*. Use

[field()][tinydantic.field] for a field by name, and

``tinydb.where()`` for a raw document key.

Args:

expr: A class-level field expression (e.g. ``User.name``).

Returns:

The expression unchanged, typed as a Query.

Raises:

TypeError: If ``expr`` is not a TinyDB Query — for example

when called with an instance attribute instead of

class-level field access, or with a field name string.

The message is tailored to what was passed: a method

(a shadowed field) and a plain property both carry

their own name, so those two name the exact call to

make instead of describing it.

"""

if not isinstance(expr, Query):

msg = (

f"q() expected a TinyDB Query from class-level field "

f"access like Model.field, got "

f"{type(expr).__name__!r}. {_q_hint(expr)}"

)

raise TypeError(msg)

return expr

def _queryable_field_names(model: type[TinydanticModel]) -> list[str]:

"""Return the field names ``field()`` accepts, sorted.

Computed fields are included: pydantic serializes them, so they

reach storage and are genuinely matchable. ``id`` is excluded:

it maps to ``doc_id`` and is never written to the document

body, so a query on it would match nothing.

"""

names = set(model.model_fields) | set(model.model_computed_fields)

return sorted(names - {"id"})

def _unknown_field_msg(model: type[Any], name: str) -> str:

"""Build the message for a name the model cannot query.

Shared by [field()][tinydantic.field] and the metaclass

``__getattr__``, so a typo reports the same way whether it

was written as a string or as an attribute.

"""

return (

f"{name!r} is not a queryable field of "

f"{model.__name__!r}. Names are Python field names (not "

f"storage aliases); queryable fields: "

f"{_queryable_field_names(model)}. For keys your model "

f"does not declare (extra='allow' documents, legacy "

f"keys), use tinydb.where({name!r})."

)

def field(model: type[TinydanticModel], name: str) -> Query:

"""Build a TinyDB Query on a model field by name.

The escape hatch for fields whose names collide with model

methods (``search``, ``get``, ``count``, ...) and are therefore

unreachable through the ``Model.field`` shorthand:

```python

Command.search(field(Command, "search") == "fuzzy")

```

The model is a build-time lookup table only — it never enters

the returned object, so the condition is a ``GuardedQuery`` on

that key alone: equal to (and hashing as) ``where(name)``, and

composing with raw queries and with ``Model.field`` conditions

freely. Exactly like ``Model.field`` access, it raises

[QueryTypeError][tinydantic.QueryTypeError] where TinyDB is

silently wrong (boolean context, iteration, non-string path

steps).

Accepted names are the model's fields plus its computed fields,

which pydantic serializes and which therefore reach storage.

Computed fields are also reachable through the ``Model.field``

shorthand, so ``field()`` is the escape hatch for them only when

the name is shadowed, exactly as for ordinary fields.

``id`` is refused (it maps to ``doc_id`` and is never written to

the document body — use ``Model.id`` for document-id queries),

as are storage aliases and dotted paths.

Validation covers the top-level name only. Chaining past it is

raw TinyDB, so a misspelled sub-field in

``field(User, "address").city`` is still silently wrong.

Args:

model: The model class the name belongs to.

name: A Python field name (never a storage alias).

Returns:

A Query on that document key.

Raises:

QueryFieldError: If ``name`` is not a queryable field of

``model``. Keys the model does not declare

(``extra="allow"`` documents, legacy keys) are reachable

with ``tinydb.where()``, which the message names.

"""

if name == "id":

msg = (

f"'id' is not a queryable field of {model.__name__!r}: it "

f"maps to TinyDB's doc_id and is never written to the "

f"document body. Use {model.__name__}.id for document-id "

f"queries (q({model.__name__}.id) == 1)."

)

raise QueryFieldError(msg)

if "." in name:

head, _, rest = name.partition(".")

msg = (

f"{name!r} is not a queryable field of "

f"{model.__name__!r}: where() does not traverse dotted "

f"paths, it queries a literal key containing a dot. "

f"Chain attributes instead: "

f"field({model.__name__}, {head!r}).{rest}"

)

raise QueryFieldError(msg)

if name not in model.model_fields and (

name not in model.model_computed_fields

):

raise QueryFieldError(_unknown_field_msg(model, name))

return GuardedQuery()[name]

class _ComputedFieldQuery:

"""Descriptor making a computed field queryable by attribute.

Computed fields are serialized, so they reach storage and are

genuinely matchable — but unlike ordinary fields they survive

class creation as real class attributes, so the metaclass

``__getattr__`` that turns ``Model.field`` into a query is never

consulted for them. Without this, ``Model.computed == value``

compares a ``property`` object to a value and quietly evaluates

to ``False``.

Descriptors see the difference the metaclass cannot: class-level

access calls ``__get__`` with ``obj=None``. Class access

therefore yields the query, and instance access delegates to the

wrapped property, leaving ``user.computed`` untouched.

"""

def __init__(self, prop: Any, name: str) -> None:

"""Wrap ``prop``, the computed field published as ``name``."""

self._prop = prop

self._name = name

self.__doc__ = prop.__doc__

def __get__(self, obj: Any, owner: Any = None) -> Any:

"""Return a Query on the class, the value on an instance."""

if obj is None:

return GuardedQuery()[self._name]

return self._prop.__get__(obj, owner)

def __set__(self, obj: Any, value: Any) -> None:

"""Delegate assignment to the wrapped property.

Pydantic's ``__setattr__`` does not use descriptor protocol,

so this is not reached by ordinary assignment — it is the

delegation target for

``TinydanticModel.__setattr__``, which routes computed-field

names here so the property's own setter (or its

``AttributeError``) decides the outcome.

"""

self._prop.__set__(obj, value)

def _install_computed_field_queries(model: type[Any]) -> None:

"""Wrap a class's computed fields as query descriptors.

Only names in the class's own namespace are wrapped; inherited

computed fields already resolve to the base class's descriptor,

which reports the same key. The descriptors are also collected

into ``__tinydantic_computed_queries__`` — resolved here, once,

so ``__setattr__`` costs a single dict lookup rather than an MRO

walk on every assignment.

Frozen models get an empty mapping: pydantic's frozen-instance

error is the right answer for every attribute of a frozen model,

computed or not, so assignment is left entirely to pydantic.

"""

queries: dict[str, _ComputedFieldQuery] = {}

for name in model.model_computed_fields:

prop = model.__dict__.get(name)

if isinstance(prop, property):

setattr(model, name, _ComputedFieldQuery(prop, name))

for klass in model.__mro__:

descriptor = klass.__dict__.get(name)

if isinstance(descriptor, _ComputedFieldQuery):

queries[name] = descriptor

break

frozen = model.model_config.get("frozen", False)

# Always assigned, never left to inheritance: a frozen subclass

# of an unfrozen base must not inherit the base's mapping.

model.__tinydantic_computed_queries__ = {} if frozen else queries

class TinydanticModelMetaclass(ModelMetaclass):

"""Metaclass providing class-level field queries.

Accessing a model *field* on the model *class* (not an instance)

returns a ``GuardedQuery`` on that field, so expressions like

``User.name == "Alice"`` build TinyDB queries directly from the

model definition. A guarded query tests, hashes, and compares

exactly like ``tinydb.queries.where(field_name)`` — it stays

interchangeable with raw TinyDB conditions — and differs only

in raising

[QueryTypeError][tinydantic.QueryTypeError] for the three

protocols TinyDB leaves silently wrong: boolean context,

iteration, and non-string path steps.

The ``id`` field is special-cased: it maps to TinyDB's

``doc_id``, which never appears in the stored document body, so

``Model.id`` returns a ``DocIdQuery`` (translated by the model's

query methods to document-id operations) instead of a

``where("id")`` body query that would silently match nothing.

Computed fields are queryable too, but cannot go through

``__getattr__``: pydantic leaves them as real class attributes,

so ordinary lookup succeeds and the hook never fires. They are

handled by ``_ComputedFieldQuery`` instead, installed after the

class is built.

Classes created with ``use_revision=True`` additionally get a

``revision_id: UUID | None`` field injected before pydantic

collects fields — the optimistic-concurrency token rotated by

every write path. Injection must happen here (not in

``__init_subclass__``): pydantic builds the field set from the

class namespace, so the field has to exist in the namespace

before class creation, which is also why ``use_revision``

cannot be late-bound with ``bind()``.

"""

def __new__(

mcs,

cls_name: str,

bases: tuple[type[Any], ...],

namespace: dict[str, Any],

**kwargs: Any,

) -> type:

"""Create the class, injecting ``revision_id`` if opted in.

Raises:

RevisionFieldError: If the class resolves

``use_revision=True`` (its own kwarg or inherited)

but declares its own ``revision_id`` field.

"""

explicit = kwargs.get("use_revision")

inherited = any(

get_config_value(base, "use_revision", default=False)

for base in bases

)

resolved = inherited if explicit is None else explicit

declared = "revision_id" in _declared_annotations(namespace)

if resolved and declared:

raise RevisionFieldError(model_name=cls_name)

already_a_field = any(

"revision_id" in getattr(base, "model_fields", {})

for base in bases

)

if explicit and not already_a_field:

_inject_revision_annotation(namespace)

namespace["revision_id"] = Field(

default=None,

description="Optimistic-concurrency token",

# The token is tinydantic's bookkeeping, not part

# of the caller's document: excluding it keeps it

# out of model_dump() and out of every FastAPI

# response built from the model, where it would

# otherwise appear as an unexplained field the

# ETag flow already carries in a header.

# SkipJsonSchema (on the annotation) is what keeps

# it out of the JSON Schema — field-level exclude

# does not.

exclude=True,

)

model = super().__new__(mcs, cls_name, bases, namespace, **kwargs)

_install_computed_field_queries(model)

return model

def __getattr__(cls, attr: str) -> Any:

"""Return a field Query, falling back to normal lookup.

Pydantic calls ``__getattr__`` for each field while the model

class is still being built; returning a Query then would break

model construction (``RuntimeError: Empty query was

evaluated``). The ``__pydantic_complete__`` guard defers query

behavior until the model is fully built.

"""

if cls.__pydantic_complete__ and attr in cls.model_fields:

if attr == "id":

# id maps to TinyDB's doc_id, which never appears

# in the document body — a plain where("id") query

# would silently match nothing.

return DocIdQuery()

return GuardedQuery()[attr]

if cls.__pydantic_complete__ and not attr.startswith("_"):

# A public name that is neither a field nor a real

# class attribute is a typo, and the bare

# "AttributeError: username" Python would raise names

# neither the model nor the alternatives. Underscored

# names are left alone: they are pydantic, copy, and

# pickle probing, never a user querying a field.

raise QueryFieldError(_unknown_field_msg(cls, attr))

return super().__getattr__(attr) # type: ignore[misc]

class TinydanticModel(BaseModel, metaclass=TinydanticModelMetaclass):

"""Base class for tinydantic models.

Subclass to define a document model, passing tinydantic

configuration as class keyword arguments:

```python

from tinydb import TinyDB

from tinydantic import TinydanticModel

db = TinyDB("db.json")

class User(TinydanticModel, database=db, table_name="users"):

name: str

```

Configuration is stored per class in ``__tinydantic_config__`` and

resolved by walking the MRO — deliberately NOT in pydantic's

``model_config``; see the ``tinydantic._config`` module docstring

for the design rationale (pydantic#9992).

"""

model_config: ClassVar[ConfigDict] = ConfigDict(

# Reserve the tinydantic_ prefix so future tinydantic methods

# cannot collide with user-defined fields (the use case Samuel

# Colvin described in pydantic#10315). protected_namespaces

# REPLACES the inherited value rather than extending it, so

# pydantic's own defaults are restated here — otherwise a

# tinydantic model would be less protected than the plain

# BaseModel the user would have written. Keep in sync with

# pydantic's default (test_model_config.py asserts it).

protected_namespaces=(

"tinydantic_",

"model_validate",

"model_dump",

),

# Assignment must not silently corrupt an instance that will

# later be persisted; subclasses may opt out via their own

# model_config.

validate_assignment=True,

)

__tinydantic_config__: ClassVar[TinydanticConfig] = {}

# Computed-field name -> query descriptor, resolved at class

# creation. Empty for models without computed fields (and for

# frozen ones), which is what keeps __setattr__ cheap.

__tinydantic_computed_queries__: ClassVar[

dict[str, _ComputedFieldQuery]

] = {}

def __setattr__(self, name: str, value: Any) -> None:

"""Set an attribute, routing computed fields to the property.

Class-level access to a computed field yields a Query — the

whole point of ``_ComputedFieldQuery`` — and pydantic's

assignment dispatch decides what to do by inspecting exactly

that (``attr = getattr(cls, name)``, then

``isinstance(attr, property)``). It therefore no longer

recognizes the property and reports the field as unknown:

"Object has no attribute 'shout'", of an attribute that

plainly exists. Routing the name here first restores the

property's own behavior — its setter if it has one, its

``AttributeError`` if it does not.

Args:

name: The attribute being assigned.

value: The value to assign.

"""

descriptor = type(self).__tinydantic_computed_queries__.get(name)

if descriptor is not None:

descriptor.__set__(self, value)

return

super().__setattr__(name, value)

# --- model fields ---

id: int | None = Field(

default=None,

description="Document ID",

)

if TYPE_CHECKING:

# Static-only declaration of the field the metaclass

# injects on use_revision=True models, so user code like

# ``book.revision_id`` (ETag flows) type-checks. The block

# never runs, so pydantic sees no field here and models

# without use_revision raise AttributeError at runtime.

revision_id: SkipJsonSchema[UUID | None] = None

# --- lifecycle hooks ---

def before_write(

self,

fields: Mapping[str, Any], # noqa: ARG002

) -> Mapping[str, Any] | None:

"""Contribute fields to any instance-level write of ``self``.

Fires once at the start of ``insert()``, ``save()``,

``replace()``, ``upsert()``, ``patch()``, and for each

document of ``insert_many()``. ``fields`` is the

model-field mapping about to be written, as validated

Python values — every field on a whole-model write, only

the caller's fields on ``patch()`` — and never contains

``id`` or ``revision_id``.

Return a mapping of fields to add or override (audit

timestamps are the classic case), or ``None`` to

contribute nothing. Returned values are validated,

persisted with the write, and set on this instance.

Always *return* the fields you want written — assigning to

``self`` here is silently dropped by ``patch()``, which

writes only named fields. Raising aborts the write with

nothing written.

The five update verbs — ``update()``, ``update_by_ids()``,

``update_all()``, ``update_many()``, and

``FindQuery.update()`` — apply a mapping to stored

documents with no instance in hand, so none of them call

this. The default does nothing; overrides can chain with

``super().before_write(fields)``.

"""

return None

def after_read(self) -> None:

"""Run after a document is validated into a model.

Fires at the end of

[from_tinydb_document][tinydantic.TinydanticModel.from_tinydb_document]

— i.e. after every read that materializes an instance —

with the real ``id`` set. Not called on ordinary

construction, and changes made here affect only the

in-memory instance (reads persist nothing). The default

does nothing; overrides can chain with

``super().after_read()``.

"""

# --- hook plumbing ---

def _write_fields(self) -> dict[str, Any]:

"""Every field a whole-model write will persist.

The mapping handed to

[before_write][tinydantic.TinydanticModel.before_write] for

writes that persist the entire model, as validated Python

values with the fields tinydantic manages removed.

"""

return {

name: getattr(self, name)

for name in type(self).model_fields

if name not in _MANAGED_FIELDS

}

def _run_before_write(

self,

fields: Mapping[str, Any],

) -> dict[str, Any]:

"""Call ``before_write()`` and apply what it contributed.

Validates the hook's returned mapping under the same rules

``patch()`` applies to caller input and hands it back for

the caller to merge into the write. Returns an empty

mapping when the hook contributed nothing.

Applying the result to the instance is a separate step

(``_apply_write_fields()``) so each caller can choose when

the instance changes: ``patch()`` applies only after its

write succeeds, while whole-model writes must apply first

because they serialize from ``self``.

Raises:

DocumentIDUpdateError: If the hook returned ``id``.

RevisionUpdateError: If the hook returned

``revision_id`` on a ``use_revision=True`` model.

UnknownUpdateFieldError: If the hook returned a key

that is not a model field.

"""

cls = type(self)

contributed = self.before_write(fields)

if not contributed:

return {}

unknown: list[str] = []

validated: dict[str, Any] = {}

for key, value in contributed.items():

if key == "id":

raise DocumentIDUpdateError(model_name=cls.__name__)

if key == "revision_id" and cls._uses_revision():

raise RevisionUpdateError(model_name=cls.__name__)

if key not in cls.model_fields:

unknown.append(key)

continue

validated[key] = cls._field_adapter(key).validate_python(value)

if unknown:

raise UnknownUpdateFieldError(

model_name=cls.__name__,

keys=unknown,

)

return validated

def _apply_write_fields(self, validated: Mapping[str, Any]) -> None:

"""Set already-validated field values on this instance.

Direct ``__dict__`` assignment, NOT ``setattr``: with

``validate_assignment`` on, per-field assignment can trip a

cross-field invariant on a transient state even though the

final state is valid. The write path validates the result.

"""

if not validated:

return

# The cast is for pyright; see _set_revision().

cast("dict[str, Any]", self.__dict__).update(validated)

self.__pydantic_fields_set__.update(validated)

def _before_write_whole_model(self) -> None:

"""Run the write hook for a write that persists everything.

Whole-model writes serialize from ``self``, so the hook's

contribution has to land on the instance before

serialization rather than after the write succeeds.

Models that never override the hook skip the whole thing:

building the field mapping is pure waste when the default

no-op is all it would be handed.

"""

if type(self).before_write is TinydanticModel.before_write:

return

self._apply_write_fields(

self._run_before_write(self._write_fields()),

)

# --- configuration ---

def __init_subclass__( # noqa: PLR0913

cls,

*,

database: TinyDB | None = None,

table_name: str | None = None,

use_revision: bool | None = None,

validate_writes: bool | None = None,

shadowed_fields: tuple[str, ...] | None = None,

constraints: tuple[UniqueConstraint, ...] | None = None,

**kwargs: Any,

) -> None:

"""Capture tinydantic class keywords.

Pydantic pops its own known config keys from class keyword

arguments and forwards the rest here (a public extension

point), so no metaclass involvement is needed for

configuration. Only explicitly provided values are stored on

this class — ``None`` means "not provided", and resolution

falls through to base classes via

``get_config_value`` in ``tinydantic._config``.

"""

super().__init_subclass__(**kwargs)

config: TinydanticConfig = {}

if database is not None:

config["database"] = database

if table_name is not None:

config["table_name"] = table_name

if use_revision is not None:

config["use_revision"] = use_revision

if validate_writes is not None:

config["validate_writes"] = validate_writes

if shadowed_fields is not None:

config["shadowed_fields"] = shadowed_fields

if constraints is not None:

# Validation waits for __pydantic_init_subclass__ —

# model_fields does not exist yet at this point.

config["constraints"] = constraints

setattr(cls, CONFIG_ATTR, config)

@classmethod

def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:

"""Validate config after pydantic finishes building the class.

Raises:

AmbiguousConfigError: If unrelated base classes supply

conflicting tinydantic config (see

``check_config_ambiguity`` in ``tinydantic._config``).

ShadowedFieldError: If a model field shadows an

existing class attribute (the ``Model.field``

query sugar would silently break) and is not

listed in the ``shadowed_fields=`` class kwarg.

ConstraintFieldError: If a ``constraints=`` entry names

``id`` or a name that is not a model field (see

``_validate_constraints``).

"""

super().__pydantic_init_subclass__(**kwargs)

check_config_ambiguity(cls)

allowed = get_config_value(cls, "shadowed_fields", default=())

shadowed: dict[str, str] = {}

for name in cls.model_fields:

if name == "id" or name in allowed:

# id is tinydantic's own (Model.id builds a

# DocIdQuery); allowed names are the documented

# opt-out.

continue

for klass in cls.__mro__:

if name in vars(klass):

# A real class attribute wins over the

# metaclass __getattr__ that builds queries,

# so this field's sugar is broken.

shadowed[name] = f"{klass.__name__}.{name}"

break

if shadowed:

raise ShadowedFieldError(

model_name=cls.__name__,

shadowed=shadowed,

)

cls._validate_constraints(

get_config_value(cls, "constraints", default=()) or (),

)

markers: list[UniqueConstraint] = []

for name, info in cls.model_fields.items():

for meta in info.metadata:

# The bare class is accepted alongside instances:

# treating Annotated[str, Unique] as inert would be

# a silent failure.

if meta is Unique:

markers.append(UniqueConstraint(name))

break

if isinstance(meta, Unique):

markers.append(

UniqueConstraint(name, key=meta.key),

)

break

setattr(cls, _UNIQUE_MARKERS_ATTR, tuple(markers))

@classmethod

def _validate_constraints(

cls,

constraints: tuple[UniqueConstraint, ...],

) -> None:

"""Reject constraints naming ``id`` or non-model fields.

Both failure modes would otherwise be *silent*: ``id`` is

never stored in the document body (it maps to TinyDB's

``doc_id``), so an id constraint would never match; an

unknown field reads as ``None`` in every body, so its

constraint would never enforce.

Raises:

ConstraintFieldError: On the first offending field.

"""

for constraint in constraints:

for name in constraint.fields:

if name == "id":

raise ConstraintFieldError(

model_name=cls.__name__,

constraint_fields=constraint.fields,

field=name,

reason="id",

)

if name not in cls.model_fields:

raise ConstraintFieldError(

model_name=cls.__name__,

constraint_fields=constraint.fields,

field=name,

reason="unknown",

)

@classmethod

def bind(

cls,

*,

database: TinyDB | None = None,

table_name: str | None = None,

validate_writes: bool | None = None,

shadowed_fields: tuple[str, ...] | None = None,

constraints: tuple[UniqueConstraint, ...] | None = None,

) -> None:

"""Bind or rebind tinydantic config after class definition.

The late-binding escape hatch for tests and application

factories where no TinyDB instance exists at import time:

```python

class User(TinydanticModel):

name: str

User.bind(database=TinyDB("db.json"))

```

Five of the six

[TinydanticConfig][tinydantic.TinydanticConfig] keys can be

bound late — ``database``, ``table_name``,

``validate_writes``, ``shadowed_fields``, and

``constraints``. ``use_revision`` cannot: it injects a

``revision_id`` field, and pydantic collects fields from

the class namespace during class creation, so it is

settable only as a class keyword.

Only the keys passed are updated; other keys keep their

current (possibly inherited) values. Binding a subclass

never affects its parents. The inverse is

[unbind][tinydantic.TinydanticModel.unbind].

Args:

database: The TinyDB instance to store documents in.

table_name: The table to use (defaults to the

snake_cased class name).

validate_writes: Whether write paths validate the

merged document before storing it.

shadowed_fields: Field names allowed to shadow class

attributes, queried with

[field()][tinydantic.field].

constraints: Multi-field uniqueness constraints.

Raises:

ConstraintFieldError: If a ``constraints`` entry names

``id`` or a name that is not a model field.

"""

config = cast(

"TinydanticConfig",

dict(cls.__dict__.get(CONFIG_ATTR, {})),

)

if database is not None:

config["database"] = database

if table_name is not None: