Starlette integration for Dishka

This package provides integration of Dishka dependency injection framework and Starlette, a lightweight ASGI framework.

Installation

pip install starlette-dishka

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

1. Import

from starlette_dishka import (
    FromDishka,
    StarletteProvider,
    inject,
    setup_dishka,
)
from dishka import make_async_container, Provider, provide, Scope

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

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

3. Mark those of your handlers parameters which are to be injected with FromDishka[] and decorate them using @inject

@inject
async def endpoint(
    request: Request,
    *,
    gateway: FromDishka[Gateway],
) -> ResponseModel:
    ...

4. (optional) Use StarletteProvider() when creating container if you are going to use starlette.Request or starlette.WebSocket in providers

container = make_async_container(YourProvider(), StarletteProvider())

5. Setup dishka integration.

setup_dishka(container=container, app=app)

Websockets

In starlette 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