/
planver
/
planver-main
Обзор
Документация
Войти
/
planver
/
planver-main
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
master
src/core/db_postgres.py
52 строки
1 KB
waldemarX
polish cats
18 ноя 2025, 21:01
18 ноя 2025, 21:01
8397003
Код
Авторство
О чём код?
from collections.abc import AsyncGenerator from typing import Annotated from fastapi import Depends from sqlalchemy import MetaData from sqlalchemy.orm import DeclarativeBase from sqlalchemy.ext.asyncio import ( AsyncSession, create_async_engine, async_sessionmaker, AsyncAttrs, ) from src.core.db_base_map import BaseDBMap from src.config import settings engine = create_async_engine( url=settings.DB_PG_URL, echo=settings.postgres_echo, pool_recycle=280 ) sessionmaker = async_sessionmaker( bind=engine, expire_on_commit=False, ) db_meta = MetaData(schema=settings.postgres_schema) class BaseDBModel(DeclarativeBase, AsyncAttrs): metadata = db_meta class BaseDB(BaseDBModel, BaseDBMap): """ Базовый класс для всех моделей """ __abstract__ = True async def get_db() -> AsyncGenerator[AsyncSession, None]: async with sessionmaker() as session: yield session async def recreate_tables(): async with engine.begin() as conn: # await conn.run_sync(BaseDBModel.metadata.drop_all) await conn.run_sync(BaseDBModel.metadata.create_all) CurrentSession = Annotated[AsyncSession, Depends(BaseDBMap._get_db)]