baldur.adapters.django — Django Adapter

Django-specific adapter pack. Provides BaldurConfig (AppConfig entry-point), RBAC group setup, statistics adapter, and abstract models / admin classes for DLQ + postmortem inventories.

django

Django Adapters Package for Baldur System.

This package provides Django-specific adapters for statistics and dashboards. Note: Runtime repositories use Redis (not Django ORM) for performance.

Provides: - BaldurConfig: Django AppConfig for Baldur system - create_baldur_groups: RBAC group creation signal handler - AbstractFailedOperation: Domain-free abstract model for DLQ entries - AbstractPostmortemRecord: Domain-free abstract model for Postmortem records - BasePostmortemRecordAdmin: Base Admin class for Postmortem records - BaseDLQEntryAdmin: Base Admin class for DLQ (FailedOperation) entries - BaseCircuitBreakerStateAdmin: Base Admin class for Circuit Breaker states - DjangoStatisticsAdapter: Statistics adapter using Django ORM - connect_session_signals: Wire up Django session signal handlers - disconnect_session_signals: Detach Django session signal handlers (test-only) - configure_baldur: Invoked from consumer settings.py to auto-wrap configuration

Status: Public

BaldurConfig

Bases: AppConfig

Django app configuration for baldur.

ready

ready()

Called when the app is ready (every server start).

Responsibilities: 1. Connect post_migrate signal for RBAC group creation 2. Log environment variable snapshot for audit trail 3. Validate config with Safe Defaults (Fail-Safe Default) 4. Hydrate metric gauges with jitter (Startup Hydration) 5. Start pre-computed cache worker (V3 Optimization)

Note: Environment snapshot is logged here (not in post_migrate) because env vars can change on every restart, not just during migrations. This aligns with 12-Factor App principles and Spring Boot patterns.

start_background_threads `classmethod`

start_background_threads()

Public entry point for Gunicorn post_worker_init hook.

Resets duplicate-start guards so threads can be started fresh in each forked Worker, then starts all background threads.

stop_background_threads `classmethod`

stop_background_threads()

Public entry point for Gunicorn worker_exit hook.

Resets the duplicate-start guard flags for the Django-tracked background threads. The threads are daemon=True and terminate when the worker process exits, so there is nothing to join here; in-flight work is drained upstream by the ShutdownCoordinator wait in the worker_exit hook (baldur.adapters.gunicorn.hooks) that runs before this call.

reset_hydration_state `classmethod`

reset_hydration_state()

Hydration state reset (for testing).

DjangoStatisticsAdapter

DjangoStatisticsAdapter(
    failed_operation_model: (
        type[AbstractFailedOperation] | None
    ) = None,
    circuit_breaker_model: type[Model] | None = None,
)

Bases: StatisticsRepositoryInterface

Django ORM implementation of StatisticsRepositoryInterface.

This adapter is domain-free - models are provided by the application. It uses Django's powerful ORM for complex aggregate queries.

Attributes:

Name	Type	Description
`failed_operation_model`		Django model for DLQ entries
`circuit_breaker_model`		Django model for CB state (optional)

Initialize Django Statistics Adapter.

Parameters:

Name	Type	Description	Default
`failed_operation_model`	`type[AbstractFailedOperation] \| None`	Concrete model inheriting from AbstractFailedOperation (manager + fields visible to django-stubs).	`None`
`circuit_breaker_model`	`type[Model] \| None`	User-provided model class for CB state. No abstract base is shipped — caller's choice of concrete model.	`None`

get_status_counts

get_status_counts() -> StatusCounts

Get count of DLQ entries by status using Django aggregation.

get_domain_distribution

get_domain_distribution(
    limit: int = 10,
) -> list[DomainDistribution]

Get distribution of DLQ entries by domain.

get_failure_type_distribution

get_failure_type_distribution(
    limit: int = 10,
) -> list[FailureTypeDistribution]

Get distribution of DLQ entries by failure type.

get_recent_activity

get_recent_activity(
    hours: int = 24, days: int = 7
) -> RecentActivity

