Fastapi: [QUESTION] Middleware and dependencies

[DrPyser]

Description

I'm wondering how middlewares and dependencies can work together, if at all.

That is, I'd like to exploit dependencies(and dependency caching) in my middleware, but I'm not sure that's supported right now?

Basic idea is:

• my middleware declares, or otherwise obtain the dependency. The dependency is cached(if configured so in the declaration).
• A route also declares the same dependency. The dependency was cached by the middleware, so the same value is provided.

For example:

Session = sqlalchemy.orm.sessionmaker()
def get_database():
    return sqlalchemy.create_engine(os.getenv("DATABASE_URL"))

@app.middleware("http")
async def get_org_repo(request: Request, call_next, database = Depends(get_database)):
    try:
        session = Session(bind=database)
        request.state.session = session
        response = await call_next(request)
    finally:
        # cleanup
        session.close()
    return response

[tiangolo]

Middlewares are handled at a lower level, by Starlette.

But you can declare a router-level dependency that is applied to all the sub routes/path operations in it, achieving more or less the same I understand you want: https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-prefix-tags-responses-and-dependencies

[DrPyser]

In my example, I use the Middleware to tie resource lifecycle to the request, which I don't think dependencies can replicate.

[dmontagu]

@DrPyser Below, I've included a stab at an approach that could implement dependencies for "MiddlewareResource"s. The main feature loss is arbitrary dependency functions; to get around this I've just required that any Depends of a MiddlewareResource is itself a MiddlewareResource.

Before including the implementation, here is an example showing the API:

class DatabaseResource(MiddlewareResource):
    def __init__(self, request: Request) -> None:
        self.database = sqlalchemy.create_engine(os.getenv("DATABASE_URL"))


class SessionResource(MiddlewareResource):
    session_maker = sqlalchemy.orm.sessionmaker()

    def __init__(self, request: Request, database_resource: DatabaseResource = Depends()) -> None:
        self.session = self.session_maker(bind=database_resource.database)

    def clean_up(self):
        self.session.close()

Subclasses of MiddlewareResource behave like singletons that are scoped to a resource instance -- calling Resource(request) multiple times on the same request returns the same instance.

This allows you to then have "traditional" FastAPI dependencies that look like this:

def get_database(request: Request):
    return DatabaseResource(request).database


def get_session(request: Request):
    return SessionResource(request).session

In order to use MiddlewareResource instances, you need to add the following middleware:

@app.middleware("http")
async def cleanup_middleware(request: Request, call_next):
    try:
        request.state.resources: Dict[Type[MiddlewareResource], MiddlewareResource] = {}
        response = await call_next(request)
    finally:
        for resource in request.state.resources.values():
            resource.clean_up()
    return response

---

Here's the implementation; it probably could benefit from a little additional tweaking, so use in its current state at your own peril 😬:

from abc import ABCMeta
from contextlib import suppress
import inspect
import os
from typing import Any, Dict, Type

from fastapi.params import Depends
from pydantic.utils import lenient_issubclass
import sqlalchemy.orm
from starlette.requests import Request


class MiddlewareResourceMeta(ABCMeta):
    def __call__(cls, request: Request, *args: Any, **kwargs: Any) -> Any:
        resources = request.state.resources
        if cls not in resources:
            cls_init_kwargs: Dict[str, Any] = {"request": request}
            parameters = inspect.signature(cls.__init__).parameters
            for position, (keyword, param) in enumerate(parameters.items()):
                if position == 0 or keyword == "request":
                    continue
                with suppress(TypeError):
                    cls_init_kwargs[keyword] = param.annotation(request)
            resources[cls] = super().__call__(**cls_init_kwargs)
        return resources[cls]


class MiddlewareResource(metaclass=MiddlewareResourceMeta):
    def __init_subclass__(cls):
        """
        Validate the subclass __init__ signature
        """
        init_parameters = inspect.signature(cls.__init__).parameters
        if "request" not in init_parameters or init_parameters["request"].annotation is not Request:
            raise TypeError(f"{cls.__name__}.__init__ must have the argument `request: Request` in its signature")
        if init_parameters["request"].default is not inspect.Parameter.empty:
            raise TypeError(f"The `request` parameter for {cls.__name__}.__init__ must not have a default value")
        for position, (keyword, param) in enumerate(init_parameters.items()):
            if position == 0 or keyword == "request":
                continue  # ignore `self` argument and `request` argument
            default = param.default
            annotation = param.annotation
            if default is not inspect.Parameter.empty:
                if not isinstance(default, Depends):
                    raise ValueError(f"`Depends()` is the only valid default parameter value for {cls}.__init__")
                if default.dependency is not None:
                    raise ValueError(f"Cannot use callable dependencies in {cls}.__init__ (must be Depends())")
            if not lenient_issubclass(annotation, MiddlewareResource):
                raise ValueError(
                    f"All parameters of {cls.__name__}.__init__ should be type-hinted as a MiddlewareResource"
                )

    def __new__(cls, request: Request, *args, **kwargs):
        if cls is MiddlewareResource:
            raise TypeError(f"Must subclass {cls.__name__} before instantiating")
        return super().__new__(cls)

    def clean_up(self):
        pass

[DrPyser]

@dmontagu Sorry, I had missed your response.
This does seem like a perfectly valid approach, thanks a lot!

What would be really nice, is for fastapi's dependency framework to be extracted from fastapi and offered as a standalone library. I haven't found another implementation of a dependency injection framework in python that is as seamless and easy to use.
Perhaps with a manual entrypoint, like resolve_dependencies(f) -> {"dep1": ..., "dep2": ...}.

[dmontagu]

@DrPyser you might check this out: https://github.com/dry-python/dependencies I haven't used it but have been meaning to try it out -- it seems pretty nice!

[tiangolo]

@DrPyser in your original example, you wanted to have dependencies to create a DB session and then close it at the end, right?

In recent versions of FastAPI there are dependencies with yield that you can use for exactly that: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/

The new SQL tutorial uses them: https://fastapi.tiangolo.com/tutorial/sql-databases/

That would probably be the best way to handle something as part of a single request, including teardown/exit code.

[github-actions[bot]]

Assuming the original issue was solved, it will be automatically closed now. But feel free to add more comments or create new issues.