Dependencies¶
A velociraptor doesn't rebuild its senses for every hunt. It uses the same keen nose, the same watchful eyes, the same read of the wind - over and over - without cluttering the strike itself. VelociPy's dependency injection works the same way: declare a reusable piece of logic once, then inject it wherever the pack needs it.
Depends lets you pull values - a database connection, a user session, a set of
pagination rules - into route handlers without tangling construction details
into your endpoint code. Sub-dependencies compose automatically. Generators run
teardown after the response leaves. And during tests you can swap a real
dependency for a mock with a single line.
See examples/dependencies.py for a runnable
demo of the patterns below.
A simple dependency¶
The simplest dependency is just a callable that returns a value. VelociPy calls it for you and hands the result to the handler.
from typing import Annotated
from velocipy import Depends, Query, VelociPy
app = VelociPy()
async def pagination_params(
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(le=100)] = 20,
):
return {"page": page, "size": size}
@app.get("/items")
async def list_items(params: Annotated[dict, Depends(pagination_params)]):
return params
Annotated[dict, Depends(...)] keeps the dependency declaration in the type
annotation, where VelociPy can read it without consuming the parameter
default.
You can also assign Depends(...) as the parameter default. This works,
but the type checker will need a type: ignore comment because the default
type does not match the parameter annotation.
Note
Dependency-injected parameters are hidden from OpenAPI automatically - only the inputs the dependency declares (query, header, path, body) show up in the generated docs.
Dependencies that read the request¶
Many dependencies need to inspect the current request: a header, a cookie, the
body, or a path parameter. VelociPy passes the live Request and any declared
parameters straight through, so the dependency stays focused on its job.
from velocipy import Depends, Header, HTTPException, Request, VelociPy
from velocipy.status import HTTP_401_UNAUTHORIZED
app = VelociPy()
def require_token(request: Request) -> str:
token = request.header("x-token")
if not token:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Missing token",
)
return token
@app.get("/protected")
async def protected(token: Annotated[str, Depends(require_token)]):
return {"token": token}
Dependencies can also declare Query, Header, Cookie, path, or body
parameters directly, letting VelociPy validate and parse the inputs for them:
from velocipy import Cookie, Depends, Header
def current_user(
token: Annotated[str, Header(alias="x-token")],
session: Annotated[str, Cookie()],
) -> dict[str, str]:
return {"token": token, "session": session}
@app.get("/whoami")
async def whoami(user: Annotated[dict[str, str], Depends(current_user)]):
return user
Sub-dependencies¶
Dependencies can depend on other dependencies. VelociPy resolves the whole tree in order, caching each result by default so a single dependency is never called twice during the same request.
from typing import Annotated
from velocipy import Depends, VelociPy
app = VelociPy()
def require_token() -> str:
return "raptor-token"
def require_admin(token: Annotated[str, Depends(require_token)]) -> str:
if token != "raptor-token":
raise HTTPException(status_code=401, detail="Not admin")
return token
@app.get("/admin-only")
async def admin_only(admin: Annotated[str, Depends(require_admin)]):
return {"admin": admin}
In this hunt, require_token runs first, then require_admin receives its
value. The tree is built from the leaves inward.
Generator teardown¶
Real resources usually need cleanup: close a database connection, release a
lock, flush telemetry. VelociPy supports generator dependencies with yield.
Everything after the yield runs after the response is sent, even if the
handler raises an exception.
from collections.abc import AsyncGenerator
from typing import Annotated, Any
from velocipy import Depends, VelociPy
app = VelociPy()
async def get_db() -> AsyncGenerator[dict[str, Any], None]:
db = {"connected": True}
try:
yield db
finally:
db["connected"] = False
@app.get("/items")
async def list_items(db: Annotated[dict[str, Any], Depends(get_db)]):
return {"items": [], "connected": db["connected"]}
Async context managers work too
Any contextlib.asynccontextmanager decorated function can be used as a
dependency. VelociPy enters the context before the handler and exits it
afterward.
import contextlib
from collections.abc import AsyncGenerator
from typing import Annotated
from velocipy import Depends
@contextlib.asynccontextmanager
async def get_resource() -> AsyncGenerator[str, None]:
yield "resource"
@app.get("/resource")
async def resource_page(r: Annotated[str, Depends(get_resource)]):
return {"resource": r}
Controlling caching with use_cache¶
By default, the result of a dependency is cached for the lifetime of the request. If you need a fresh value for every injection site, disable caching:
from typing import Annotated
from velocipy import Depends
def counter() -> int:
# This will be called once per injection site when use_cache=False.
return 1
@app.get("/nocache")
async def no_cache(
a: Annotated[int, Depends(counter, use_cache=False)],
b: Annotated[int, Depends(counter, use_cache=False)],
):
return {"a": a, "b": b}
Warning
Use use_cache=False sparingly. Most of the time a single database
connection, settings object, or user identity per request is exactly what
you want.
Overrides for testing¶
The easiest way to test a handler in isolation is to replace its dependencies.
app.dependency_overrides maps the real dependency callable to a substitute.
The override applies transitively: if a sub-dependency is swapped, every
dependency that uses it sees the replacement.
from typing import Annotated, Any
from velocipy import Depends, VelociPy
from velocipy.testing import TestClient
app = VelociPy()
def real_db() -> dict[str, Any]:
return {"source": "production"}
@app.get("/items")
async def list_items(db: Annotated[dict[str, Any], Depends(real_db)]):
return db
# Swap the dependency only for this test run.
app.dependency_overrides[real_db] = lambda: {"source": "test"}
client = TestClient(app)
response = client.get("/items")
assert response.json() == {"source": "test"}
Tip
Always clear overrides between tests to keep hunts from bleeding into each other:
At a glance¶
| Pattern | How to declare | Teardown supported? |
|---|---|---|
| Plain callable | Annotated[T, Depends(func)] |
No |
| Sub-dependency | Declare dependencies as parameters of another dependency | No |
| Generator | yield value, cleanup after |
Yes |
| Async generator | async def ... yield |
Yes |
| Async context manager | @contextlib.asynccontextmanager |
Yes |
| Per-injection value | Depends(func, use_cache=False) |
Depends on func |
| Test override | app.dependency_overrides[func] = replacement |
No |
Next steps¶
- See
examples/dependencies.pyfor sub-dependencies, generator teardown, and overrides in one file. - Read Security to learn how
APIKeyHeader,APIKeyQuery, andOAuth2PasswordBearerplug intoDepends. - Read Rate limiting to enforce token-bucket or windowed limits as dependencies.