Skip to content

baldur.services — Circuit Breaker & Rate Limiting

Circuit-breaker configuration and result types, the manual open/close controls, the breaker state enum, and the rate-limit tracker.

Configuration & results

CircuitBreakerConfig dataclass 

CircuitBreakerConfig(
    enabled: bool = False,
    failure_threshold: int = 5,
    recovery_timeout: int = 60,
    success_threshold: int = 2,
    minimum_calls: int = 10,
    sliding_window_size: int = 100,
    failure_rate_threshold: float = 50.0,
    fallback_strategy: str = "cache",
    fallback_cache_ttl_seconds: int = 300,
    cb_open_burn_rate_multiplier: float = 10.0,
    cb_open_base_consumption_minutes: float = 1.0,
    manual_override_ttl_minutes: int = 90,
    half_open_max_calls: int = 3,
    max_pending_duration_hours: int = 4,
    max_retry_lifetime_hours: int = 24,
    rate_limit_cascade_threshold: int = 10,
    rate_limit_cascade_window_seconds: int = 60,
    rate_limit_cascade_rate: float = 10.0,
    rate_limit_cascade_minimum_calls: int = 20,
    self_ddos_protection_enabled: bool = True,
    self_ddos_rps_limit: int = 200,
    self_ddos_window_seconds: int = 10,
    self_ddos_backoff_multiplier: float = 2.0,
    rate_limit_distributed: bool = False,
)

Configuration for circuit breaker operations.

from_settings classmethod 

from_settings() -> CircuitBreakerConfig

Load configuration from RuntimeConfigManager (preferred) or core config.

CircuitBreakerResult dataclass 

CircuitBreakerResult(
    success: bool,
    service_name: str,
    previous_state: str = "",
    new_state: str = "",
    message: str = "",
    error: str | None = None,
)

Result of a circuit breaker operation.

succeeded classmethod 

succeeded(
    service_name: str,
    previous_state: str,
    new_state: str,
    message: str = "",
) -> CircuitBreakerResult

Factory for successful operation.

failed classmethod 

failed(
    service_name: str, error: str
) -> CircuitBreakerResult

Factory for failed operation.

CircuitState

Bases: str, Enum

Circuit breaker states

Manual controls

should_allow_request 

should_allow_request(service_name: str) -> bool

Convenience function to check if requests should be allowed.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
bool

True if requests should be allowed

force_open_circuit 

force_open_circuit(
    service_name: str, reason: str = ""
) -> CircuitBreakerResult

Convenience function to force open a circuit breaker.

Actor information is read from ActorContext (set by Django middleware or Celery task signal handlers).

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required
reason str

Reason for opening

 ''

Returns:

Type Description
CircuitBreakerResult

CircuitBreakerResult with operation outcome

force_close_circuit 

force_close_circuit(
    service_name: str,
    reason: str = "",
    trigger_replay: bool = False,
) -> CircuitBreakerResult

Convenience function to force close a circuit breaker.

Actor information is read from ActorContext (set by Django middleware or Celery task signal handlers).

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required
reason str

Reason for closing

 ''
trigger_replay bool

Whether to trigger conditional replay

 False

Returns:

Type Description
CircuitBreakerResult

CircuitBreakerResult with operation outcome

Rate limiting

RateLimitTracker 

RateLimitTracker()

Hybrid L1 (memory) + L2 (Redis) rate limit tracker.

record_rate_limit 

record_rate_limit(service_name: str) -> None

Record a 429 rate limit response.

record_request 

record_request(service_name: str) -> None

Record a request attempt.

get_rate_limit_count 

get_rate_limit_count(
    service_name: str, window_seconds: int
) -> int

Get the number of rate limits in the time window.

get_request_count 

get_request_count(
    service_name: str, window_seconds: int
) -> int

Get the number of requests in the time window.

get_backoff_level 

get_backoff_level(service_name: str) -> int

Get current backoff level for a service.

increment_backoff 

increment_backoff(service_name: str) -> int

Increment and return the new backoff level.

reset_backoff 

reset_backoff(service_name: str) -> None

Reset backoff level to zero.

clear_service 

clear_service(service_name: str) -> None

Clear all tracking data for a service.

get_rate_limit_tracker 

get_rate_limit_tracker() -> RateLimitTracker

Get the singleton rate limit tracker instance.