Легкий путь в Python



        def decorator(method):
            @wraps(method)
            async def wrapper(*args, **kwargs):
                async with self.async_session_maker() as session:
                    try:
                        # Устанавливаем уровень изоляции, если он указан
                        if isolation_level:
                            await session.execute(text(f"SET TRANSACTION ISOLATION LEVEL {isolation_level}"))

                        # Выполняем декорируемый метод
                        result = await method(*args, session=session, **kwargs)

                        # Выполняем коммит, если параметр commit=True
                        if commit:
                            await session.commit()

                        return result
                    except Exception as e:
                        await session.rollback()  # Откатываем транзакцию при ошибке
                        raise e  # Поднимаем исключение дальше
                    finally:
                        await session.close()  # Закрываем сессию

            return wrapper

        return decorator

    @property
    def session_dependency(self) -> Callable:
        """
        Зависимость FastAPI для внедрения сессии без управления транзакцией.

        Возвращает:
        - Зависимость для FastAPI, которая предоставляет асинхронную сессию `AsyncSession`.
        """
        return Depends(self.get_db)

    @property
    def transaction_session_dependency(self) -> Callable:
        """
        Зависимость FastAPI для внедрения сессии с управлением транзакцией.

        Возвращает:
        - Зависимость для FastAPI, которая предоставляет асинхронную сессию `AsyncSession`
          с автоматическим управлением транзакцией (коммит или откат).
        """
        return Depends(self.get_db_with_transaction)


# Создание экземпляра DatabaseSessionManager
session_manager = DatabaseSessionManager(async_session_maker)

# не делает коммит
SessionDep = session_manager.session_dependency

# делает коммит
TransactionSessionDep = session_manager.transaction_session_dependency


# Пример использования декоратора
# @db_manager.connection(isolation_level="SERIALIZABLE", commit=True)
# async def example_method(*args, session: AsyncSession, **kwargs):
#     # Основная логика метода
#     pass

Штуку одну готовлю универсальную. Решил что неудобно будет прыгать между получением сессии под FastApi и под обычные функции / хендлеры телеграмм ботов. Сегодня протестирую ее. Если коротко, то будет один пакет dao, который будет закрывать вообще все случаи жизни для использования SQLAlchemy)

🔥17❤3👍3

1.95K views17:17