FastAPI dependencies (yield): how to call them manually?

Question

FastAPI uses Depends() to inject variables either returned or yielded. Eg, FastAPI/SQL:

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
...
def create_user(db: Session = Depends(get_db)):
...

If I wanted to use that get_db() somewhere else (outside a FastAPI route), how would I do that? I know it's Python core knowledge, but I can't seem to figure it out. My initial thought was db = yield from get_db(), but I can't call yield from in async functions (and don't know if it would work besides). Then I tried:

with get_db() as db:
   pass

Which fails as the original get_db() isn't wrapped as a @contextmanager. (Note, I don't want to decorate this - I'm using get_db as an example, I need to work with more complicated dependencies). Finally, I tried db = next(get_db()) - which works, but I don't think that's the correct solution. When/how will finally be invoked - when my method returns? And in some other dependencies, there's post-yield code that needs to execute; would I need to call next() again to ensure that code executes? Seems like next() isn't the right way. Any ideas?

score 16 · Answer 1 · answered Mar 10 '21 at 09:00

16

You can use contextmanager not as a decorator but as a function returning context manager:

from contextlib import contextmanager

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


# synchronously
with contextmanager(get_db)() as session:  # execute until yield. Session is yielded value
    pass
# execute finally on exit from with

But keep in mind that the code will execute synchronously. If you want to execute it in a thread, then you can use the FastAPI tools:

import asyncio
from contextlib import contextmanager

from fastapi.concurrency import contextmanager_in_threadpool


async def some_coro():
    async with contextmanager_in_threadpool(contextmanager(get_db)()) as session:
        pass

answered Mar 10 '21 at 09:00

alex_noname

26,459
5
69
86

1

Oi, I'd never have figured that out - glad I asked! Thanks – lefnire Mar 10 '21 at 19:41
Why don't we use context manager as a decorator for get_db function and then use `with get_db() as session` – saeedghadiri Nov 12 '22 at 16:52
What if the `get_db` is async as well? – Pratik K. Feb 13 '23 at 22:24

score 0 · Answer 2 · answered Sep 01 '23 at 08:55

If you try to use next(get_db()) outside of this context, you might encounter problems such as:

No Database Session: The get_db() function might not return a valid session because it relies on FastAPI's dependency injection system.

Session Leaks: Manually managing the database session's lifecycle outside of the request context can lead to resource leaks if you forget to close sessions properly.

Thread Safety: Depending on how your FastAPI application is deployed (e.g., using multiple worker processes or threads), manually managing the session may not be thread-safe.

If you need to perform database operations outside of FastAPI route handlers or middleware (e.g., in a script or background task), it's better to create and manage the database session explicitly without relying on FastAPI's Depends mechanism. You can use the sessionmaker from SQLAlchemy to create sessions and handle their lifecycle as needed:

from sqlalchemy.orm import sessionmaker

SessionLocal =    sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Usage example
def some_function():
    # Create a session
    session = SessionLocal()

    try:
        # Perform database operations     using the session
        user = session.query(User).filter_by(username="example_user").first()
        # Other database operations here
   
        # Commit the transaction (if applicable)
        session.commit()
    except Exception as e:
        # Handle exceptions, rollback on error, and/or log
        session.rollback()
    finally:
        # Close the session
        session.close()

Call the function to perform database operations some_function()

FastAPI dependencies (yield): how to call them manually?

2 Answers2

Linked