Method session

backend/app/dao/base.py:20–37

Context manager yielding the active context session or a new one.

(self)

    @asynccontextmanager
    async def session(self) -> AsyncGenerator[AsyncSession, None]:
        """Context manager yielding the active context session or a new one."""
        context_session = _session_ctx.get()
        if context_session is not None:
            yield context_session
        else:
            async with async_session() as session:
                token = _session_ctx.set(session)
                try:
                    yield session
                    if hasattr(session, "commit"):
                        await session.commit()
                except Exception:
                    if hasattr(session, "rollback"):
                        await session.rollback()
                    raise
                finally:
                    _session_ctx.reset(token)

    async def get(self, id: Any) -> ModelType | None:
        """Fetch a single record by its primary key ID."""