Metadata-Version: 2.5
Name: fastapi-injected
Version: 0.3.6
Summary: Yet another library to reuse fastapi dependency injection
Project-URL: Repository, https://github.com/uriyyo/fastapi-injected
Author-email: Yurii Karabas <1998uriyyo@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.12
Requires-Dist: fastapi>=0.141.0
Requires-Dist: typing-extensions>=4.16.0
Requires-Dist: typing-inspection>=0.4.4
Description-Content-Type: text/markdown

# fastapi-injected

Yet another attempt to reuse FastAPI's dependency injection outside of request handlers.

This is an opinionated library: it takes the DI machinery you already know from FastAPI (`Depends`, generator dependencies with teardown, dependency caching) and makes it usable in plain async functions — background jobs, CLI commands, workers, scripts — without a `Request` in sight.

## Installation

```sh
pip install fastapi-injected
```

Requires Python 3.12+.

## Usage

Declare dependencies as regular classes and annotate fields with `Dep[...]`:

```python
from dataclasses import dataclass
from typing import AsyncIterator

from fastapi_injected import Dep, DepFactory, Injected, inject


@dataclass
class Session:
    closed: bool = False


async def session_dep() -> AsyncIterator[Session]:
    session = Session()
    try:
        yield session
    finally:
        session.closed = True


@dataclass
class Repository:
    session: DepFactory[Session, session_dep]


@dataclass
class Service:
    repo: Dep[Repository]


@inject
async def handler(*, service: Dep[Service] = Injected) -> None:
    ...  # service is built and injected, session is closed on exit


await handler()
```

