api/database.py

Async SQLAlchemy engine and session factory for FastAPI, supporting multiple DB dialects.

# Source: https://github.com/agents-universe/agents-universe | License: Apache-2.0 | Copyright You must retain, in the Source form of any Derivative Works
# Provided via justclaw.it.com; keep this notice with the code.
"""SQLAlchemy async engine and session factory."""
from __future__ import annotations

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from .config import get_settings

settings = get_settings()

_db_url = settings.effective_db_url
_is_sqlite = _db_url.startswith("sqlite")
_engine_kwargs: dict = {"echo": False}
if not _is_sqlite:
    _engine_kwargs.update({
        "pool_pre_ping": True,
        "pool_size": settings.db_pool_size,
        "max_overflow": settings.db_max_overflow,
        "po

... (truncated -- full source via MCP)