FastAPI¶

Though it is not required, you can use dishka-fastapi integration. It features:

automatic REQUEST and SESSION scope management using middleware
passing Request object as a context data to providers for both Websockets and HTTP requests
automatic injection of dependencies into handler function.

How to use¶

Note

We suppose that you are using AsyncContainer with FastAPI. If it is not right and you are using sync Container, the overall approach is the same, but there are few corrections, see below.

Import

from dishka.integrations.fastapi import (
    DishkaRoute,
    FromDishka,
    FastapiProvider,
    inject,
    setup_dishka,
)
from dishka import make_async_container, Provider, provide, Scope

Create provider. You can use fastapi.Request as a factory parameter to access on REQUEST-scope, and fastapi.WebSocket on SESSION-scope

class YourProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def create_gateway(self, request: Request) -> Gateway:
         ...

2a. (optional) Set route class to each of your fastapi routers to enable automatic injection (it works only for HTTP, not for websockets)

router = APIRouter(route_class=DishkaRoute)

Mark those of your handlers parameters which are to be injected with FromDishka[]

@router.get('/')
async def endpoint(
    request: str, gateway: FromDishka[Gateway],
) -> Response:
    ...

3a. (optional) decorate them using @inject if you are not using DishkaRoute or use websockets

@router.get('/')
@inject
async def endpoint(
    gateway: FromDishka[Gateway],
) -> ResponseModel:
    ...

(optional) Use FastapiProvider() when creating container if you are going to use fastapi.Request or fastapi.WebSocket in providers

container = make_async_container(YourProvider(), FastapiProvider())

(optional) Setup lifespan to close container on app termination

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await app.state.dishka_container.close()

app = FastAPI(lifespan=lifespan)

Setup dishka integration.

setup_dishka(container=container, app=app)

Using with `fastapi.Depends`¶

To inject Dishka-managed dependencies into fastapi.Depends functions, declare a factory function using FromDishka[] and decorate it with @inject.

@inject
def get_dependency(
    gateway: FromDishka[Gateway],
) -> Dependency:
    ...

@router.get('/')
@inject
async def endpoint(
    dependency: Annotated[Dependency, Depends(get_dependency)],
) -> ResponseModel:
    ...

Note

The @inject decorator is required for Depends functions even when using DishkaRouter.

Websockets¶

For most cases we operate single events like HTTP-requests. In this case we operate only 2 scopes: APP and REQUEST. Websockets are different: for one application you have multiple connections (one per client) and each connection delivers multiple messages. To support this we use additional scope: SESSION:

APP → SESSION → REQUEST

In FastAPI your view function is called once per connection and then you retrieve messages in loop. So, inject decorator can be only used to retrieve SESSION-scoped objects. To achieve REQUEST-scope you can enter in manually:

@inject
async def get_with_request(
    websocket: WebSocket,
    a: FromDishka[A],  # object with Scope.SESSION
    container: FromDishka[AsyncContainer],  # container for Scope.SESSION
) -> None:
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        # enter the nested scope, which is Scope.REQUEST
        async with container() as request_container:
            b = await request_container.get(B)  # object with Scope.REQUEST

Auto-injection is not available for websockets.

Using with sync Container¶

If you are using sync Container, created using make_container function, you should change several things:

use inject_sync instead of inject function
use DishkaSyncRoute instead of DishkaRoute if you are using autoinjection
use Container in websocket handler instead of AsyncContainer and normal with instead of async with.

from dishka.integrations.fastapi import (
    FromDishka,
    FastapiProvider,
    inject_sync,
    setup_dishka,
)
from dishka import make_container, Provider, provide, Scope


router = APIRouter()


@router.get('/')
@inject_sync
def endpoint(
    gateway: FromDishka[Gateway],
) -> ResponseModel:
    ...


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    app.state.dishka_container.close()

app = FastAPI(lifespan=lifespan)
app.include_router(router)
container = make_container(YourProvider(), FastapiProvider())
setup_dishka(container=container, app=app)

FastAPI¶

How to use¶

Using with fastapi.Depends¶

Websockets¶

Using with sync Container¶

Using with `fastapi.Depends`¶