FastAPI without db:Session

If you tried to do CRUD for FastAPI or take business logic out of controllers, then you may have seen or made such a terrible construction:

@router.get("/resources", response_model=List[ResourceResponse])
def read_resource_list(db: Session = Depends(get_db)) -> Any:
    return resource.get_multi(db)

  def get_multi(self, db: Session, *, skip: int = 0, limit: int = 100
               ) -> List[ModelType]:
      return db.query(self.model).offset(skip).limit(limit).all()

Similar code is written in the official documentation by fast API.
Terrible I call the design db: Session = Depends(get_db). Pass the session object to the controller to pass it to CRUD. This seems like a strange and ugly solution.

Now the application works like this:

This is what get_db() looks like

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

If we have one database, why not create a session object in CRUD? I decided to do this and I ran into a problem:

Depends(get_db)

An interesting thing, the functionality of which I did not fully understand. Previously used this class only to check request headers.

In this case, Depends is needed to get the session object from the generator. But since the session object is taken out of the controller, Depends cannot be used.

Custom Depends decorator

I’ll wrap get_db in a context manager and make a decorator to pass the session object to the function.

db_session = contextmanager(get_db)


def with_db_session(func):
    def wrapper(*args, **kwargs):
        with db_session() as db:
            self, *args = args
            return func(self, db, *args, **kwargs)

    return wrapper

Almost done

It remains to remove db: Session = Depends(get_db) from controllers and wrap CRUD in decorators.

@router.get("/resources", response_model=List[ResourceResponse])
def read_resource_list() -> Any:
    return resource.get_multi()

@with_db_session
def get_multi(self, db: Session, *, skip: int = 0, limit: int = 100
             ) -> List[ModelType]:
    return db.query(self.model).offset(skip).limit(limit).all()

Now the application works like this:

Result

Each request creates a new connection to the database (as it was). However, no data is sent via FastAPI. This allows you to change the framework without reworking CRUD.