Request-Time Injection¶
While Wireup primarily handles dependency injection in FastAPI routes, you can also use dependencies in other functions during the request lifecycle to build reusable decorators:
- Reusable Route Decorators:
@require_admin,@require_permission,@rate_limit - Middleware Integration: request context setup and cross-cutting request logic
- Migration Support: composing Wireup services with external FastAPI dependencies
Advanced Feature - Requires Middleware Mode
The examples on this page require middleware_mode=True during Wireup setup.
wireup.integration.fastapi.setup(container, app, middleware_mode=True)
Normally, the request-scoped container is created just before the route handler is called. With middleware mode enabled, it's created at the start of the HTTP request lifecycle, making it available in middleware and other request handlers. These examples apply to HTTP requests only, not WebSocket handlers.
Composable Route Decorators¶
Route decorators let you extract cross-cutting concerns into reusable components that can be applied to multiple endpoints. Examples below:
Authentication & Authorization¶
import contextlib
from collections.abc import AsyncIterator
from wireup import Injected
from wireup.integration.fastapi import inject
@contextlib.asynccontextmanager
@inject
async def require_auth(auth: Injected[AuthService]) -> AsyncIterator[None]:
if not await auth.is_authenticated():
raise HTTPException(status_code=401, detail="Authentication required")
yield
@router.get("/users")
@require_auth()
async def get_users(user_service: Injected[UserService]): ...
Rate Limiting¶
import contextlib
from collections.abc import AsyncIterator
from wireup import Injected
from wireup.integration.fastapi import inject
def rate_limit(*, max_requests: int = 100):
@contextlib.asynccontextmanager
@inject
async def guard(
rate_limiter: Injected[RateLimiterService],
) -> AsyncIterator[None]:
if not await rate_limiter.check(max_requests):
raise HTTPException(status_code=429, detail="Too many requests")
yield
return guard()
@router.get("/api/data")
@rate_limit(max_requests=50)
async def get_data(data_service: Injected[DataService]): ...
Middleware Integration¶
Access Wireup services in FastAPI middleware for request setup or other cross-cutting concerns.
Middleware Injection¶
FastAPI's dependency system (Depends) does not apply to middleware. Middleware runs outside route handlers, so this is
one of the places where FastAPI cannot inject your services for you.
Use @inject here when you want request-scoped services in middleware, and run your service layer directly before
passing control to the route handler.
from wireup import Injected
from wireup.integration.fastapi import inject
@app.middleware("http")
@inject
async def request_middleware(
request: Request,
call_next,
request_context: Injected[RequestContextService],
) -> Response:
await request_context.initialize(request)
return await call_next(request)
Integrating with FastAPI Dependencies¶
When migrating to Wireup or composing with external libraries that provide FastAPI dependencies, you can access Wireup
services within Depends() functions.
from wireup import Injected
from wireup.integration.fastapi import inject
@inject
async def get_user_dependency(
request: Request,
auth_header: Annotated[str, Depends(get_auth_header)],
auth_service: Injected[AuthService],
user_service: Injected[UserService],
):
user = await auth_service.authenticate(auth_header)
return await user_service.enrich(user)
@router.get("/users")
async def get_users(
user: Annotated[User, Depends(get_user_dependency)],
user_service: Injected[UserService],
): ...
Avoid When Possible
This should be rare. Prefer pure Wireup injection. Use this only when:
- Migrating gradually to Wireup from pure FastAPI
Depends() - External libraries require
Depends()integration
Remember: Wireup services cannot depend on Depends() providers.
Testing¶
When testing code that uses request-time injection (@inject or get_request_container()), make sure the request-scoped container is available.
Testing Route Decorators¶
from fastapi.testclient import TestClient
import wireup
import wireup.integration.fastapi
# Set up app with middleware_mode enabled
app = FastAPI()
container = wireup.create_async_container(
injectables=[AuthService, UserService]
)
wireup.integration.fastapi.setup(container, app, middleware_mode=True)
@router.get("/users")
@require_auth()
async def get_users(user_service: Injected[UserService]): ...
def test_require_auth_denied():
with TestClient(app) as client:
# AuthService will be mocked via container overrides
response = client.get("/users")
assert response.status_code == 401
def test_require_auth_allowed():
with TestClient(app) as client:
# Override AuthService to return True
with get_app_container(app).override.injectable(
AuthService, new=MockAuthService(allow=True)
):
response = client.get("/users")
assert response.status_code == 200
Testing Middleware¶
def test_request_middleware_runs():
with TestClient(app) as client:
response = client.get("/api/users")
assert response.status_code == 200
Tip
Use get_app_container(app).override.injectable() to inject mocks and fakes during tests. This works for both route
decorators and middleware.
Direct Container Access¶
For utilities or other cases where you need direct access to the container APIs:
from wireup.integration.fastapi import get_app_container, get_request_container
# Access the request-scoped container (used for the current request).
request_container = get_request_container()
# Access the application-wide container (created via `wireup.create_async_container`).
# Use this when you need the container outside of the request context lifecycle.
app_container = get_app_container(app)
The app container is always retrievable given an instance of the application. The request-scoped container is only
available when middleware_mode=True is enabled.
Tip
@inject keeps dependency wiring in the function signature and avoids manual container lookups. If you prefer explicit container access,
get_request_container() works anywhere during a request:
from wireup.integration.fastapi import get_request_container
async def write_audit_log(message: str) -> None:
audit = await get_request_container().get(AuditService)
await audit.write(message)