baldur.services — Bulkhead

Per-domain concurrency isolation: the bulkhead base class and its semaphore implementations, the per-domain registry, the @bulkhead decorator, policy adapters for composition, and the rejection/timeout exceptions.

Core & implementations

Bulkhead

Bases: ABC

Bulkhead abstract interface.

Prevents a failure in one component from propagating to other components through resource isolation.

Usage (context manager): bulkhead = SemaphoreBulkhead("database", max_concurrent=10)

with bulkhead.acquire(timeout=5.0):
    do_database_work()

Usage (decorator): @bulkhead.wrap def do_work(): pass

name abstractmethod property

name: str

Bulkhead name (domain identifier).

acquire abstractmethod

acquire(
    timeout: float | None = None,
) -> Generator[None, None, None]

Acquire the resource (context manager).

Parameters:

Name	Type	Description	Default
timeout	float | None	Wait timeout (seconds). If None, fail immediately (non-blocking).	None

Yields:

Type	Description
None	None

Raises:

Type	Description
BulkheadFullError	When resource acquisition fails

try_acquire abstractmethod

try_acquire(timeout: float | None = None) -> bool

Attempt to acquire the resource.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum time (seconds) the implementation may wait for capacity — an upper bound on waiting, not a guarantee of it. None means no waiting (immediate verdict). Implementations whose capacity model already buffers bursts (bounded-queue thread pools) may return an immediate verdict for any timeout value.	None

Returns:

Type	Description
bool	True if acquisition succeeded, False if it failed

release abstractmethod

release() -> None

Release the resource.

get_state abstractmethod

get_state() -> BulkheadState

Return the current state.

execute

execute(
    fn: Callable[..., T],
    *args: Any,
    timeout: float | None = None,
    **kwargs: Any
) -> T

Execute a function inside the bulkhead compartment.

Default implementation: acquire a permit (bounded by timeout), run fn in the calling thread, release on return. timeout bounds the admission wait only — once admitted, fn occupies the caller until it returns. Implementations that offload execution (dedicated worker pools) override this to also bound execution time.

Non-abstract so existing third-party subclasses stay instantiable.

Parameters:

Name	Type	Description	Default
fn	Callable[..., T]	Function to execute	required
*args	Any	Positional arguments	()
timeout	float | None	Admission-wait timeout (seconds). If None, fail immediately when no permit is available (non-blocking).	None
**kwargs	Any	Keyword arguments	{}

Returns:

Type	Description
T	Function execution result

Raises:

Type	Description
BulkheadFullError	When no permit is available within timeout

wrap

wrap(fn: Callable[..., T]) -> Callable[..., T]

Decorator that wraps a function with the bulkhead.

Parameters:

Name	Type	Description	Default
fn	Callable[..., T]	Function to wrap	required

Returns:

Type	Description
Callable[..., T]	Function with the bulkhead applied

SemaphoreBulkhead

SemaphoreBulkhead(
    name: str, max_concurrent: int = 10, fair: bool = True
)

Bases: Bulkhead

Semaphore-based bulkhead.

Limits the concurrent execution count to prevent resource exhaustion. Suitable for I/O-bound work (DB queries, cache lookups, etc.).

Features: - Maximum concurrent execution count limit - Timeout-based wait support - Rejection statistics tracking

Parameters:

Name	Type	Description	Default
name	str	Bulkhead name (domain identifier)	required
max_concurrent	int	Maximum concurrent execution count	10
fair	bool	If True, guarantees FIFO ordering (future implementation)	True

name property

name: str

Bulkhead name.

acquire

acquire(
    timeout: float | None = None,
) -> Generator[None, None, None]

Acquire the resource.

Parameters:

Name	Type	Description	Default
timeout	float | None	Wait timeout (seconds). If None, fail immediately (non-blocking).	None

Yields:

Type	Description
None	None

Raises:

Type	Description
BulkheadFullError	When resource acquisition fails

try_acquire

try_acquire(timeout: float | None = None) -> bool

Attempt to acquire. Non-blocking if timeout is None, otherwise waits up to the given time.

Parameters:

Name	Type	Description	Default
timeout	float | None	Wait timeout (seconds). If None, succeed/fail immediately.	None

Returns:

