flask-demo-api/flask_demo_api/ioc.py

import os

import redis  # type: ignore
from dishka import Container, Provider, Scope, make_container, provide

from flask_demo_api.config import Settings
from flask_demo_api.protocols.repository import Repository
from flask_demo_api.repository.config import RedisSettings
from flask_demo_api.repository.redis import RedisRepository
from flask_demo_api.usecase.add import PostKey
from flask_demo_api.usecase.delete import DelKey
from flask_demo_api.usecase.get import GetKey
from flask_demo_api.usecase.put import PutKey


class RedisSettingsProvider(Provider):
    @provide(scope=Scope.APP)
    def redis_settings(self) -> RedisSettings:
        url = os.getenv("REDISURL") or "redis://localhost:6379/0"
        return RedisSettings(url=url)


class SettingsProvider(Provider):
    @provide(scope=Scope.APP)
    def main_settings(self, redis_settings: RedisSettings) -> Settings:
        return Settings(redis=redis_settings)


class RedisPoolProvider(Provider):
    @provide(scope=Scope.APP)
    def get_redis_pool(self, redis_settings: RedisSettings) -> redis.Redis:
        return redis.from_url(redis_settings.url, decode_responses=False)


class RepositoryProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def get_db(self, redis_pool: redis.Redis) -> Repository:
        return RedisRepository(redis_pool=redis_pool)


class UseCasesProvider(Provider):
    scope = Scope.REQUEST

    get = provide(GetKey)
    post = provide(PostKey)
    put = provide(PutKey)
    delete = provide(DelKey)


def create_container() -> Container:
    return make_container(
        RedisSettingsProvider(),
        SettingsProvider(),
        RedisPoolProvider(),
        RepositoryProvider(),
        UseCasesProvider(),
    )