Get recent activity statistics.

get_resolution_rate

get_resolution_rate(days: int = 30) -> float

Calculate resolution success rate.

get_avg_retry_count

get_avg_retry_count() -> float

Get average retry count across all DLQ entries.

list_entries

list_entries(
    page: int = 1,
    page_size: int = 20,
    status: str | None = None,
    domain: str | None = None,
    failure_type: str | None = None,
    order_by: str = "-created_at",
) -> PaginatedResult

List DLQ entries with pagination and filtering.

get_entry_detail

get_entry_detail(entry_id: str) -> dict[str, Any] | None

Get detailed information about a specific DLQ entry.

get_sla_breaches

get_sla_breaches(
    sla_threshold_hours: int = 4,
    statuses: list[str] | None = None,
) -> dict[str, int]

Get count of SLA breaches by domain.

Finds DLQ entries that have exceeded the SLA threshold for resolution.

get_cleanup_stats

get_cleanup_stats() -> CleanupStats

Get statistics for cleanup operations.

archive_old_entries

archive_old_entries(older_than_days: int = 30) -> int

Archive old resolved entries.

purge_archived

purge_archived(
    ids: list[str] | None = None,
    older_than_days: int | None = None,
) -> int

Permanently delete archived entries.

get_circuit_breaker_summary

get_circuit_breaker_summary() -> CircuitBreakerSummary

Get summary of all circuit breakers.

list_circuit_breakers

list_circuit_breakers() -> list[CircuitBreakerInfo]

List all circuit breakers with their current state.

persist_entry

persist_entry(entry_data: dict[str, Any]) -> str | None

Persist a DLQ entry to the statistics store.

sync_from_runtime

sync_from_runtime(entries: list[dict[str, Any]]) -> int

Bulk sync entries from runtime repository.

get_audit_trail_by_entity

get_audit_trail_by_entity(
    entity_id: str, entity_type: str = "dlq_entry"
) -> EntityAuditTrail

Get complete audit trail for a specific entity.

Retrieves all audit log entries related to a DLQ entry and verifies the hash chain integrity.

link_audit_entry

link_audit_entry(
    entity_id: str,
    entity_type: str,
    action: str,
    actor_id: str | None = None,
    status: str | None = None,
    details: str | None = None,
    audit_record_hash: str | None = None,
) -> bool

Link an audit record to an entity.

Creates a mapping between DLQ entries and their audit records for efficient trail retrieval.

should_persist_async

should_persist_async() -> bool

Check if async persistence is configured.

Reads from Django settings: BALDUR_ASYNC_PERSISTENCE = True

get_async_persist_task_name

get_async_persist_task_name() -> str | None

Get the Celery task name for async persistence.

configure_baldur

configure_baldur(
    namespace: dict,
    *,
    early_group: list[str] | None = None,
    post_auth_group: list[str] | None = None,
    tail_group: list[str] | None = None,
    domains: list[str] | None = None,
    disable_auto_otel: bool = False
) -> None

Called from consumer settings.py to wrap baldur configuration explicitly.

Parameters:

Name	Type	Description	Default
`namespace`	`dict`	globals() of the consumer settings module — modifies MIDDLEWARE, REST_FRAMEWORK, etc.	required
`early_group`	`list[str] \| None`	List of middleware to insert before Django core (default: DEFAULT_EARLY_GROUP)	`None`
`post_auth_group`	`list[str] \| None`	List of middleware to insert after AuthenticationMiddleware (default: DEFAULT_POST_AUTH_GROUP)	`None`
`tail_group`	`list[str] \| None`	List of middleware to insert at the tail (default: DEFAULT_TAIL_GROUP)	`None`
`domains`	`list[str] \| None`	BALDUR_CORE_DOMAINS setting (list of business domains)	`None`
`disable_auto_otel`	`bool`	If True, skip OTEL-related setup (used when deferring to a Gunicorn hook)	`False`

Note