Type	Description
bool	True if acquisition succeeded, False if it failed

release

release() -> None

Release the resource.

get_state

get_state() -> BulkheadState

Return the current state.

AsyncSemaphoreBulkhead

AsyncSemaphoreBulkhead(name: str, max_concurrent: int = 10)

Async semaphore-based bulkhead.

Limits concurrency in an asyncio environment to prevent resource exhaustion. Does not block the event loop; suited to async I/O work.

Features: - asyncio.Semaphore-based non-blocking wait - asyncio.wait_for-based timeout - Rejection statistics tracking

Parameters:

Name	Type	Description	Default
name	str	Bulkhead name (domain identifier)	required
max_concurrent	int	Maximum concurrent executions	10

name property

name: str

Bulkhead name.

acquire async

acquire(
    timeout: float | None = None,
) -> AsyncGenerator[None, None]

Acquire a resource asynchronously.

Parameters:

Name	Type	Description	Default
timeout	float | None	Wait timeout (seconds). None fails immediately (non-blocking).	None

Yields:

Type	Description
AsyncGenerator[None, None]	None

Raises:

Type	Description
BulkheadFullError	When resource acquisition fails

try_acquire async

try_acquire(timeout: float | None = None) -> bool

Acquisition attempt, mirroring this class's acquire timeout contract.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum time (seconds) to wait for capacity — an upper bound on waiting, not a guarantee of it. None means no waiting (immediate verdict).	None

Returns:

Type	Description
bool	True on success, False on failure

release async

release() -> None

Release the resource.

get_state

get_state() -> BulkheadState

Return the current state (synchronous method).

BulkheadState dataclass

BulkheadState(
    name: str,
    bulkhead_type: BulkheadType,
    max_concurrent: int,
    active_count: int,
    waiting_count: int,
    rejected_count: int,
    last_rejection_time: datetime | None = None,
)

Bulkhead current state data.

name instance-attribute

name: str

Bulkhead name (domain identifier)

bulkhead_type instance-attribute

bulkhead_type: BulkheadType

Bulkhead type

max_concurrent instance-attribute

max_concurrent: int

Maximum allowed concurrent execution count

active_count instance-attribute

active_count: int

Number of currently running tasks

waiting_count instance-attribute

waiting_count: int

Number of waiting tasks

rejected_count instance-attribute

rejected_count: int

Total number of rejected requests

last_rejection_time class-attribute instance-attribute

last_rejection_time: datetime | None = None

Last rejection time

available_permits property

available_permits: int

Number of available permits.

utilization_percent property

utilization_percent: float

Utilization (0-100%).

BulkheadType

Bases: str, Enum

Bulkhead type.

SEMAPHORE class-attribute instance-attribute

SEMAPHORE = 'semaphore'

Semaphore-based - limits only the concurrent execution count

THREAD_POOL class-attribute instance-attribute

THREAD_POOL = 'thread_pool'

Thread-pool-based - isolated execution in a dedicated thread pool

Registry

BulkheadRegistry

BulkheadRegistry(settings: BulkheadSettings | None = None)

Per-domain bulkhead registry.

Integrates with the existing ConnectionType and also supports custom domains.

Features: - Automatic registration of default bulkheads per ConnectionType - Custom domain support - Automatic creation of asynchronous bulkheads - Fine-grained bulkheads per DB alias / cache instance

Parameters:

Name	Type	Description	Default
settings	BulkheadSettings | None	Bulkhead settings. Uses defaults if None.	None

get

get(name: str | ConnectionType) -> Bulkhead

Look up a bulkhead.

Parameters:

Name	Type	Description	Default
name	str | ConnectionType	Domain name or ConnectionType	required

Returns:

Type	Description
Bulkhead	Bulkhead instance

Raises:

Type	Description
BulkheadNotFoundError	Unregistered domain. Subclasses KeyError, so existing except KeyError consumers keep working; the message names the missing domain and lists the registered compartments.

get_or_create

get_or_create(
    name: str,
    max_concurrent: int | None = None,
    bulkhead_type: str = "semaphore",
) -> Bulkhead

Look up or create a bulkhead.

Parameters:

