Ограничитель подключечний на базе Redis

1. Настройка аргументов при инициализации библиотеки в сервисах, использующих Initer

Аргументы по умолчанию:

url: str = 'redis://localhost:6379/0',

key_time_to_live_s: int = 120,

sleep_time_s: int = 3

lock_tools/redis_connector.py

class RedisConnector(AsyncInitable, AsyncDeinitable):
    @dataclass
    class Config:
        url: str = 'redis://localhost:6379/0'

lock_tools/redis_semaphore.py

class RedisSemaphore(AbstractSemaphore):
    @dataclass
    class Context:
        redis: RedisConnector

    @dataclass
    class Config:
        key_time_to_live_s: int = 120
        sleep_time_s: int = 3

При необходимости, указанные аргументы, можно переопределить через передачу в config.ini

[redis]
url = redis://localhost:9999/1

[semaphore]
key_time_to_live_s = 60
sleep_time_s = 5

2.1 Инициализация библиотеки в сервисах, использующих Initer

from dataclasses import dataclass

import init_helpers
from aiohttp import ClientSession
from async_tools import AsyncDeinitable, AsyncInitable
from http_tools import HttpServer
from lock_tools import RedisConnector, RedisSemaphore

@dataclass
class Initer:
    @dataclass
    class Config:
        logging: init_helpers.LogsConfig
        http_server: HttpServer.Config
        redis: RedisConnector.Config
        semaphore: RedisSemaphore.Config

    config: Config

    @dataclass
    class Context(AsyncInitable, AsyncDeinitable):
        instance_id: str = 'Example'
        session: ClientSession = None
        http_server: HttpServer = None
        redis: RedisConnector = None
        semaphore: RedisSemaphore = None
        
        def __post_init__(self) -> None:
            AsyncInitable.__init__(self)
            AsyncDeinitable.__init__(self)

    context: Context

    def __init__(self) -> None:
        self.config = init_helpers.parse_args(config_file=init_helpers.Arg.ini_file_to_dataclass(self.Config))
        self.context = self.Context()

    async def __aenter__(self) -> None:
        self.context.http_server = HttpServer(self.config.http_server, self.context)
        self.context.redis = RedisConnector(self.config.redis)
        self.context.semaphore = RedisSemaphore(self.config.semaphore, self.context)
        await self.context.async_init()

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if self.context.session is not None:
            await self.context.session.close()
        await self.context.async_deinit()

2.2 Инициализация в контроллере.

from dataclasses import dataclass
from aiohttp import ClientSession

from yarl import URL
from lock_tools import RedisSemaphore


class Controller:
    @dataclass
    class Context:
        session: ClientSession
        semaphore: RedisSemaphore

    @dataclass
    class Config:
        max_connections_to_host: int = 10

    def __init__(self, config: Config, context: Context):
        self._context = context
        self._config = config

    async def get_data(self, url: str):
        async with self._context.semaphore.restrict(URL(url).host, self._config.max_connections_to_host):
            request = self._context.session.get(url)
            return request