The domain layer (forze.domain) holds business logic and data structures. It has no knowledge of databases, HTTP, or external services. For the conceptual overview, see Domain Layer. This page is the complete API reference.

Base models

CoreModel

Base for all domain models. Extends Pydantic's BaseModel with:

• Field docstrings enabled for schema generation
• Stripped string fields
• Sorted set serialization for stable JSON output

For example:

from forze.domain.models import CoreModel

class Settings(CoreModel):
    name: str
    tags: set[str] = set()

BaseDTO

Extends CoreModel with frozen semantics (immutable after creation). Use for command DTOs, update payloads, and read projections:

from forze.domain.models import BaseDTO

class UpdateProjectCmd(BaseDTO):
    title: str | None = None
    description: str | None = None

Document

Document is the base class for versioned aggregates. Every document carries identity, revision tracking, and timestamps.

from forze.domain.models import Document

class Project(Document):
    title: str
    description: str = ""

Built-in fields

Frozen fields raise ValidationError if an update attempts to change them.

update()

Apply a validated update and return the new document plus a computed diff:

project = Project(title="Alpha", description="First")
updated, diff = project.update({"title": "Beta"})
# updated.title == "Beta"
# diff == {"title": "Beta", "last_update_at": <timestamp>}

The update flow:

1. Validate — reject unknown and frozen fields
2. Compute diff — apply the patch to a JSON dump, calculate the minimal merge patch
3. Bump timestamp — set last_update_at to now
4. Run validators — execute registered @update_validator hooks
5. Return — new immutable copy and the diff

If the patch produces no changes, returns the original instance and an empty dict.

touch()

Update only last_update_at without changing other fields:

touched, diff = project.touch()
# diff == {"last_update_at": <timestamp>}

validate_historical_consistency()

Check whether applying a patch to an older state would conflict with the current state:

is_safe = current.validate_historical_consistency(
    old_state,
    incoming_patch,
)

Returns True when the incoming patch does not touch fields that changed between old_state and current. Used by adapters reconstructing state from history to prevent conflicting concurrent merges.

Commands and read models

CreateDocumentCmd

Base for create commands. Frozen DTO that optionally accepts id and created_at for imports and migrations:

from forze.domain.models import CreateDocumentCmd

class CreateProjectCmd(CreateDocumentCmd):
    title: str
    description: str

ReadDocument

Base for read projections returned by ports. Includes the standard document fields:

from forze.domain.models import ReadDocument

class ProjectRead(ReadDocument):
    title: str
    description: str

DocumentHistory

Stores a snapshot of a document at a given revision:

from forze.domain.models import DocumentHistory

history_entry: DocumentHistory[Project]

Update validators

Validators enforce business rules during Document.update(). Decorate methods with @update_validator to register them:

from forze.domain.validation import update_validator
from forze.base.errors import ValidationError

class Project(Document):
    title: str
    status: str = "draft"

    @update_validator
    def _block_published(before, after, diff):
        if before.status == "published" and "title" in diff:
            raise ValidationError("Cannot change title after publishing.")

Validator signatures

The decorator normalizes three function signatures:

Field-scoped validators

Restrict a validator to specific fields with the fields parameter. The validator only runs when the diff touches at least one of the listed fields:

@update_validator(fields={"status"})
def _validate_transition(before, after, diff):
    allowed = {"draft": {"active"}, "active": {"archived"}}
    if after.status not in allowed.get(before.status, set()):
        raise ValidationError("Invalid status transition.")

Validator collection

Validators are collected from the full class hierarchy at class definition time via collect_update_validators(). Conflict resolution when a subclass redefines a validator name:

Override the default by setting _update_validators_on_conflict on the class:

class StrictProject(Document):
    _update_validators_on_conflict = "error"

Mixins

Reusable domain concerns composed via multiple inheritance. Each mixin adds a focused capability.

SoftDeletionMixin

Adds an is_deleted boolean and a validator that blocks updates to soft-deleted documents (except toggling the flag itself):

from forze.domain.mixins import SoftDeletionMixin

class Project(SoftDeletionMixin, Document):
    title: str

Once is_deleted is True, updating any field other than is_deleted raises ValidationError.

NameMixin

Adds a required name and optional display_name, short_name, and description:

from forze.domain.mixins import (
    NameMixin,
    NameCreateCmdMixin,
    NameUpdateCmdMixin,
)

class Workspace(NameMixin, Document):
    pass

class CreateWorkspaceCmd(NameCreateCmdMixin, CreateDocumentCmd):
    pass

class UpdateWorkspaceCmd(NameUpdateCmdMixin, BaseDTO):
    pass

NumberMixin

Adds a required positive integer number_id for human-readable identification. Typically populated by a counter adapter during the create mapping step:

from forze.domain.mixins import NumberMixin, NumberCreateCmdMixin

class Ticket(NumberMixin, Document):
    title: str

class CreateTicketCmd(NumberCreateCmdMixin, CreateDocumentCmd):
    title: str

CreatorMixin

Adds a frozen creator_id field (UUID). Typically injected by a mapping step that reads the current actor context:

from forze.domain.mixins import CreatorMixin, CreatorCreateCmdMixin

class Comment(CreatorMixin, Document):
    body: str

class CreateCommentCmd(CreatorCreateCmdMixin, CreateDocumentCmd):
    body: str

Domain constants

String constants used across layers for consistent field naming:

from forze.domain.constants import ID_FIELD, REV_FIELD, SOFT_DELETE_FIELD