Name	Type	Description	Default
name	str	Domain name	required
max_concurrent	int | None	Maximum concurrent execution count (default if None)	None
bulkhead_type	str	"semaphore" or "thread_pool". A thread-pool request routes through the overridable builder seam — on the base registry it falls back to semaphore isolation (see _build_thread_pool_bulkhead).	'semaphore'

Returns:

Type	Description
Bulkhead	Bulkhead instance

get_async

get_async(
    name: str | ConnectionType,
) -> AsyncSemaphoreBulkhead

Look up an asynchronous bulkhead.

Creates/returns the asynchronous version deriving its capacity from the synchronous twin. Strict: a domain with no synchronous twin is treated identically to get() — provisioning must precede the async lookup, so the async twin can never be a registry-invisible, default-capacity mint.

Parameters:

Name	Type	Description	Default
name	str | ConnectionType	Domain name or ConnectionType	required

Returns:

Type	Description
AsyncSemaphoreBulkhead	AsyncSemaphoreBulkhead instance

Raises:

Type	Description
BulkheadNotFoundError	No synchronous twin for name. Subclasses KeyError; message lists the registered compartments.

get_for_database

get_for_database(alias: str = 'default') -> Bulkhead

Return the bulkhead for a DB alias.

Parameters:

Name	Type	Description	Default
alias	str	Django DB alias (default, replica, analytics, etc.)	'default'

Returns:

Type	Description
Bulkhead	The bulkhead for the given alias

get_for_cache

get_for_cache(name: str = 'default') -> Bulkhead

Return the bulkhead for a cache instance.

Parameters:

Name	Type	Description	Default
name	str	Cache name (default, session, etc.)	'default'

Returns:

Type	Description
Bulkhead	The bulkhead for the given cache

register

register(bulkhead: Bulkhead) -> None

Register a custom bulkhead.

Overwriting a built-in name (one of the four ConnectionType values) is allowed — it is the only path for swapping a built-in's implementation type and is load-bearing for the re-register capacity flow — but it permanently replaces a settings-owned protection compartment, so a WARNING is emitted to flag the footgun.

The async twin for the same name is invalidated so async callees pick up the new capacity on next access.

Parameters:

Name	Type	Description	Default
bulkhead	Bulkhead	Bulkhead to register	required

unregister

unregister(name: str) -> bool

Unregister a bulkhead.

The async twin for the same name is invalidated alongside the sync entry.

Parameters:

Name	Type	Description	Default
name	str	Domain name	required

Returns:

Type	Description
bool	True if unregistration succeeded, False if the name was not registered

Raises:

Type	Description
ValueError	name is a built-in compartment (one of the four ConnectionType values). Built-ins are settings-owned protection compartments — removing one silently degrades isolation, so the mutation is blocked rather than allowed.

get_all_states

get_all_states() -> dict[str, BulkheadState]

Return all bulkhead states.

list_names

list_names() -> list[str]

Return all registered domain names.

get_bulkhead_registry

get_bulkhead_registry() -> BulkheadRegistry

Return the active BulkheadRegistry — the resolution chain.

Chain: the ProviderRegistry.bulkhead_registry slot when populated (a registry overlay registered by an extension package), else the lazily-built base singleton. The registry is the user-facing provisioning API (get_or_create before @bulkhead on custom domains), so a single getter structurally prevents provisioning into an inactive registry.

