events

Audit event models and type definitions for the append-only audit trail.

RecordLike module-attribute

RecordLike = Mapping[str, Any] | BaseModel

AuditOperation

Bases: StrEnum

CRUD operations captured by the audit log.

Source code in src/horde_model_reference/audit/events.py

class AuditOperation(StrEnum):
    """CRUD operations captured by the audit log."""

    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"

CREATE class-attribute instance-attribute

CREATE = 'create'

UPDATE class-attribute instance-attribute

UPDATE = 'update'

DELETE class-attribute instance-attribute

DELETE = 'delete'

AuditFieldChange

Bases: BaseModel

Represents a field-level delta for update operations.

Source code in src/horde_model_reference/audit/events.py

class AuditFieldChange(BaseModel):
    """Represents a field-level delta for update operations."""

    old: Any = Field(description="Previous value")
    new: Any = Field(description="New value")

old class-attribute instance-attribute

old: Any = Field(description='Previous value')

new class-attribute instance-attribute

new: Any = Field(description='New value')

AuditPayload

Bases: BaseModel

Payload recorded with an audit event (full snapshots or deltas).

Source code in src/horde_model_reference/audit/events.py

class AuditPayload(BaseModel):
    """Payload recorded with an audit event (full snapshots or deltas)."""

    before: dict[str, Any] | None = Field(default=None, description="Full record state prior to the change")
    after: dict[str, Any] | None = Field(default=None, description="Full record state after the change")
    delta: dict[str, AuditFieldChange] | None = Field(
        default=None,
        description="Sparse representation of changed fields for updates",
    )

    @overload
    @staticmethod
    def from_create(record: Mapping[str, Any]) -> AuditPayload: ...

    @overload
    @staticmethod
    def from_create(record: BaseModel) -> AuditPayload: ...

    @staticmethod
    def from_create(record: RecordLike) -> AuditPayload:
        """Build payload for create operations using the new record snapshot."""
        return AuditPayload(after=_coerce_record(record))

    @overload
    @staticmethod
    def from_delete(record: Mapping[str, Any]) -> AuditPayload: ...

    @overload
    @staticmethod
    def from_delete(record: BaseModel) -> AuditPayload: ...

    @staticmethod
    def from_delete(record: RecordLike) -> AuditPayload:
        """Build payload for delete operations using the removed record snapshot."""
        return AuditPayload(before=_coerce_record(record))

    @overload
    @staticmethod
    def from_update(before: Mapping[str, Any], after: Mapping[str, Any]) -> AuditPayload: ...

    @overload
    @staticmethod
    def from_update(before: BaseModel, after: BaseModel) -> AuditPayload: ...

    @staticmethod
    def from_update(before: RecordLike, after: RecordLike) -> AuditPayload:
        """Build payload for update operations using a sparse delta representation."""
        return AuditPayload(delta=_compute_delta(_coerce_record(before), _coerce_record(after)))

before class-attribute instance-attribute

before: dict[str, Any] | None = Field(
    default=None,
    description="Full record state prior to the change",
)

after class-attribute instance-attribute

after: dict[str, Any] | None = Field(
    default=None,
    description="Full record state after the change",
)

delta class-attribute instance-attribute

delta: dict[str, AuditFieldChange] | None = Field(
    default=None,
    description="Sparse representation of changed fields for updates",
)

from_create staticmethod

from_create(record: Mapping[str, Any]) -> AuditPayload

from_create(record: BaseModel) -> AuditPayload

from_create(record: RecordLike) -> AuditPayload

Build payload for create operations using the new record snapshot.

Source code in src/horde_model_reference/audit/events.py

@staticmethod
def from_create(record: RecordLike) -> AuditPayload:
    """Build payload for create operations using the new record snapshot."""
    return AuditPayload(after=_coerce_record(record))

from_delete staticmethod

from_delete(record: Mapping[str, Any]) -> AuditPayload

from_delete(record: BaseModel) -> AuditPayload

from_delete(record: RecordLike) -> AuditPayload

Build payload for delete operations using the removed record snapshot.

Source code in src/horde_model_reference/audit/events.py

@staticmethod
def from_delete(record: RecordLike) -> AuditPayload:
    """Build payload for delete operations using the removed record snapshot."""
    return AuditPayload(before=_coerce_record(record))

from_update staticmethod

from_update(
    before: Mapping[str, Any], after: Mapping[str, Any]
) -> AuditPayload

from_update(
    before: BaseModel, after: BaseModel
) -> AuditPayload

