Custom Router Class - How to get all Router Function Input Parameters

[mtgraves]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import List, Callable

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class CustomRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:

            # HOW CAN I GET THE INPUTS TO sum_numbers FUNCTION HERE?

            _r = await original_route_handler(request)

            return _r

        return custom_route_handler


def create_app():
    app = FastAPI()
    app.router.route_class = CustomRoute

    return app


app = create_app()


@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body(), things: str = "stuff"):
    return {"sum": sum(numbers), "things": things}

Description

Is there a way to get the input parameters for a fastapi APIRoute function when defining a custom router class (per the docs)?

For example, in the example below, if the things variable wasn't sent in the request, could I still get a hold of the value passed to it in my custom router class?

Operating System

Windows

Operating System Details

No response

FastAPI Version

0.110.0

Pydantic Version

1.10.14

Python Version

3.10.11

Additional Context

No response

[YuriiMotov]

As I understand, custom APIRoute is actually like middleware. It wraps your request handler. And request validation is only happens after you call _r = await original_route_handler(request).
So, you can't get access to validated input before you call original_route_handler.

But you can read raw input data (not validated yet) with await request.body(), request.query_params, request.path_params and so on..
https://www.starlette.io/requests/

[YuriiMotov]

        async def custom_route_handler(request: Request) -> Response:
            things = request.query_params.get("things", None)
            numbers = await request.json()
            print(f"{things=}, {numbers=}")

            _r = await original_route_handler(request)
            return _r

But remember, that this data is not validated yet at that point.
So, if you call your endpoint without specifying things parameter, things in the code above will be None

[mtgraves]

ok - specifically, one of the things I want to get a hold of is the db: Session = Depends(get_db) as from this part of the tutorial: https://fastapi.tiangolo.com/tutorial/sql-databases/

Is this possible?

[YuriiMotov]

I don't think it's possible.. At least in easy way.

Can you describe your use-case? What do you want to achieve?

[mtgraves]

@YuriiMotov

Please see the inline comments. Essentially here's a toy model of what i am attempting. Was hoping to move the functionality out of that decorator that I created and into a custom route class, so that I wouldn't have to decorate all my routes/routers but instead all would just get the same functionality.

from typing import Callable
from functools import wraps
from sqlalchemy import event

from fastapi import FastAPI, Request, Response, Depends
from fastapi.routing import APIRoute
from pydantic import BaseModel

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy import Column, Integer, String

engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


class Things(Base):
    __tablename__ = "things"

    id = Column(Integer, primary_key=True)
    stuff = Column(String)


class ThingsModel(BaseModel):
    id: int
    stuff: str

    class Config:
        orm_mode = True


def detect_database_changes(_f):
    @wraps(_f)
    async def wrapper(*args, **kwargs):

        # -------------------------------------------------------------
        # WAS REALLY HOPING TO MOVE THIS FUNCTIONALITY.....
        db_session = kwargs["db"]

        def receive_before_flush(session, flush_context, instances):
            for _n in session.new:
                print("detected a new thing!")
                print(_n)

        event.listen(db_session, "before_flush", receive_before_flush)
        # -------------------------------------------------------------

        _r = await _f(*args, **kwargs)

        event.remove(db_session, "before_flush", receive_before_flush)

        return _r

    return wrapper


class CustomRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:

            # -------------------------------------------------------------
            # .... INTO HERE!
            # ...tried the following and it doesn't work the same:
            db_session = next(get_db())

            def receive_before_flush(session, flush_context, instances):
                for _n in session.new:
                    print("detected a new thing!")
                    print(_n)

            event.listen(db_session, "before_flush", receive_before_flush)
            # -------------------------------------------------------------

            _r = await original_route_handler(request)

            return _r

        return custom_route_handler


def create_app():
    app = FastAPI()
    app.router.route_class = CustomRoute
    Base.metadata.create_all(bind=engine)

    return app


app = create_app()


@app.get("/things", response_model=list[ThingsModel])
async def get_things(db: Session = Depends(get_db)):
    things = db.query(Things).all()
    return things


@app.post("/things", response_model=ThingsModel)
@detect_database_changes
async def create_thing(thing: ThingsModel, db: Session = Depends(get_db)):
    new_thing = Things(id=thing.id, stuff=thing.stuff)
    db.add(new_thing)
    db.commit()

    return new_thing

[YuriiMotov]

When you call db_session = next(get_db()) you get a session object, but it's another instance of session (not the same that will be passed into the endpoint function). So this is not an option..

I don't see any good ways to implement that in the custom route class..
Why don't you move that code into the get_db function?

Or, you can create another dependency, that will inject this logic and use this dependency instead of get_db when you need it:

async def get_db_with_logging(db_session: Session = Depends(get_db)):
    def receive_before_flush(session, flush_context, instances):
        for _n in session.new:
            print("detected a new thing!")
            print(_n)

    event.listen(db_session, "before_flush", receive_before_flush)
    yield db_session



@app.get("/things", response_model=list[ThingsModel])
async def get_things(db: Session = Depends(get_db)):
    things = db.query(Things).all()
    return things


@app.post("/things", response_model=ThingsModel)
async def create_thing(thing: ThingsModel, db: Session = Depends(get_db_with_logging)):
    new_thing = Things(id=thing.id, stuff=thing.stuff)
    db.add(new_thing)
    db.commit()

    return new_thing