A provider registered into the slot MUST NOT itself consult the slot (its factory runs while the slot's non-reentrant lock is held — a re-entrant resolution fails loud with RuntimeError).

reset_bulkhead_registry

reset_bulkhead_registry() -> None

Reset the base singleton (for testing).

Clears only the chain's fallback leg; a populated provider slot is owned (and reset) by whoever registered it.

Decorator & policy

bulkhead

bulkhead(
    name: str | ConnectionType,
    timeout: float | None = None,
    fallback: Callable[..., T] | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Bulkhead decorator (sync/async auto-dispatching).

Detects the function type via asyncio.iscoroutinefunction() and automatically applies the appropriate bulkhead (sync/async).

Custom domains must be provisioned (via get_or_create(), register(), or a policy/settings helper) before the decorated function is first called. An unregistered domain raises BulkheadNotFoundError — naming the registered compartments — on both sync and async callees; the error is raised before any fallback is considered, so fallback does not mask it. The four built-in compartments (database, cache, external_api, message_queue) are always available.

Parameters:

Name	Type	Description	Default
name	str | ConnectionType	Domain name or ConnectionType	required
timeout	float | None	Resource acquisition wait timeout (seconds). If None, fail immediately.	None
fallback	Callable[..., T] | None	Alternative function to call when the bulkhead is full	None

Returns:

Type	Description
Callable[[Callable[..., T]], Callable[..., T]]	Decorator with the bulkhead applied

Examples:

Synchronous function (built-in compartment)

@bulkhead(ConnectionType.DATABASE) def db_query(): return execute_query()

Asynchronous function (built-in compartment)

@bulkhead(ConnectionType.DATABASE) async def async_db_query(): return await execute_async_query()

Timeout setting (built-in compartment)

@bulkhead("external_api", timeout=5.0) def call_external_api(): return requests.get(url)

Custom domain with fallback — provision it first, then decorate

from baldur.services.bulkhead import get_bulkhead_registry

get_bulkhead_registry().get_or_create("reports", max_concurrent=5)

@bulkhead("reports", fallback=lambda: {"status": "unavailable"}) def get_data(): return fetch_data()

bulkhead_for_database

bulkhead_for_database(
    alias: str = "default",
    timeout: float | None = None,
    fallback: Callable[..., T] | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Per-DB-alias bulkhead decorator.

Internally uses BulkheadPolicy/AsyncBulkheadPolicy and looks up the bulkhead for the given alias via the Registry's get_for_database().

Parameters:

Name	Type	Description	Default
alias	str	Django DB alias (default, replica, analytics, etc.)	'default'
timeout	float | None	Resource acquisition wait timeout (seconds)	None
fallback	Callable[..., T] | None	(deprecated) Alternative function to call when the bulkhead is full. Using the BulkheadPolicy + FallbackPolicy combination directly is recommended.	None

Examples:

@bulkhead_for_database("default") def write_to_db(): Model.objects.using("default").create(...)

@bulkhead_for_database("replica") def read_from_replica(): return Model.objects.using("replica").all()

bulkhead_for_cache

bulkhead_for_cache(
    name: str = "default",
    timeout: float | None = None,
    fallback: Callable[..., T] | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Per-cache-instance bulkhead decorator.

Internally uses BulkheadPolicy/AsyncBulkheadPolicy and looks up the bulkhead for the given cache via the Registry's get_for_cache().

Parameters:

Name	Type	Description	Default
name	str	Cache name (default, session, etc.)	'default'
timeout	float | None	Resource acquisition wait timeout (seconds)	None
fallback	Callable[..., T] | None	(deprecated) Alternative function to call when the bulkhead is full. Using the BulkheadPolicy + FallbackPolicy combination directly is recommended.	None

Examples:

@bulkhead_for_cache("default") def get_cached_value(key: str): return cache.get(key)

@bulkhead_for_cache("session") def get_session_data(session_id: str): return session_cache.get(session_id)

BulkheadPolicy

BulkheadPolicy(
    bulkhead: Bulkhead, timeout: float | None = None
)

Bases: ResiliencePolicy[T]

Synchronous Bulkhead Policy — resource isolation.

Delegates execution polymorphically to Bulkhead.execute() — each implementation supplies its own execution semantics (semaphore: run in the calling thread, timeout bounds the admission wait; worker-pool implementations may offload and also bound execution time).

Exception handling contract (same as CircuitBreakerPolicy): - BulkheadFullError → absorbed into PolicyResult(outcome=REJECTED) - BulkheadTimeoutError → absorbed into PolicyResult(outcome=TIMEOUT) (unreachable on the semaphore path — admission failure raises FullError) - Business exception during function execution → re-propagated via raise

Parameters:

Name	Type	Description	Default
bulkhead	Bulkhead	Bulkhead ABC implementation. Injected via DI. Registry lookup uses the bulkhead_policy() factory.	required
timeout	float | None	Timeout (seconds) forwarded to Bulkhead.execute(). Semaphore implementations bound the admission wait with it (None = fail immediately, Fast Fail); worker-pool implementations bound execution time with it.	None

name property

name: str

Policy name.

bulkhead_name property

bulkhead_name: str

Domain identifier of the internal Bulkhead instance.

execute

execute(
    func: Callable[..., T],
    *args: Any,
    context: PolicyContext | None = None,
    **kwargs: Any
) -> PolicyResult[T]

Execute the function within the bulkhead resource.

A single polymorphic call — self._bulkhead.execute() — carries the implementation-specific semantics; this policy only maps exceptions to outcomes:

• BulkheadFullError → PolicyOutcome.REJECTED
• BulkheadTimeoutError → PolicyOutcome.TIMEOUT
• Business exceptions during function execution re-propagate via raise.

AsyncBulkheadPolicy

AsyncBulkheadPolicy(
    async_bulkhead: AsyncSemaphoreBulkhead,
    timeout: float | None = None,
)

Asynchronous Bulkhead Policy — AsyncResiliencePolicy Protocol implementation.

Wraps AsyncSemaphoreBulkhead and returns results in PolicyResult form. Follows the same exception handling contract as the synchronous BulkheadPolicy.

Since AsyncSemaphoreBulkhead is a separate class that does not inherit from Bulkhead(ABC), it is implemented as a class fully separated from the synchronous BulkheadPolicy.

Parameters:

Name	Type	Description	Default
async_bulkhead	AsyncSemaphoreBulkhead	AsyncSemaphoreBulkhead instance (DI).	required
timeout	float | None	Resource acquisition timeout (seconds). If None, fail immediately.	None

name property

name: str

Policy name.

bulkhead_name property

bulkhead_name: str

Domain identifier of the internal AsyncSemaphoreBulkhead.

execute async

execute(
    func: Callable[..., T],
    *args: Any,
    context: PolicyContext | None = None,
    **kwargs: Any
) -> PolicyResult[T]

Execute the function within the asynchronous bulkhead resource.

BulkheadFullError → absorbed into PolicyResult(outcome=REJECTED). Business exceptions during function execution are re-propagated via raise.

bulkhead_policy

bulkhead_policy(
    name: str,
    max_concurrent: int | None = None,
    timeout: float | None = None,
    bulkhead_type: str = "semaphore",
) -> BulkheadPolicy

BulkheadPolicy factory — BulkheadRegistry singleton integration.

Calls the Registry's get_or_create() to guarantee a single global Bulkhead instance for the same name.

The Registry dependency occurs only in this factory function. The BulkheadPolicy class itself does not know the Registry (test-friendly).

Parameters:

Name	Type	Description	Default
name	str	Domain name (Registry key)	required
max_concurrent	int | None	Maximum concurrent execution count (Registry default if None)	None
timeout	float | None	Resource acquisition timeout (fail immediately if None)	None
bulkhead_type	str	"semaphore" or "thread_pool"	'semaphore'

Returns:

Type	Description
BulkheadPolicy	BulkheadPolicy instance (uses the Registry singleton Bulkhead)

async_bulkhead_policy

async_bulkhead_policy(
    name: str,
    max_concurrent: int | None = None,
    timeout: float | None = None,
) -> AsyncBulkheadPolicy

AsyncBulkheadPolicy factory — BulkheadRegistry singleton integration.

Calls the Registry's get_async() to guarantee a single global AsyncSemaphoreBulkhead instance for the same name.

Parameters:

Name	Type	Description	Default
name	str	Domain name (Registry key)	required
max_concurrent	int | None	Maximum concurrent execution count (Registry default if None). The synchronous Bulkhead is provisioned first (so the domain is registry-visible to the admin API, metrics, and shutdown iteration) and used as the configuration basis for the asynchronous instance.	None
timeout	float | None	Resource acquisition timeout (fail immediately if None)	None

Returns:

Type	Description
AsyncBulkheadPolicy	AsyncBulkheadPolicy instance (uses the Registry singleton AsyncSemaphoreBulkhead)

Metrics

BulkheadMetricsUpdater

BulkheadMetricsUpdater(interval: float = 10.0)

Background thread that periodically refreshes every bulkhead's metrics.

Pulls the current state of every registered bulkhead and writes it into the Prometheus metrics on a schedule.

Usage

updater = BulkheadMetricsUpdater(interval=10.0) updater.start()

... application runs ...

updater.stop()

Parameters:

Name	Type	Description	Default
interval	float	metric update interval (seconds)	10.0

start

start() -> None

Start the metrics updater.

stop

stop() -> None

Stop the metrics updater.

get_metrics_updater

get_metrics_updater(
    interval: float = 10.0,
) -> BulkheadMetricsUpdater

Return the BulkheadMetricsUpdater singleton.

start_metrics_updater

start_metrics_updater(
    interval: float = 10.0,
) -> BulkheadMetricsUpdater

Start the metrics updater (convenience function).

stop_metrics_updater

stop_metrics_updater() -> None

Stop the metrics updater (convenience function).

update_bulkhead_metrics

update_bulkhead_metrics(
    bulkhead_name: str,
    bulkhead_type: str,
    active_count: int,
    max_concurrent: int,
    waiting_count: int,
) -> None

Update bulkhead state gauges (poll path).

Parameters:

Name	Type	Description	Default
bulkhead_name	str	bulkhead name	required
bulkhead_type	str	bulkhead type (semaphore, thread_pool)	required
active_count	int	current number of active requests	required
max_concurrent	int	maximum concurrent capacity	required
waiting_count	int	number of waiting requests	required

increment_rejected_count

increment_rejected_count(bulkhead_name: str) -> None

Increment the rejected counter (event path).

Parameters:

Name	Type	Description	Default
bulkhead_name	str	bulkhead name	required

Exceptions

BulkheadError

BulkheadError(message: str = '', *, code: str = '')

Bases: ResilienceError

Base exception for bulkhead-related failures.

extra_context

extra_context() -> dict

Return structlog-bindable context for bulkhead errors.

BulkheadFullError

BulkheadFullError(
    bulkhead_name: str,
    max_concurrent: int,
    active_count: int,
)

Bases: PolicyRejectedException, BulkheadError

Raised when the bulkhead has no permits available and a new call is rejected.

Multi-inherits PolicyRejectedException so the outer PolicyComposer catch hierarchy classifies bulkhead rejection as PolicyOutcome.REJECTED rather than the generic except Exception branch (which would mislabel them as FAILURE).

Parameters:

Name	Type	Description	Default
bulkhead_name	str	bulkhead identifier	required
max_concurrent	int	maximum allowed concurrent executions	required
active_count	int	number of executions currently active	required

extra_context

extra_context() -> dict

Return structlog-bindable context for bulkhead full errors.

BulkheadTimeoutError

BulkheadTimeoutError(bulkhead_name: str, timeout: float)

Bases: TimeoutPolicyError, BulkheadError

Raised when a bulkhead-managed call exceeds its timeout budget.

Emitted by ThreadPoolBulkhead when the wrapped task does not complete within the configured timeout. Multi-inherits TimeoutPolicyError so the outer PolicyComposer catch hierarchy classifies the failure as PolicyOutcome.TIMEOUT (instead of funneling into the generic except Exception branch as FAILURE).

Parameters:

Name	Type	Description	Default
bulkhead_name	str	bulkhead identifier	required
timeout	float	configured timeout in seconds	required

extra_context

extra_context() -> dict

Return structlog-bindable context for bulkhead timeout errors.

BulkheadNotFoundError

BulkheadNotFoundError(
    bulkhead_name: str, registered_names: list[str]
)

Bases: BulkheadError, KeyError

Raised when a bulkhead is requested for a domain that has not been provisioned.

Custom domains must be provisioned (via register(), get_or_create(), or a policy/settings helper) before a @bulkhead-decorated function is first called, on both sync and async callees.

Multi-inherits KeyError so the registry's long-standing not-found contract stays intact for existing except KeyError consumers (the traffic-gate skip-degradation path, the admin API 404 mapping), while except BulkheadError / except BaldurError also classify it. Multi-inheritance precedent: BulkheadFullError(PolicyRejectedException, BulkheadError).

__str__ is overridden because KeyError.__str__ would otherwise repr-quote the message (the base hierarchy defines no __str__), mangling the actionable text into "Bulkhead not found: ..." with surrounding quotes.

Parameters:

Name	Type	Description	Default
bulkhead_name	str	the requested, unregistered domain	required
registered_names	list[str]	currently registered compartment names	required

extra_context

extra_context() -> dict

Return structlog-bindable context for bulkhead not-found errors.