/
azathd
/
mutiagent
Обзор
Документация
Войти
/
azathd
/
mutiagent
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/common/db.py
38 строк
1 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47
359c81e
Код
Авторство
О чём код?
"""Async SQLAlchemy engine + session factory.""" from __future__ import annotations import os from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine def postgres_dsn(env_var: str = "DATABASE_URL") -> str: url = os.environ.get(env_var) if not url: default = ( "postgresql+asyncpg://app:changeme_secret@postgres:5432/" + os.environ.get("POSTGRES_DB", "xray_multiagent") ) return default if url.startswith("postgresql://"): return url.replace("postgresql://", "postgresql+asyncpg://", 1) return url def init_engine(pool_size: int = 5): """Create SQLAlchemy async engine.""" return create_async_engine( postgres_dsn(), pool_size=pool_size, pool_pre_ping=True, future=True, ) def async_session_fact(engine=None) -> async_sessionmaker[AsyncSession]: if engine is None: engine = init_engine() return async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)