- `Dep[T]` — resolve `T` by calling it, same as FastAPI's `Annotated[T, Depends()]`.
- `DepFactory[T, factory]` — resolve `T` via a factory, same as `Annotated[T, Depends(factory)]`. Generator factories get proper teardown.
- `Arg[T]` — keep a parameter out of the dependency graph even though its annotation is a dependency, so the caller supplies it. See [Caller-supplied arguments](#caller-supplied-arguments).
- `Injected` — a sentinel default that exists purely to make type checkers happy: without it they would complain about a missing argument at call sites. At runtime the parameter is always filled in by `@inject`.

Injected parameters mix freely with regular ones — pass your own arguments as usual and the rest is injected:

```python
@inject
async def add(a: int, b: int, *, service: Dep[Service] = Injected) -> int:
    ...


result = await add(1, 2)
```

### Caller-supplied arguments

A parameter is injected only if it is written as a dependency — `Dep[...]`, `DepFactory[...]`, an `Annotated[..., Depends(...)]` of your own, or a `Depends(...)` default. Everything else is left for the caller, annotation untouched: it is never handed to pydantic, so any type works with no marker at all.

```python
class Connection:  # not a pydantic-friendly type, and it does not have to be
    ...


@inject
async def handler(conn: Connection, *, service: Dep[Service] = Injected) -> None:
    ...


await handler(connection)
```

`Arg[...]` is for the opposite case — a parameter that *is* spelled as a dependency, in a function that wants it passed in instead:

```python
from fastapi_injected import Arg

type ServiceDep = Annotated[Service, Depends(get_service)]


@inject
async def handler(service: Arg[ServiceDep], *, session: Dep[Session] = Injected) -> None:
    ...


await handler(service)
```

`Arg[T]` is a no-op for type checkers — the parameter stays typed as `T` — and at runtime it tells `@inject` to keep this parameter out of the dependency graph.

Writing `Injected` as the default of a parameter that is not a dependency raises `NotADependencyError` at decoration time, since nothing would ever fill it in.

### Resolving a type directly

No decorator needed — resolve a dependency graph on demand:

```python
from fastapi_injected import resolve

service = await resolve(Service)
```

Like `@inject`, `resolve` accepts `new_scope=True` to force a fresh scope instead of reusing the surrounding one.

### Scopes and caching

By default every call to an injected function gets its own scope: dependencies are built, cached within the call, and torn down when it returns. Wrap several calls in `push_inject_scope()` to share one cache (and defer teardown to the end of the scope):

```python
from fastapi_injected import push_inject_scope

async with push_inject_scope():
    a = await handler()  # dependencies built here
    b = await handler()  # same instances reused
# generator dependencies are torn down here
```

Use `@inject(new_scope=True)` to opt a function out of the surrounding scope and always get fresh dependencies. It builds them again, but stays in the request they are being built for, so request-bound dependencies keep resolving.

Outside of a request there is no request to resolve against, so the scope makes one up. It answers what a dependency usually reads — `method`, `url`, `headers`, `path_params`, `client`, `state` — with the values of a request nobody sent. The application is the one thing it cannot invent: pass it when a dependency reaches for `request.app`:

```python
async with push_inject_scope(app=app):
    await handler()  # `request.app.state` resolves as it does in a route
```

Without it, reading `request.app` raises a `KeyError` naming what is missing rather than a bare `'app'`.

When a dependency reads more of the request than that — a header, a cookie, a query or path parameter, the body — describe the request with `synthetic_request` and hand it to the scope:

```python
from fastapi_injected import push_inject_scope, synthetic_request

request = synthetic_request(
    headers={"x-trace": "job-42"},
    cookies={"sid": "abc"},
    query={"page": "2"},
    path_params={"item_id": 7},
    body={"name": "readme"},  # sent as JSON; bytes and str are sent as they are
    app=app,
)

async with push_inject_scope(request=request):
    await handler()  # `Header()`, `Cookie()`, `Query()`, `Path()` and `Body()` resolve from it
```

Analysing a dependency is the expensive part of resolving one, so the result is cached — keyed by the dependency itself, not by whatever object carried it, so nothing that only passed through is kept alive. Only dependencies written in source code are cached this way: functions and classes, and annotations built from them. One made at runtime — `Given(...)`, a `MakeInjected` or `bind_deps` result, an object that brings its own `Depends` — is analysed again on every resolve, and FastAPI never gets to keep it: what it carries is released with the resolve instead of living as long as the process. `clear_dependant_cache()` drops the cache, for long-lived processes and test suites that want the memory back.

### Overriding dependencies

`push_overrides` swaps dependencies out for the duration of a `with` block — handy in tests, or anywhere you need to run the same code against a different implementation:

```python
from fastapi_injected import push_overrides

with push_overrides({Session: Session(closed=True)}):
    await handler()  # gets the override instead of the real dependency
```

A key can be the dependency itself, or the annotation you wrote in the signature — `Dep[Session]` and `DepFactory[Session, session_dep]` both work and are normalized to the same underlying dependency:

```python
with push_overrides({DepFactory[Session, session_dep]: my_session}):
    ...
```

Values are used as-is, but two wrappers make the intent explicit and cover the ambiguous cases:

- `ValueOverride(value)` — always inject `value`, even when it is itself callable.
- `FactoryOverride(factory)` — call `factory` to produce the value. Sync, async, and generator factories are all supported, with the same teardown semantics as regular dependencies.

```python
from fastapi_injected import FactoryOverride, ValueOverride

with push_overrides(
    {
        Session: ValueOverride(fake_session),
        Repository: FactoryOverride(lambda: FakeRepository()),
    },
):
    ...
```

Overrides apply to the whole graph, not just top-level parameters — overriding a nested dependency changes what its dependents receive. Nested `push_overrides` blocks merge, with the innermost one winning.

An override block is a scope of its own: it caches what it builds, and that cache goes away with the block. The surrounding scope keeps what it had resolved before, and it stays visible — except for the overridden dependencies and anything built from them, which are resolved again under the override:

```python
async with push_inject_scope():
    real = await handler()  # dependencies cached here

    with push_overrides({Session: fake}):
        await handler()  # rebuilt with `fake`, everything else reused from the cache

    await handler()  # back to the cached, real dependencies
```

Overrides can also be handed to the scope directly, which is the same thing in one call:

```python
async with push_inject_scope({Session: fake}) as scope:
    await handler()

    with scope.override({Session: other}):  # nest freely
        await handler()
```

### Dependencies that are objects

Resolved values are cached in their scope by the callable that resolves them, so a dependency that is an object — a class holding configuration, a parametrized resolver — has to be hashable to get there. A plain dataclass is not, and a frozen one still refuses as soon as it holds a list or a dict.

`MakeDataclass` is a base class that makes its subclasses dataclasses with a hash that always answers: by fields when they can be hashed, by identity when they cannot.

```python
from dataclasses import field
from fastapi_injected import MakeDataclass, resolve


class Settings(MakeDataclass):
    hosts: list[str] = field(default_factory=list)

    def __call__(self) -> list[str]:
        return self.hosts


await resolve(Settings(["a", "b"]))  # a plain dataclass would raise TypeError here
```

Every `dataclasses.dataclass` option is accepted as a class keyword and passed straight through:

```python
class Config(MakeDataclass, frozen=True, kw_only=True):
    retries: int = 3
```

### When resolution fails

A dependency can fail to resolve for the same reasons it would in a route — a missing header, a query parameter that does not validate. `DependencyResolutionError` carries those errors in the shape pydantic produced them, and turns into the response FastAPI would have returned:

```python
from fastapi_injected import DependencyResolutionError

try:
    await handler()
except DependencyResolutionError as exc:
    exc.errors  # [{'type': 'missing', 'loc': ('header', 'x-trace'), ...}]
    raise exc.as_validation_error() from exc  # a RequestValidationError, so a 422
```

It is a `ValueError`, which is what resolution used to raise, so code catching that keeps working.

### Dependencies decided at runtime

A dependency is usually written as an annotation, but sometimes it is a value a caller already has, or one picked while the program runs. `Given` turns a value into a dependency that resolves to it:

```python
from fastapi_injected import Given, resolve

doc = Doc(title="readme")

await resolve(Given(doc))  # the doc itself
```

Constants holding equal values are the same dependency, so they share a cache entry and can be overridden like any other. What `Given` returns is a marker held as a value — `DepOf[R]`, as opposed to `Dep[R]`, which is the same marker in annotation position.

`bind_deps` binds such markers to the leading parameters of a function, which is how a dependency graph gets built from markers that were not known when the function was written:

```python
from fastapi_injected import bind_deps

async def describe(doc: Doc, role: Role) -> str:
    return f"{doc.title}:{role.name}"


bound = bind_deps(describe, Given(doc), RoleDep)

await resolve(bound)  # "readme:admin", with `role` resolved as usual
```

What a parameter is bound to can be written any way a dependency can be: a marker like `Given(...)`, `Dep[T]` or `DepFactory[T, factory]`, a `Depends(...)`, or the callable itself — the same things `resolve` accepts.

Binding takes the parameters away, and a type checker sees what is left: `bind_deps(describe, Given(doc))` is a function of `(role)`, and binding both leaves one of no arguments at all. The bound function is a dependency like any other — resolve it, `@inject` it, override what it depends on. Binding more markers than the function has parameters is a `TypeError` rather than a silent truncation.

### Dependencies as objects

When the markers belong together, `MakeInjected` makes a dataclass of them: fields annotated with `DepOf` hold the markers, and the call receives what they resolve to.

```python
from fastapi_injected import Given, MakeInjected
from fastapi_injected.types import DepOf


class IsOwner(MakeInjected):
    doc: DepOf[Doc]
    role: DepOf[Role]
    loud: bool = False        # a plain field stays a plain field

    async def __call__(self, doc: Doc, role: Role) -> bool:
        return doc.owner == role.name


await resolve(IsOwner(doc=Given(doc), role=RoleDep))
```

It is a `MakeDataclass`, so instances built from equal markers are equal — and therefore one dependency, resolved once per scope and overridable as a whole. For callables that are not dataclasses, the two halves of the binding are available on their own: `signature_with_deps(func, deps)` builds the signature, and `remap_dep_args` turns the arguments it names back into positional ones.

### Inspecting annotations

A few helpers are exported for code that needs to reason about `Dep[...]` annotations — building override maps, custom decorators, and the like:

- `is_dep(tp)` — whether `tp` is a `Dep`/`DepFactory` annotation.
- `unwrap_dep_tp(tp)` — the annotated type (`Any` for a bare `Dep`).
- `unwrap_dep_dependency(tp)` — the callable that resolves it: the factory for `DepFactory[T, factory]`, the type itself for `Dep[T]`.

### FastAPI request integration

Inside a FastAPI app, `@inject`-ed functions and `resolve` can share the request's own dependency cache — the same instances FastAPI built for the handler. Register `init_inject_scope` as a dependency:

```python
from fastapi import Depends, FastAPI
from fastapi_injected import Dep, init_inject_scope, resolve

app = FastAPI(dependencies=[Depends(init_inject_scope)])


@app.get("/")
async def route(service: Dep[Service]) -> str:
    same = await resolve(Service)  # same instance as `service`
    ...
```

Anything called from the handler — including `@inject`-ed helpers — resolves against the request's cache, so a per-request dependency like a DB session stays a single instance for the whole request.

WebSocket routes work the same way: the scope is bound to the `WebSocket` instead of a `Request`, and dependencies that ask for one get it. A websocket handed to a dependency shares its handshake with the one the route received, so it can be talked through after the route has called `accept()`:

```python
async def notify(websocket: WebSocket) -> None:
    await websocket.send_json({"ready": True})


@app.websocket("/ws")
async def ws_route(websocket: WebSocket, service: Dep[Service]) -> None:
    await websocket.accept()
    await resolve(notify)  # sends through the connection the route accepted
```

Dependencies that take request parameters — `Header()`, `Cookie()`, `Query()`, `Path()`, and the body as `Body()`, `Form()`, `File()` or a pydantic model — resolve from the request the scope is bound to, exactly as they would in the route:

```python
class Item(BaseModel):
    name: str


async def get_item(item: Item) -> Item:
    return item


@app.post("/items")
async def create(request: Request) -> str:
    item = await resolve(get_item)  # the request body, parsed as FastAPI would
    ...
```

A request parameter is a dependency of its own as well - resolve a marker, or an annotation carrying one, without writing a function around it. On its own it has no parameter name, so the marker says what to read; a marker without an alias raises `UnnamedParamError`:

```python
api_key = await resolve(Header(alias="x-api-key"))  # the raw header, as `Any`
page = await resolve(Annotated[int, Query(alias="page")])  # validated as `int`
item_id = await resolve(Annotated[int, Path(alias="item_id")])
item = await resolve(Annotated[Item, Body()])  # the whole body needs no name


@inject
async def handler(*, api_key: Annotated[str, Header()] = Injected) -> str:
    return api_key  # a parameter has a name - `x-api-key`, as FastAPI derives it
```

The body is read only when the resolved dependency declares one, and it is read once per request, shared with the route and every scope nested in it. When the route itself declares a body, dependencies see it in the shape the route gave it: a route with several body parameters embeds them under their names, and a dependency reads its own from there.

`push_inject_scope(request=...)` accepts either a `Request` or a `WebSocket`, and so does everything that reads `InjectScope.request` — the union is `fastapi_injected.types.BoundConnection`.

## What is public

Everything the package supports is importable from `fastapi_injected` itself, and that is the surface a release keeps:

| | |
| --- | --- |
| Markers | `Dep`, `DepFactory`, `DepOf`, `Arg`, `Given`, `Injected` |
| Resolving | `inject`, `resolve`, `bind_deps`, `signature_with_deps`, `remap_dep_args`, `clear_dependant_cache` |
| Scopes and overrides | `InjectScope`, `push_inject_scope`, `inside_inject_scope`, `synthetic_request`, `push_overrides`, `Overrides`, `OverridesProvider`, `ValueOverride`, `FactoryOverride` |
| Building on top | `MakeDataclass`, `MakeInjected`, `HasDependsHook`, `ArgMarker`, `is_arg`, `is_dep`, `unwrap_dep_tp`, `unwrap_dep_dependency` |
| FastAPI integration | `add_injected_scope`, `init_inject_scope` |
| Errors | `DependencyResolutionError`, `MissedDependencyError`, `MissingDependencyCacheError`, `NotADependencyError`, `UnboundDepArgsError`, `UnboundScopeError`, `UnnamedParamError` |

`fastapi_injected.types` holds the typing vocabulary the signatures are written in — `DepReturn`, `DepShape`, `DepDecl`, `AsyncFunc`, `Coro` and friends — and is public too. Anything else, including every module whose name starts with an underscore, is machinery that can change in a patch release.

## License

MIT
