Code Docs

class BaseAuthJWT:
    '''
    Пример:

    ```python
    class AuthJWT(BaseAuthJWT):
        async def check_auth(username: str, password: str) -> bool:
            """Проверка введенного логина и пароля."""
            return username == "admin" and password == "admin"

        async def add_jwt_body(username: str) -> dict:
            """Функция для добавление дополнительных данных в JWT токен пользователя"""
            return {"version": username.title()}


    # Подключить аутентификацию по JWT
    AuthJWT.mount_auth(app)
    ```

    Пример защиты API метода:

    ```python
    @app.get("/cheack_protected", summary="Проверить аутентификацию по JWT")
    async def protected_route(jwt: dict = Depends(jwt_auth)):
        return {"message": "This is a protected route", "user": jwt}
    ```
    '''

    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 30
    # Установиться автоматически в mount_auth
    secret_key = None

    @abc.abstractmethod
    async def check_auth(username: str, password: str) -> bool:
        """Проверка введенного логина и пароля."""
        raise NotImplementedError()

    @abc.abstractmethod
    async def add_jwt_body(username: str) -> dict:
        """Функция для добавление дополнительных данных в JWT токен пользователя"""

    @classmethod
    def mount_auth(cls, app: FastAPI):
        """Подключение аутентификации по JWT"""
        # Установить класс для аутентификации
        app.state.auth_jwt = cls
        cls.secret_key = app.state.SECRET_KEY

        @app.post("/token", summary="Аутентификация по JWT", tags=["common"])
        async def login(
            user: Annotated[OAuth2PasswordRequestForm, Depends()],
        ) -> Token:
            if await cls.check_auth(user.username, user.password):
                return Token(
                    access_token=cls._create_access_token(
                        data={
                            "sub": user.username,
                            **await cls.add_jwt_body(user.username),
                        },
                    ),
                    token_type="bearer",
                )
            else:
                raise HTTPException(status_code=401, detail="Invalid credentials")

        @app.get(
            "/check_protected",
            summary="Проверить аутентификацию по JWT",
            tags=["common"],
        )
        async def protected_route(request: Request, jwt: dict = Depends(jwt_auth)):
            return {"message": "This is a protected route", "user": jwt}

        return login, protected_route

    @classmethod
    def _create_access_token(
        cls,
        data: dict,
        expires_delta: Union[timedelta, None] = None,
    ) -> str:
        """Создание JWT токена"""
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(
                minutes=cls.ACCESS_TOKEN_EXPIRE_MINUTES
            )
        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, cls.secret_key, algorithm=cls.ALGORITHM)
        return encoded_jwt

    @classmethod
    def _verify_token(cls, token: str) -> Optional[dict]:
        """Проверка валидности JWT токена"""
        try:
            payload = jwt.decode(token, cls.secret_key, algorithms=[cls.ALGORITHM])
            return payload
        except jwt.ExpiredSignatureError:
            return None
        except jwt.InvalidTokenError:
            return None

class NoInstanceMeta(type):
    """Нельзя создавать экземпляры этого класса"""

    def __call__(cls, *args, **kwargs):
        raise TypeError("You cannot create instances of this class")

class SingletonMeta(type):
    """Мета класс для реализации паттерна Одиночка"""

    instance = None

    def __call__(cls, *args, **kwargs):
        if cls.instance is None:
            cls.instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
        return cls.instance

class AppOrm:
    """Брать методы получения сессии из настроек APP DATABASE_MANAGER"""

    @classmethod
    async def aget_orm(cls, request: Request):
        """Метод получения сессии берем из настоек APP"""
        database_manager: MainDatabaseManager = DATABASE_MANAGER() or DATABASE_MANAGER(
            request.app
        )
        async for orm in database_manager.aget_orm():
            yield orm

class BaseViewSet(abc.ABC):
    def _register_endpoint(self):
        """Регистраций API обработчиков"""

