-
Notifications
You must be signed in to change notification settings - Fork 3
/
session.py
28 lines (20 loc) · 913 Bytes
/
session.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
"""Database engine module for database interaction from business logic layer."""
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from aqueductcore.backend.settings import settings
async_engine = create_async_engine(
f"postgresql+asyncpg://{settings.postgres_username}:"
f"{settings.postgres_password}@{settings.postgres_host}:"
f"{settings.postgres_port}/{settings.postgres_db}"
)
async_session = async_sessionmaker(bind=async_engine, expire_on_commit=False)
@asynccontextmanager
async def get_session() -> AsyncGenerator[AsyncSession, None]:
"""Returns an async database session for SQLAlchemy ORM."""
async with async_session() as session:
try:
yield session
finally:
await session.close()