baldur.interfaces — Repository Data Model
The enums and data classes shared across the repository interfaces — the status and domain enums, plus the operation/state/incident/postmortem DTOs.
Enums
DLQCompressedStatus
Bases: str, Enum
Lifecycle state machine for compressed DLQ entries.
FailedOperationDomain
Bases: str, Enum
Domain classification for failed operations (domain-neutral).
Core domains are framework-agnostic. Application-specific domains (like 'payment', 'order') should be registered via adapter configuration.
FailedOperationStatus
Bases: str, Enum
State machine for DLQ item lifecycle
CircuitBreakerStateEnum
Bases: str, Enum
Circuit breaker states
SecurityIncidentType
Bases: str, Enum
Types of security incidents (domain-neutral)
SecuritySeverity
Bases: str, Enum
Severity levels for security incidents
SecurityIncidentStatus
Bases: str, Enum
Investigation status for security incidents
Data classes
FailedOperationData
dataclass
FailedOperationData(
id: str,
domain: str,
failure_type: str,
status: str,
entity_type: str | None = None,
entity_id: str | None = None,
entity_refs: dict[str, Any] = dict(),
user_id: int | None = None,
snapshot_data: dict[str, Any] = dict(),
error_code: str = "",
error_message: str = "",
retry_count: int = 0,
max_retries: int = 2,
last_retry_at: datetime | None = None,
request_data: dict[str, Any] = dict(),
response_data: dict[str, Any] = dict(),
metadata: dict[str, Any] = dict(),
resolved_at: datetime | None = None,
resolved_by_id: int | None = None,
resolution_type: str = "",
resolution_note: str = "",
next_action_hint: str = "",
recommended_action: str = "",
created_at: datetime | None = None,
updated_at: datetime | None = None,
expires_at: datetime | None = None,
)
Data transfer object for FailedOperation model.
Contains all necessary fields for DLQ operations without Django model dependencies.
is_pending
property
is_pending: bool
Check if operation is pending review
is_resolved
property
is_resolved: bool
Check if operation is resolved
can_retry
property
can_retry: bool
Check if operation can be retried
CircuitBreakerStateData
dataclass
CircuitBreakerStateData(
service_name: str,
id: int | None = None,
state: str = CircuitBreakerStateEnum.CLOSED.value,
failure_count: int = 0,
success_count: int = 0,
last_failure_at: datetime | None = None,
opened_at: datetime | None = None,
manually_controlled: bool = False,
controlled_by_id: int | None = None,
control_reason: str = "",
manual_override_expires_at: datetime | None = None,
half_open_request_count: int = 0,
half_open_window_started_at: datetime | None = None,
metadata: dict[str, Any] = dict(),
created_at: datetime | None = None,
updated_at: datetime | None = None,
)
Data transfer object for CircuitBreakerState model.
Represents the current state of a circuit breaker for a service.
is_open
property
is_open: bool
Check if circuit is open (blocking requests)
is_closed
property
is_closed: bool
Check if circuit is closed (allowing requests)
is_half_open
property
is_half_open: bool
Check if circuit is half-open (testing)
SecurityIncidentData
dataclass
SecurityIncidentData(
id: int,
incident_type: str,
severity: str,
status: str,
source_ip: str | None = None,
user_agent: str = "",
user_id: int | None = None,
entity_refs: dict[str, int] = dict(),
description: str = "",
raw_payload: dict[str, Any] = dict(),
assigned_to_id: int | None = None,
investigation_notes: str = "",
resolved_at: datetime | None = None,
created_at: datetime | None = None,
updated_at: datetime | None = None,
)
Data transfer object for SecurityIncident model.
Security incidents are NEVER auto-healed and require human intervention.
is_critical
property
is_critical: bool
Check if incident is critical severity
needs_investigation
property
needs_investigation: bool
Check if incident needs investigation
PostmortemData
dataclass
PostmortemData(
incident_id: str,
started_at: datetime | None = None,
resolved_at: datetime | None = None,
duration_seconds: float = 0.0,
affected_services: list[str] = list(),
timeline: list[dict[str, Any]] = list(),
auto_actions: list[dict[str, Any]] = list(),
recommendations: list[str] = list(),
system_snapshot: dict[str, Any] = dict(),
created_at: datetime | None = None,
source: str = "auto",
id: str = "",
)
Bases: SerializableMixin
Framework-agnostic Postmortem domain model.
Replaces AbstractPostmortemRecord's data role. Django Model remains as adapter-internal persistence detail.
incident_id
instance-attribute
incident_id: str
Unique incident identifier.
started_at
class-attribute
instance-attribute
started_at: datetime | None = None
Incident start time.
resolved_at
class-attribute
instance-attribute
resolved_at: datetime | None = None
Incident resolution time.
duration_seconds
class-attribute
instance-attribute
duration_seconds: float = 0.0
Incident duration in seconds.
affected_services
class-attribute
instance-attribute
affected_services: list[str] = field(default_factory=list)
List of services affected by the incident.
timeline
class-attribute
instance-attribute
timeline: list[dict[str, Any]] = field(default_factory=list)
Chronological event timeline.
auto_actions
class-attribute
instance-attribute
auto_actions: list[dict[str, Any]] = field(
default_factory=list
)
Automatic recovery actions taken.
recommendations
class-attribute
instance-attribute
recommendations: list[str] = field(default_factory=list)
Recommended remediation steps.
system_snapshot
class-attribute
instance-attribute
system_snapshot: dict[str, Any] = field(
default_factory=dict
)
System state snapshot at incident time.
created_at
class-attribute
instance-attribute
created_at: datetime | None = None
Record creation time.
source
class-attribute
instance-attribute
source: str = 'auto'
Record origin: 'auto' (system-generated) or 'manual'.
id
class-attribute
instance-attribute
id: str = ''
UUID string identifier. Auto-generated if empty.
from_incident_dict
classmethod
from_incident_dict(
incident_data: dict[str, Any],
) -> PostmortemData
Create from raw incident dictionary (In-Memory format).
Handles ISO date parsing and field extraction — replaces AbstractPostmortemRecord.create_from_incident_dict().