class GenericViewSet:
    """
    from app.api.v1.schemas.file import File
    from app.models.file import File as FileDb

    router = APIRouter()

    class FileViewSet(FullViewSet):
        db_model = FileDb
        pydantic_model = File

    router.views = [
        FileViewSet().as_view(router, prefix="/file"),
    ]
    """

    # Модель БД SqlAlchemy
    db_model: Type
    # Модель Схемы
    pydantic_model: Type[BaseModel]

    # Теги для URL
    tags: Optional[List[str]]

    # Пагинация для List
    paginator_class: Optional[BasePaginatorClass]

    # Класс для кешировать ответ List, Retrieve
    cache_class: Optional[BaseCache]
    # Время жизни ответа в кеше
    cache_ttl: Optional[timedelta]

    # Включить поддержку вложенных схем pydantic
    # это значит что будет происходить рекурсивное создание, обновление, удаление связанных записей
    deep_schema: Optional[bool]

    # Зависимости на весь ViewSet
    # Например можно указать аутентификацию по JWT `[Depends(jwt_auth)]`
    dependencies: Optional[list[Callable]]

    # Какой тип у первичного ключа для Api метода
    type_item_id: Optional[type]

    def as_view(self, router: APIRouter, prefix: str):
        self._mro_class = self.__class__.mro()
        # Модель БД SqlAlchemy
        self.router: APIRouter = router
        # Модель Схемы
        self.db_model: Type = self.db_model
        self.pydantic_model: Type[BaseModel] = self.pydantic_model
        # Префикс для URL
        self.prefix: str = prefix
        # Теги
        self.tags: List[str] = getattr(self, "tags", [self._mro_class[0].__name__])
        # Имя столбца с первичным ключом
        self.name_pk: str = get_pk(self.db_model)
        # Включить поддержку вложенных схем
        self.deep_schema = getattr(self, "deep_schema", False)
        # Зависимости на весь ViewSet
        self.dependencies = getattr(self, "dependencies", [])
        # Пагинация для List
        self.paginator_class = getattr(self, "paginator_class", None)
        # Какой тип у первичного ключа для Api метода
        self.type_item_id = getattr(
            self,
            "type_item_id",
            # Попытаться определить тип автоматически по типу первичного ключа в модели
            SQL_TO_PYTHON_TYPE.get(
                class_mapper(self.db_model).primary_key[0].type.__class__,
                Union[str, int, UUID],
            ),
        )

        # Логика кеширования
        self.cache = False
        if cache_class := getattr(self, "cache_class", False):
            self.cache = True
            self.cache_class = cache_class
            self.cache_ttl = getattr(self, "cache_ttl", timedelta(seconds=3))
        else:
            self.cache_class = None
            self.cache_ttl = None

        # Подключить API обработчики
        for cls in self._mro_class:
            if cls in (
                ViewSetList,
                ViewSetCreate,
                ViewSetRetrieve,
                ViewSetUpdate,
                ViewSetDelete,
            ):
                cls._register_endpoint(self)

        # Документирование тегов
        self.openapi_tag = {"name": self.tags[0], "description": str(self)}
        return self

    def __str__(self) -> str:
        """Сформировать документацию для ViewSet"""
        description = "ViewSet"
        if doc := self._mro_class[0].__doc__:
            description = "{doc}: {db_model}{dependencies}{cache}{deep_schema}".format(
                doc=doc.strip(),
                db_model=f" **db_model**={self.db_model.__name__}",
                dependencies=(
                    f" **dependencies**=[{', '.join(d.dependency.__name__ for d in self.dependencies)}]"
                    if self.dependencies
                    else ""
                ),
                cache=(" **cache**=On" if self.cache else ""),
                deep_schema=(" **deep_schema**=On" if self.deep_schema else ""),
            )
        return description

class ViewSetCreate(BaseViewSet):
    def _register_endpoint(self):
        self.create()

    def _to_db_item(self, item: Type[BaseModel]) -> T:
        """Преобразовать элемент pydantic в db_item"""
        db_item = (
            deep_instance(self.db_model, item)
            if self.deep_schema
            else self.db_model(**item.dict())
        )
        return db_item

    def create(self):
        """API обработки"""

        @self.router.post(
            f"{self.prefix}", tags=self.tags, dependencies=self.dependencies
        )
        async def create_item(
            item: self.pydantic_model, aorm: OrmAsync = Depends(AppOrm.aget_orm)
        ) -> self.pydantic_model:
            return await self.db_create(item, aorm)

        return create_item

    async def db_create(self, item: Type[BaseModel], aorm: OrmAsync) -> object:
        """Функция для создание записи в БД."""
        return await aorm.create_item(self._to_db_item(item), deep=self.deep_schema)

