Skip to content

audit_events

Typed audit event models for the pending change queue.

Each queue lifecycle action has a dedicated event class that carries only the fields relevant to that action. All classes expose a to_audit_dict() method that flattens the event to the dict[str, Any] shape expected by AuditPayload.from_create().

PendingQueueAction

Bases: StrEnum

Lifecycle actions emitted by the pending queue.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
20
21
22
23
24
25
26
27
28
class PendingQueueAction(StrEnum):
    """Lifecycle actions emitted by the pending queue."""

    ENQUEUE = "enqueue"
    APPROVE = "approve"
    REJECT = "reject"
    APPLY = "apply"
    PURGE = "purge"
    BATCH_SPLIT = "batch_split"

ENQUEUE class-attribute instance-attribute 

ENQUEUE = 'enqueue'

APPROVE class-attribute instance-attribute 

APPROVE = 'approve'

REJECT class-attribute instance-attribute 

REJECT = 'reject'

APPLY class-attribute instance-attribute 

APPLY = 'apply'

PURGE class-attribute instance-attribute 

PURGE = 'purge'

BATCH_SPLIT class-attribute instance-attribute 

BATCH_SPLIT = 'batch_split'

_PendingQueueEventBase

Bases: BaseModel

Shared serialisation helper for all queue events.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class _PendingQueueEventBase(BaseModel):
    """Shared serialisation helper for all queue events."""

    def to_audit_dict(self) -> dict[str, Any]:
        """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
        data = self.model_dump(mode="json", exclude_none=True)
        # ``action`` must always be present
        data["action"] = self._action().value
        return data

    def to_audit_payload(self) -> AuditPayload:
        """Convert this event directly to an ``AuditPayload``."""
        return AuditPayload.from_create(self.to_audit_dict())

    def _action(self) -> PendingQueueAction:
        raise NotImplementedError

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Flatten the event to a plain dict suitable for AuditPayload.from_create().

Source code in src/horde_model_reference/pending_queue/audit_events.py 
34
35
36
37
38
39
def to_audit_dict(self) -> dict[str, Any]:
    """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
    data = self.model_dump(mode="json", exclude_none=True)
    # ``action`` must always be present
    data["action"] = self._action().value
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
45
46
def _action(self) -> PendingQueueAction:
    raise NotImplementedError

EnqueueEvent

Bases: _PendingQueueEventBase

A new change was submitted to the pending queue.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class EnqueueEvent(_PendingQueueEventBase):
    """A new change was submitted to the pending queue."""

    change_id: int
    operation: AuditOperation
    category: MODEL_REFERENCE_CATEGORY
    model_name: str = Field(serialization_alias="model")

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.ENQUEUE

    def to_audit_dict(self) -> dict[str, Any]:
        """Serialize enqueue-specific fields using enum values and the expected key names."""
        data = super().to_audit_dict()
        # The operation and category fields store enum *values* by convention
        data["operation"] = self.operation.value
        data["category"] = self.category.value
        # Use "model" key as expected by audit_view._process_enqueue
        data["model"] = data.pop("model_name", self.model_name)
        return data

change_id instance-attribute 

change_id: int

operation instance-attribute 

operation: AuditOperation

category instance-attribute 

category: MODEL_REFERENCE_CATEGORY

model_name class-attribute instance-attribute 

model_name: str = Field(serialization_alias='model')

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
57
58
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.ENQUEUE

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Serialize enqueue-specific fields using enum values and the expected key names.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
60
61
62
63
64
65
66
67
68
def to_audit_dict(self) -> dict[str, Any]:
    """Serialize enqueue-specific fields using enum values and the expected key names."""
    data = super().to_audit_dict()
    # The operation and category fields store enum *values* by convention
    data["operation"] = self.operation.value
    data["category"] = self.category.value
    # Use "model" key as expected by audit_view._process_enqueue
    data["model"] = data.pop("model_name", self.model_name)
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

ApproveEvent

Bases: _PendingQueueEventBase

A pending change was approved and assigned to a batch.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
71
72
73
74
75
76
77
78
79
class ApproveEvent(_PendingQueueEventBase):
    """A pending change was approved and assigned to a batch."""

    change_id: int
    batch_id: int | None
    batch_title: str

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.APPROVE

change_id instance-attribute 

change_id: int

batch_id instance-attribute 

batch_id: int | None

batch_title instance-attribute 

batch_title: str

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
78
79
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.APPROVE

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Flatten the event to a plain dict suitable for AuditPayload.from_create().

