swarmauri_middleware_bulkhead 0.11.0.dev1

Bulkhead middleware implementation for Swarmauri

Swarmauri Middleware Bulkhead

Concurrency isolation middleware for FastAPI applications. Limit the number of simultaneous requests to protect resources from overload and ensure reliable service operation.

Features

• Concurrency control restricts the number of in-flight requests using an asyncio.Semaphore.
• Configurable limits let you tune max_concurrency (default: 10) to match service capacity.
• Fail-fast validation guards against non-positive concurrency limits at initialization.
• Structured logging surfaces request flow and failure details for easier diagnostics.
• FastAPI compatibility enables seamless integration into ASGI middleware stacks.

Installation

Choose the tool that matches your workflow:

# pip
pip install swarmauri_middleware_bulkhead

# Poetry
poetry add swarmauri_middleware_bulkhead

# uv
uv add swarmauri_middleware_bulkhead

Quickstart

The middleware wraps a call_next handler and ensures that no more than max_concurrency requests execute at once. Run the example below with python quickstart.py to see the concurrency ceiling in action.

import asyncio

from fastapi import Request

from swarmauri_middleware_bulkhead import BulkheadMiddleware


async def main() -> None:
    bulkhead = BulkheadMiddleware(max_concurrency=2)
    active_requests = 0
    peak_active_requests = 0
    lock = asyncio.Lock()

    async def call_next(request: Request):
        nonlocal active_requests, peak_active_requests
        async with lock:
            active_requests += 1
            peak_active_requests = max(peak_active_requests, active_requests)

        try:
            await asyncio.sleep(0.05)
            return {"path": request.scope.get("path"), "handled": True}
        finally:
            async with lock:
                active_requests -= 1

    async def simulate_request(idx: int):
        request = Request(scope={"type": "http", "path": f"/task/{idx}"})
        return await bulkhead.dispatch(request, call_next)

    responses = await asyncio.gather(*(simulate_request(i) for i in range(5)))

    assert peak_active_requests <= bulkhead.max_concurrency
    print("Peak concurrent requests:", peak_active_requests)
    print("Responses:", responses)


if __name__ == "__main__":
    asyncio.run(main())

Want to help?

If you want to contribute to swarmauri-sdk, read up on our guidelines for contributing that will help you get started.