How to get a session from async_session() generator FastApi Sqlalchemy-CodePudding

I see in many places an approach for getting SqlAlchemy session just like this one below:

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

It used together with Depends:

@app.post("/endpoint")
async def vieww(session: AsyncSession = Depends(get_session)):
    session.execute(some_statement)

So my question is how to get a session from get_session ooutside Depends?

I had a lot of attempts and got a headache.. I tried with

s = await get_session()
s.execute(stmt)

And I get AttributeError: 'async_generator' object has no attribute 'execute'

CodePudding user response：

get_session is an asynchronous generator function. It returns an asynchronous iterator. You can't await those. You can await their __anext__ method though.

from asyncio import run
from collections.abc import AsyncIterator


async def get_generator() -> AsyncIterator[int]:
    yield 1


async def main() -> None:
    generator = get_generator()
    print(await generator.__anext__())


if __name__ == "__main__":
    run(main())

Output: 1

Since Python 3.10 we can use the built-in anext, which makes this look less hacky:

...
async def main() -> None:
    generator = get_generator()
    print(await next(generator)
...

So in your case, you could use s = await anext(get_session()).

But I struggle to see the use of this. This get_session function is designed as an asynchronous generator for the specific use case in the FastAPI Depends construct. Outside of that, you may as well just use the context manager normally:

async with async_session() as session:
    ... # do things with the session
    session.execute()

Otherwise you'll have to remember to call __anext__ on that generator again (and catch the StopAsyncIteration error), otherwise the session will never be closed.