Source code in src/horde_model_reference/pending_queue/audit_events.py 
34
35
36
37
38
39
def to_audit_dict(self) -> dict[str, Any]:
    """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
    data = self.model_dump(mode="json", exclude_none=True)
    # ``action`` must always be present
    data["action"] = self._action().value
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

RejectEvent

Bases: _PendingQueueEventBase

A pending change was rejected.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
82
83
84
85
86
87
88
89
90
91
class RejectEvent(_PendingQueueEventBase):
    """A pending change was rejected."""

    change_id: int
    batch_id: int | None
    batch_title: str
    reason: str | None = None

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.REJECT

change_id instance-attribute 

change_id: int

batch_id instance-attribute 

batch_id: int | None

batch_title instance-attribute 

batch_title: str

reason class-attribute instance-attribute 

reason: str | None = None

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
90
91
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.REJECT

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Flatten the event to a plain dict suitable for AuditPayload.from_create().

Source code in src/horde_model_reference/pending_queue/audit_events.py 
34
35
36
37
38
39
def to_audit_dict(self) -> dict[str, Any]:
    """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
    data = self.model_dump(mode="json", exclude_none=True)
    # ``action`` must always be present
    data["action"] = self._action().value
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

ApplyEvent

Bases: _PendingQueueEventBase

An approved change was applied to the live dataset.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
 94
 95
 96
 97
 98
 99
100
101
102
class ApplyEvent(_PendingQueueEventBase):
    """An approved change was applied to the live dataset."""

    change_id: int
    batch_id: int | None
    job_id: str | None = None

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.APPLY

change_id instance-attribute 

change_id: int

batch_id instance-attribute 

batch_id: int | None

job_id class-attribute instance-attribute 

job_id: str | None = None

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
101
102
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.APPLY

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Flatten the event to a plain dict suitable for AuditPayload.from_create().

Source code in src/horde_model_reference/pending_queue/audit_events.py 
34
35
36
37
38
39
def to_audit_dict(self) -> dict[str, Any]:
    """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
    data = self.model_dump(mode="json", exclude_none=True)
    # ``action`` must always be present
    data["action"] = self._action().value
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

PurgeEvent

Bases: _PendingQueueEventBase

A queued change was removed without being applied.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class PurgeEvent(_PendingQueueEventBase):
    """A queued change was removed without being applied."""

    change_id: int
    category: MODEL_REFERENCE_CATEGORY
    model_name: str = Field(serialization_alias="model")
    requested_by: str
    purged_by_username: str

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.PURGE

    def to_audit_dict(self) -> dict[str, Any]:
        """Serialize purge-specific fields using enum values and the expected key names."""
        data = super().to_audit_dict()
        data["category"] = self.category.value
        data["model"] = data.pop("model_name", self.model_name)
        return data

change_id instance-attribute 

change_id: int

category instance-attribute 

category: MODEL_REFERENCE_CATEGORY

model_name class-attribute instance-attribute 

model_name: str = Field(serialization_alias='model')

requested_by instance-attribute 

requested_by: str

purged_by_username instance-attribute 

purged_by_username: str

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
114
115
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.PURGE

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Serialize purge-specific fields using enum values and the expected key names.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
117
118
119
120
121
122
def to_audit_dict(self) -> dict[str, Any]:
    """Serialize purge-specific fields using enum values and the expected key names."""
    data = super().to_audit_dict()
    data["category"] = self.category.value
    data["model"] = data.pop("model_name", self.model_name)
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())

BatchSplitEvent

Bases: _PendingQueueEventBase

Remaining approved changes were reassigned to a new batch after partial application.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
125
126
127
128
129
130
131
132
133
134
class BatchSplitEvent(_PendingQueueEventBase):
    """Remaining approved changes were reassigned to a new batch after partial application."""

    original_batch_id: int
    new_batch_id: int
    reassigned_change_ids: list[int]
    reason: str = "partial_apply"

    def _action(self) -> PendingQueueAction:
        return PendingQueueAction.BATCH_SPLIT

original_batch_id instance-attribute 

original_batch_id: int

new_batch_id instance-attribute 

new_batch_id: int

reassigned_change_ids instance-attribute 

reassigned_change_ids: list[int]

reason class-attribute instance-attribute 

reason: str = 'partial_apply'

_action 

_action() -> PendingQueueAction
Source code in src/horde_model_reference/pending_queue/audit_events.py 
133
134
def _action(self) -> PendingQueueAction:
    return PendingQueueAction.BATCH_SPLIT

to_audit_dict 

to_audit_dict() -> dict[str, Any]

Flatten the event to a plain dict suitable for AuditPayload.from_create().

Source code in src/horde_model_reference/pending_queue/audit_events.py 
34
35
36
37
38
39
def to_audit_dict(self) -> dict[str, Any]:
    """Flatten the event to a plain dict suitable for ``AuditPayload.from_create()``."""
    data = self.model_dump(mode="json", exclude_none=True)
    # ``action`` must always be present
    data["action"] = self._action().value
    return data

to_audit_payload 

to_audit_payload() -> AuditPayload

Convert this event directly to an AuditPayload.

Source code in src/horde_model_reference/pending_queue/audit_events.py 
41
42
43
def to_audit_payload(self) -> AuditPayload:
    """Convert this event directly to an ``AuditPayload``."""
    return AuditPayload.from_create(self.to_audit_dict())