from_update(
    before: RecordLike, after: RecordLike
) -> AuditPayload

Build payload for update operations using a sparse delta representation.

Source code in src/horde_model_reference/audit/events.py

@staticmethod
def from_update(before: RecordLike, after: RecordLike) -> AuditPayload:
    """Build payload for update operations using a sparse delta representation."""
    return AuditPayload(delta=_compute_delta(_coerce_record(before), _coerce_record(after)))

AuditEvent

Bases: BaseModel

Single append-only audit event.

Source code in src/horde_model_reference/audit/events.py

class AuditEvent(BaseModel):
    """Single append-only audit event."""

    event_id: int
    timestamp: int = Field(description="Unix timestamp (UTC) when the event was recorded")
    domain: CanonicalFormat
    category: str
    model_name: str
    operation: AuditOperation
    logical_user_id: str = Field(description="Immutable Horde user identifier")
    request_id: str | None = Field(default=None, description="Optional idempotency or tracing identifier")
    payload: AuditPayload | None = Field(default=None, description="Snapshot or delta payload")

    @staticmethod
    def new(
        *,
        event_id: int,
        domain: CanonicalFormat,
        category: str,
        model_name: str,
        operation: AuditOperation,
        logical_user_id: str,
        timestamp: int | None = None,
        request_id: str | None = None,
        payload: AuditPayload | None = None,
    ) -> AuditEvent:
        """Create an audit event while filling defaults such as timestamp."""
        return AuditEvent(
            event_id=event_id,
            timestamp=timestamp or int(datetime.now(tz=UTC).timestamp()),
            domain=domain,
            category=category,
            model_name=model_name,
            operation=operation,
            logical_user_id=logical_user_id,
            request_id=request_id,
            payload=payload,
        )

event_id instance-attribute

event_id: int

timestamp class-attribute instance-attribute

timestamp: int = Field(
    description="Unix timestamp (UTC) when the event was recorded"
)

domain instance-attribute

domain: CanonicalFormat

category instance-attribute

category: str

model_name instance-attribute

model_name: str

operation instance-attribute

operation: AuditOperation

logical_user_id class-attribute instance-attribute

logical_user_id: str = Field(
    description="Immutable Horde user identifier"
)

request_id class-attribute instance-attribute

request_id: str | None = Field(
    default=None,
    description="Optional idempotency or tracing identifier",
)

payload class-attribute instance-attribute

payload: AuditPayload | None = Field(
    default=None, description="Snapshot or delta payload"
)

new staticmethod

new(
    *,
    event_id: int,
    domain: CanonicalFormat,
    category: str,
    model_name: str,
    operation: AuditOperation,
    logical_user_id: str,
    timestamp: int | None = None,
    request_id: str | None = None,
    payload: AuditPayload | None = None,
) -> AuditEvent

Create an audit event while filling defaults such as timestamp.

Source code in src/horde_model_reference/audit/events.py

@staticmethod
def new(
    *,
    event_id: int,
    domain: CanonicalFormat,
    category: str,
    model_name: str,
    operation: AuditOperation,
    logical_user_id: str,
    timestamp: int | None = None,
    request_id: str | None = None,
    payload: AuditPayload | None = None,
) -> AuditEvent:
    """Create an audit event while filling defaults such as timestamp."""
    return AuditEvent(
        event_id=event_id,
        timestamp=timestamp or int(datetime.now(tz=UTC).timestamp()),
        domain=domain,
        category=category,
        model_name=model_name,
        operation=operation,
        logical_user_id=logical_user_id,
        request_id=request_id,
        payload=payload,
    )

_coerce_record

_coerce_record(record: RecordLike) -> dict[str, Any]

Source code in src/horde_model_reference/audit/events.py

def _coerce_record(record: RecordLike) -> dict[str, Any]:
    if isinstance(record, BaseModel):
        return record.model_dump(mode="json")
    return {key: record[key] for key in record}

_compute_delta

_compute_delta(
    before: dict[str, Any], after: dict[str, Any]
) -> dict[str, AuditFieldChange]

Source code in src/horde_model_reference/audit/events.py

def _compute_delta(before: dict[str, Any], after: dict[str, Any]) -> dict[str, AuditFieldChange]:
    delta: dict[str, AuditFieldChange] = {}
    keys = set(before) | set(after)
    for key in keys:
        old_value = before.get(key)
        new_value = after.get(key)
        if old_value != new_value:
            delta[key] = AuditFieldChange(old=old_value, new=new_value)
    return delta