SqlAlchemy

Unified Tutorial

• Unified Tutorial
• ORM Reference index of everything related to SqlAlchemy ORM
• ORM Mapped Classes using declarative mapping. The v2 syntax differs substantially from v1.x.

Sessions and SessionMaker

It is critically important to understand how SqlAlchemy Sessions work, to avoid errors in code. This reference page explains their use and the lifecycle: https://docs.sqlalchemy.org/en/20/orm/session_basics.html

Configure an Async Engine and AsyncSession

This article explains some parameters for pooling Async Database Sessions in FastAPI. The get_session and get_db code seems like overkill.

Get an AsyncSessionGenerator or asynccontextmanager?

Originally in database.py I implemented "get session" as:

class Database:
    self.asyncSessionMaker = async_sessionmaker(
            self.engine,
            expire_on_commit=False,   # SqlAlchemy recommends avoid expire on commit for async
            autoflush=False           # Default is False (I think). Not recommended for async.
            # class_=AsyncSession,
        )

    @asynccontextmanager
    async def get_session(self) -> AsyncGenerator[AsyncSession, None]:
        """Async context manager that yields a generator for creating AsyncSession
           and ensures it's closed after commit.

           :returns: async_sessionmaker[AsyncSession]
        """
        async with self.asyncSessionMaker() as session:
            try:
                yield session
                await session.commit()
            except Exception as ex:
                await session.rollback()
                raise
                # raise Exception(f"Database operation failed: {ex}")
            finally:
                await session.close()

In code, I use it in a with block. db is a reference to the shared Database instance:

async def create_user(username: str, email: str):
    async with db.get_session() as session:
        user = models.User(username=username, email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        print(f"Added user with id {user.id} on {user.created_at}")

But this technique fails if db.get_session is used with dependency injection in FastAPI:

# users "router" for FastAPI

@router.get("/users/{user_id}", response_model=schemas.User)
async def get_user(user_id: int, 
                   session: AsyncSession = Depends(db.get_session)):
    user = await user_dao.get_user_by_id(session, user_id)
    if user is None:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, 
                            detail=f"User id {user_id} not found"
                           )
    return user

The explanation is:

FastAPI expects a callable returning Awaitable or AsyncGenerator, not a context manager function.

So I removed the @asynccontextmanager annotation on get_session. But this breaks all the code using

    async with db.get_session() as session:
        perform orm operations in session