class ViewSetDelete(BaseViewSet):
    def _register_endpoint(self):
        self.delete()

    def delete(self):
        """API обработки"""

        @self.router.delete(
            f"{self.prefix}/{{item_id}}", tags=self.tags, dependencies=self.dependencies
        )
        async def delete_item(
            item_id: self.type_item_id, aorm: OrmAsync = Depends(AppOrm.aget_orm)
        ):
            return await self.db_delete(item_id, aorm)

        return delete_item

    async def db_delete(self, item_id: Union[str, int, UUID], aorm: OrmAsync):
        """Функция для удаления записи в БД"""
        return await aorm.delete_item(self.db_model, item_id, deep=self.deep_schema)

class ViewSetList(BaseViewSet):
    def _register_endpoint(self):
        self.list_paginator() if self.paginator_class else self.list()

    @staticmethod
    def get_offset(page: int, size: int) -> int:
        return (page - 1) * size

    def list_paginator(self):
        """Обработчик API с пагинацией"""

        @self.router.get(
            f"{self.prefix}", tags=self.tags, dependencies=self.dependencies
        )
        @cache_redis(self.cache_class, self.cache_ttl, self.cache)
        async def get_list_items_paginator(
            request: Request,
            page: int = Query(1, gt=0),
            size: int = Query(10, gt=0),
            aorm: OrmAsync = Depends(AppOrm.aget_orm),
        ) -> self.paginator_class.Schema:
            offset = self.get_offset(page, size)
            response = await aorm.get_list(
                select(self.db_model).order_by(self.name_pk).offset(offset).limit(size),
                deep=self.deep_schema,
                db_model=self.db_model,
            )
            return self.paginator_class.json(page, size, response)

        return get_list_items_paginator

    def list(self):
        """Обработчик API"""

        @self.router.get(
            f"{self.prefix}", tags=self.tags, dependencies=self.dependencies
        )
        @cache_redis(self.cache_class, self.cache_ttl, self.cache)
        async def get_list_items(
            request: Request,
            skip: int = Query(0, gte=0),
            limit: int = Query(100, gt=0),
            aorm: OrmAsync = Depends(AppOrm.aget_orm),
        ) -> List[self.pydantic_model]:
            return await aorm.get_list(
                select(self.db_model).offset(skip).limit(limit),
                deep=self.deep_schema,
                db_model=self.db_model,
            )

        return get_list_items

class ViewSetUpdate(BaseViewSet):
    def _register_endpoint(self):
        self.update()

    def update(self):
        """API обработки"""

        @self.router.put(
            f"{self.prefix}/{{item_id}}", tags=self.tags, dependencies=self.dependencies
        )
        async def update_item(
            item_id: self.type_item_id,
            item: self.pydantic_model,
            aorm: OrmAsync = Depends(AppOrm.aget_orm),
        ) -> self.pydantic_model:
            return await self.db_update(item_id, item, aorm)

        return update_item

    async def db_update(
        self, item_id: Union[str, int, UUID], item: Type[BaseModel], aorm: OrmAsync
    ) -> object:
        """Функция для обновление записи в БД"""
        return await aorm.update_item(
            self.db_model, item_id, item.dict(exclude_unset=True), deep=self.deep_schema
        )

class BaseDatabaseManager(metaclass=SingletonMeta):
    """Базовый класс для работы с РСУБД"""

    def __init__(
        self,
        database_url: str,
        *,
        pool_size=10,
        max_overflow=0,
        echo=True,
        DEV_STATUS: bool = False,
    ):
        self.database_url = None
        self.engine = None
        self.session = None
        self.Base = None
        self.adatabase_url = None
        self.aengine = None
        self.asession = None
        self._aBase = None
        self.DEV_STATUS = None

    def get_session() -> Generator[Session, None, None]:
        pass

    def get_session_transaction() -> Generator[Session, None, None]:
        pass

    async def aget_session() -> AsyncGenerator[AsyncSession, None]:
        pass

    async def aget_session_transaction() -> AsyncGenerator[AsyncSession, None]:
        pass

    def check_dev(self):
        """Проверять включенный режим разработки"""
        if not self.DEV_STATUS:
            raise ValueError("Такое действие можно выполнять только в DEV режиме.")

    def create_all(self):
        pass

    def drop_all(self):
        pass

    async def acreate_all(self):
        pass

    async def adrop_all(self):
        pass

