fastapi_di.py

T = TypeVar('T')


class DependencyKey(Generic[T]):
    __slots__ = ('key',)

    def __init__(
        self, tp: type[T] | None = None, dependency_key: str | None = None
    ) -> None:
        if dependency_key:
            self.key = dependency_key
        elif tp:
            self.key = tp.__name__.lower()
        else:
            raise UndefinedDependencyError(tp=tp, dependency_key=dependency_key)


def dependency_getter(
    tp: type[T], dependency_key: str | None = None
) -> (
    Callable[[Request], Coroutine[Any, Any, T]]
    | Callable[[Context], Coroutine[Any, Any, T]]
):
    """Utility функция для удобного получения зависимости.

    Работает только для простых случаев, когда надо просто
    вытащить зависимость из стейта.
    """
    key = DependencyKey(tp=tp, dependency_key=dependency_key)

    async def getter(request: Request = TaskiqDepends()) -> T:
        return cast(
            T,
            getattr(request.state, key.key),
        )

    return taskiq_getter


def setup_dependencies(config: Config):
    """Устанавливает все зависимости уровня APP."""
    async def lifespan(*args, **kwargs):
      async with AsyncExitStack() as stack:
          engine = create_async_engine(
              config.postgres.database_uri,
              pool_size=config.postgres.pool_size,
              max_overflow=config.postgres.overflow_pool_size,
              pool_pre_ping=True,
              # echo=True,
          )

          dependencies = [
              await stack.enter_async_stack(ClientSession(...)), # Пример без DependencyKey
              (
                  DependencyKey(dependency_key='s3_client'), # Пример с DependencyKey
                  await stack.enter_async_context(
                      AioSession().create_client(
                          service_name='s3',
                          endpoint_url=config.s3.base_url,
                          aws_access_key_id=config.s3.access_key_id,
                          aws_secret_access_key=config.s3.secret_access_key,
                          region_name=config.s3.region_name,
                          use_ssl=config.s3.secure,
                          verify=config.s3.verify,
                          config=boto3.session.Config(
                              signature_version=config.s3.signature_version
                          ),
                      ),
                  ),
              ),
          ]
              engine,
          ]

          dependencies_result = {}
          for dependency in dependencies:
              match dependency:
                  case [DependencyKey(key=key), dependency]:
                      dependencies_result[key] = dependency
                  case dependency:
                      dependencies_result[DependencyKey(tp=type(dependency)).key] = (
                          dependency
                      )
          yield dependencies_result
          await engine.dispose()
          await stack.aclose()
        
def get_fastapi_app(logging_middleware: bool = True) -> FastAPI:
    """Получаем объект фастапи c прогруженными роутами.

    Returns:
        FastAPI
    """
    config = Config()
    
    app = FastAPI(
        default_response_class=ORJSONResponse,
        version='1.0.1',
        debug=config.common.environment in (EnvEnum.LOCAL, EnvEnum.DEV),
        lifespan=setup_dependencies(config),
    )
    
    app.include_router(router)
    return app