Skip to content

baldur.interfaces — Incident & Archive Repositories

Repository interfaces for security incidents, postmortems, and the long-term archives (cascade events, recovery sessions). The shared enums and DTOs live on the data-model page.

SecurityIncidentRepository

Bases: ABC

Abstract repository for SecurityIncident data access.

Security incidents are stored separately and NEVER auto-replayed.

create abstractmethod 

create(
    incident_type: str,
    severity: str,
    description: str = "",
    source_ip: str | None = None,
    user_agent: str = "",
    user_id: int | None = None,
    entity_refs: dict[str, int] | None = None,
    raw_payload: dict[str, Any] | None = None,
) -> SecurityIncidentData

Create a new security incident

get_by_id abstractmethod 

get_by_id(id: int) -> SecurityIncidentData | None

Get a security incident by ID

get_open_incidents abstractmethod 

get_open_incidents(
    limit: int = 100,
) -> list[SecurityIncidentData]

Get all open (unresolved) incidents

get_by_type abstractmethod 

get_by_type(
    incident_type: str, limit: int = 100
) -> list[SecurityIncidentData]

Get incidents by type

get_by_severity abstractmethod 

get_by_severity(
    severity: str, limit: int = 100
) -> list[SecurityIncidentData]

Get incidents by severity

update_status abstractmethod 

update_status(
    id: int,
    status: str,
    investigation_notes: str = "",
    assigned_to_id: int | None = None,
) -> bool

Update incident status

mark_as_resolved abstractmethod 

mark_as_resolved(
    id: int, investigation_notes: str = ""
) -> bool

Mark incident as resolved

get_recent_by_ip abstractmethod 

get_recent_by_ip(
    source_ip: str, hours: int = 24, limit: int = 100
) -> list[SecurityIncidentData]

Get recent incidents from a specific IP

count_by_type_since abstractmethod 

count_by_type_since(
    incident_type: str, since: datetime
) -> int

Count incidents of a type since a given time

PostmortemRepository

Bases: ABC

Abstract repository for Postmortem data access.

Postmortem records are immutable audit artifacts. Supports hybrid storage (DB primary + In-Memory cache).

save abstractmethod 

save(data: PostmortemData) -> bool

Persist a postmortem record.

Returns True if saved successfully, False otherwise. Implementations should handle duplicate incident_id gracefully: - DB-backed (Django): skip duplicate, return False (UNIQUE constraint) - InMemory: overwrite existing, return True (no constraint)

get_by_incident_id abstractmethod 

get_by_incident_id(
    incident_id: str,
) -> PostmortemData | None

Retrieve a single postmortem by incident ID.

find abstractmethod 

find(
    *,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    service: str | None = None,
    min_duration: float | None = None,
    offset: int = 0,
    limit: int = 100
) -> list[PostmortemData]

Query postmortems with optional filters.

Results ordered by started_at DESC.

count abstractmethod 

count(
    *,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    service: str | None = None,
    min_duration: float | None = None
) -> int

Count postmortems matching filters.

update_fields abstractmethod 

update_fields(
    incident_id: str, fields: dict[str, Any]
) -> bool

Partial update of specific fields.

JSONField values are deep-merged with existing data. Returns True if the record was found and updated.

CascadeEventArchiveRepository

Bases: ABC

Cascade Event archive persistence.

save abstractmethod 

save(data: CascadeEventData) -> bool

Persist a cascade event record.

Idempotent — Django: IntegrityError → False, InMemory: overwrite → True.

get_by_cascade_id abstractmethod 

get_by_cascade_id(
    cascade_id: str,
) -> CascadeEventData | None

Retrieve a single cascade event by ID.

find abstractmethod 

find(
    *,
    namespace: str | None = None,
    trigger_type: str | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    is_test: bool | None = None,
    offset: int = 0,
    limit: int = 100
) -> list[CascadeEventData]

Query with optional filters. Results ordered by timestamp DESC.

count abstractmethod 

count(
    *,
    namespace: str | None = None,
    trigger_type: str | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> int

Count cascade events matching filters.

delete_older_than abstractmethod 

delete_older_than(cutoff: datetime) -> int

Delete archived events older than cutoff. Returns deleted count.

get_chain abstractmethod 

get_chain(
    namespace: str,
    *,
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> list[CascadeEventData]

Retrieve hash chain for integrity verification. Ordered by timestamp ASC.

RecoverySessionArchiveRepository

Bases: ABC

Recovery Session archive persistence.

save abstractmethod 

save(data: RecoverySessionData) -> bool

Persist a recovery session record.

Idempotent — Django: IntegrityError → False, InMemory: overwrite → True.

get_by_session_id abstractmethod 

get_by_session_id(
    session_id: str,
) -> RecoverySessionData | None

Retrieve a single session by ID.

find abstractmethod 

find(
    *,
    namespace: str | None = None,
    status: str | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    offset: int = 0,
    limit: int = 100
) -> list[RecoverySessionData]

Query with optional filters. Results ordered by started_at DESC.

count abstractmethod 

count(
    *,
    namespace: str | None = None,
    status: str | None = None
) -> int

Count sessions matching filters.

update abstractmethod 

update(data: RecoverySessionData) -> bool

Full update of an existing session record.

Returns True if record found and updated.

delete_older_than abstractmethod 

delete_older_than(cutoff: datetime) -> int

Delete archived sessions older than cutoff. Returns deleted count.