/
urusai
/
code_sample
Обзор
Документация
Войти
/
urusai
/
code_sample
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/db/repositories/base.py
55 строк
2 KB
nick
linters
30 сен 2024, 16:13
30 сен 2024, 16:13
09a0f4a
Код
Авторство
О чём код?
import logging from typing import Type, Union, TypeVar, Optional, AsyncContextManager, cast from contextlib import asynccontextmanager from sqlalchemy import func, delete, exists, select, update, lambda_stmt from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.asyncio import ( AsyncSession, AsyncSessionTransaction, async_sessionmaker, ) Model = TypeVar("Model") TransactionContext = AsyncContextManager[AsyncSessionTransaction] logger = logging.getLogger(__name__) class BaseSession: def __init__( self, db_session: Union[sessionmaker, AsyncSession], ): if isinstance(db_session, async_sessionmaker): self.session: AsyncSession = cast(AsyncSession, db_session()) else: self.session = db_session def transaction(self) -> AsyncContextManager[AsyncSession]: @asynccontextmanager async def wrapped() -> AsyncContextManager[AsyncSession]: async with self.session as session: # type: AsyncSession try: yield session except Exception as exc: # noqa logger.error(f"Произошла ошибка в транзакции: {exc}") await session.rollback() raise exc finally: await session.commit() return wrapped() class BaseCRUD(BaseSession): def __init__( self, model: Type[Model], db_session: Union[sessionmaker, AsyncSession] ): super().__init__(db_session) self.model = model async def get_one(self, *args, transaction: Optional[AsyncSession] = None) -> Model: stmt = select(self.model).where(*args) cursor = await self.execute(stmt=stmt, transaction=transaction) return cursor.scalar_one()