class BaseOrm:

    def __init__(self, asession: AsyncSession):
        self.asession = asession

    async def _execute(
        self, query: Select, deep: bool = False, db_model: Type[T] | None = None
    ):
        """Выполнить Select запрос"""
        if not deep:
            response = await self.asession.execute(query)
            return response
        else:
            if not db_model:
                raise TypeError("Not set db_model")
            relationships = class_mapper(db_model).relationships
            options = [joinedload(getattr(db_model, rel.key)) for rel in relationships]
            response = await self.asession.execute(query.options(*options))
            return response

    async def get(
        self,
        query: Select,
        deep: bool = False,
        db_model: Type[T] | None = None,
    ) -> Optional[T]:
        """Получить объект по запросу"""
        response = await self._execute(query, deep, db_model)
        return response.scalar_one_or_none()

    async def get_list(
        self,
        query: Select,
        deep: bool = False,
        db_model: Type[T] | None = None,
    ) -> list[T]:
        """Получить список объектов по запросу"""
        response = await self._execute(query, deep, db_model)
        return response.scalars().all()

    async def update(self, query: Update, update_data: dict) -> T:
        """Обновить объекты по запросу"""
        query = query.values(**update_data).returning(query.table)
        result = await self.asession.execute(query)
        await self.asession.commit()
        return result.scalars().first()

    async def delete(self, query: Delete) -> bool:
        """Удалить объекты по запросу"""
        result = await self.asession.execute(query)
        await self.asession.commit()
        return result.rowcount > 0

class DatabaseAsyncSessionMixin(BaseDatabaseManager):
    """Класс для асинхронной работы с РСУБД используя пул соединений"""

    @classmethod
    async def aget_session(cls) -> AsyncGenerator[AsyncSession, None]:
        """Зависимость для получения сессии базы данных

        Получить сессию:

        async for asession in aget_orm():
            yield asession
        """
        async with cls.instance.asession() as asession:
            yield asession

    @classmethod
    async def aget_orm(cls) -> AsyncGenerator[OrmAsync, None]:
        """Зависимость для получения сессии базы данных с транзакцией

        Получить сессию:

        async for orm in aget_orm():
            yield orm
        """
        async for asession in cls.aget_session():
            yield OrmAsync(asession)

    @classmethod
    async def aget_session_transaction(cls) -> AsyncGenerator[AsyncSession, None]:
        """Зависимость для получения сессию в транзакции

        Получить сессию:

        async for asession in aget_session_transaction():
            yield asession
        """
        async with cls.instance.asession() as asession:
            async with asession.begin():
                yield asession

    @classmethod
    async def acreate_all(cls):
        cls.instance.check_dev()
        async with cls.instance.aengine.begin() as aconn:
            # await conn.run_sync(Base.metadata.drop_all)
            await aconn.run_sync(cls.instance._aBase.metadata.create_all)

    @classmethod
    async def dispose(cls):
        """Метод dispose() закрывает все неиспользуемые соединения
        в пуле соединений, связанном с данным engine.
        Это позволяет освободить ресурсы базы данных, когда они больше не нужны
        """
        await cls.instance.aengine.dispose()

class DatabaseSyncSessionMixin(BaseDatabaseManager):
    """Класс для синхронной работы с РСУБД"""

    @classmethod
    def get_session(cls) -> Generator[Session, None, None]:
        """Зависимость для получения сессии базы данных"""
        session = cls.instance.session()
        try:
            yield session
        finally:
            session.close()

    @classmethod
    def get_session_transaction(cls) -> Generator[Session, None, None]:
        """Зависимость для получения сессии базы данных с транзакцией"""
        session = cls.instance.session()
        try:
            with session.begin():
                yield session
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    @classmethod
    def get_metadata(cls) -> MetaData:
        """Получаем обновленные метаданные"""
        metadata = MetaData()
        metadata.reflect(bind=cls.instance.engine)
        return metadata

    @classmethod
    def create_all(cls):
        """Создать все таблицы в БД"""
        cls.instance.check_dev()
        cls.instance.Base.metadata.create_all(bind=cls.instance.engine)

    @classmethod
    def drop_all(cls):
        """Удалить все таблицы из БД"""
        cls.instance.check_dev()
        cls.instance.get_metadata().drop_all(bind=cls.instance.engine)

    @classmethod
    def clear_all(cls, exclude_tables_name: list[str] = None):
        """Отчистить данные во всех таблицах

        exclude_tables_name: Список имён таблиц которые не нужно отчищать
        """
        cls.instance.check_dev()
        with cls.instance.session() as session:
            with session.begin():
                for table in reversed(cls.instance.get_metadata().sorted_tables):
                    if table.name in exclude_tables_name:
                        continue
                    session.execute(table.delete())

class OrmAsync(OrmAsyncItem):
    """Взаимодействие с БД"""

    ...