swarmauri_middleware_bulkhead 0.8.0.dev31

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.