SQLAlchemy db (engine + session) wrapper

sa-1.4-adb.py

class AsyncSQLAlchemy:
    def __init__(
        self,
        url: str,
        engine_options: Optional[Dict[str, Any]] = None,
        session_options: Optional[Dict[str, Any]] = None,
    ):
        from sqlalchemy.orm import sessionmaker
        from sqlalchemy.ext.asyncio import AsyncSession

        self.url = url

        engine_options = engine_options or {}
        self.engine = create_async_engine(self.url, **engine_options)

        session_options = session_options or {}
        session_options.setdefault("class_", AsyncSession)
        session_options.setdefault("bind", self.engine)

        self.Session: Callable[..., AsyncSession] = sessionmaker(**session_options)

    def __repr__(self) -> str:
        return f"<AsyncSQLAlchemy('{self.url}')>"


adb = AsyncSQLAlchemy(
    f"postgresql+asyncpg://{dbc.user}:{dbc.password}@{dbc.host}:{dbc.port}/{dbc.database}",
    engine_options={
        "echo": dbc.echo_sql,
        'pool_size': dbc.pool_size,
        'max_overflow': dbc.max_overflow,
    },
    session_options={
        "expire_on_commit": False,
    }
)

async with adb.Session() as s:
    user = (await s.execute(sa.select(User).limit(1))).scalar()
    print(user)