This function MUST be called at the very bottom of settings.py. It must run after MIDDLEWARE, REST_FRAMEWORK, etc. are defined to work correctly.

create_baldur_groups

create_baldur_groups(sender, **kwargs)

Create RBAC groups for Baldur system.

Called via post_migrate signal - runs only after migrations complete. Uses get_or_create for idempotency.

Note: Environment variable snapshot is logged in ready() instead, because env vars can change on every restart (not just migrations).

get_abstract_failed_operation

get_abstract_failed_operation() -> (
    type[AbstractFailedOperation]
)

Get AbstractFailedOperation model class.

This is a lazy import to avoid Django dependency at module load time.

Returns:

Type	Description
`type[AbstractFailedOperation]`	AbstractFailedOperation class

Raises:

Type	Description
`ImportError`	If Django is not installed

get_abstract_postmortem_record

get_abstract_postmortem_record() -> (
    type[AbstractPostmortemRecord]
)

Get AbstractPostmortemRecord model class.

This is a lazy import to avoid Django dependency at module load time.

Returns:

Type	Description
`type[AbstractPostmortemRecord]`	AbstractPostmortemRecord class

Raises:

Type	Description
`ImportError`	If Django is not installed

get_base_postmortem_admin

get_base_postmortem_admin() -> (
    type[BasePostmortemRecordAdmin]
)

Get BasePostmortemRecordAdmin class.

This is a lazy import to avoid Django dependency at module load time.

Returns:

Type	Description
`type[BasePostmortemRecordAdmin]`	BasePostmortemRecordAdmin class

Raises:

Type	Description
`ImportError`	If Django is not installed

get_base_dlq_admin

get_base_dlq_admin() -> type[BaseDLQEntryAdmin]

Get BaseDLQEntryAdmin class.

This is a lazy import to avoid Django dependency at module load time.

Returns:

Type	Description
`type[BaseDLQEntryAdmin]`	BaseDLQEntryAdmin class

Raises:

Type	Description
`ImportError`	If Django is not installed

get_base_circuit_breaker_admin

get_base_circuit_breaker_admin() -> (
    type[BaseCircuitBreakerStateAdmin]
)

Get BaseCircuitBreakerStateAdmin class.

This is a lazy import to avoid Django dependency at module load time.

Returns:

Type	Description
`type[BaseCircuitBreakerStateAdmin]`	BaseCircuitBreakerStateAdmin class

Raises:

Type	Description
`ImportError`	If Django is not installed

get_abstract_failed_external_request

get_abstract_failed_external_request() -> (
    type[AbstractFailedExternalRequest]
)

Get AbstractFailedExternalRequest model class (223 Host App Decoupling).

Returns:

Type	Description
`type[AbstractFailedExternalRequest]`	AbstractFailedExternalRequest class

get_abstract_security_incident

get_abstract_security_incident() -> (
    type[AbstractSecurityIncident]
)

Get AbstractSecurityIncident model class (223 Host App Decoupling).

Returns:

Type	Description
`type[AbstractSecurityIncident]`	AbstractSecurityIncident class

baldur.adapters.django — Django Adapter

django

BaldurConfig

ready

start_background_threads classmethod

stop_background_threads classmethod

reset_hydration_state classmethod

DjangoStatisticsAdapter

get_status_counts

get_domain_distribution

get_failure_type_distribution

get_recent_activity

get_resolution_rate

get_avg_retry_count

list_entries

get_entry_detail

get_sla_breaches

get_cleanup_stats

archive_old_entries

purge_archived

get_circuit_breaker_summary

list_circuit_breakers

persist_entry

sync_from_runtime

get_audit_trail_by_entity

link_audit_entry

should_persist_async

get_async_persist_task_name

configure_baldur

create_baldur_groups

get_abstract_failed_operation

get_abstract_postmortem_record

get_base_postmortem_admin

get_base_dlq_admin

get_base_circuit_breaker_admin

get_abstract_failed_external_request

get_abstract_security_incident

start_background_threads `classmethod`

stop_background_threads `classmethod`

reset_hydration_state `classmethod`