ref: make processors from tasks (#62)

Author: emptybutton

src/ttt/presentation/tasks/__init__.py renamed to src/ttt/infrastructure/processors/__init__.py

@@ -0,0 +1,29 @@
+from asyncio import sleep
+from dataclasses import dataclass
+
+from structlog.types import FilteringBoundLogger
+
+from ttt.application.invitation_to_game.game.auto_cancel_invitations_to_game import (  # noqa: E501
+    AutoCancelInvitationsToGame,
+)
+from ttt.infrastructure.dishka.next_container import NextContainer
+from ttt.infrastructure.processors.processor import Processor
+from ttt.infrastructure.retrier import Retrier
+from ttt.infrastructure.structlog.logger import unexpected_error_logging
+
+
+@dataclass(frozen=True)
+class AutoCancelInvitationsToGameProcessor(Processor):
+    _interval_seconds: float
+    _logger: FilteringBoundLogger
+
+    async def __call__(self, container: NextContainer) -> None:
+        while True:
+            await sleep(self._interval_seconds)
+            async with unexpected_error_logging(self._logger):  # noqa: SIM117
+                async with container() as request:
+                    retrier = await request.get(Retrier)
+                    cancel_invitations = await request.get(
+                        AutoCancelInvitationsToGame,
+                    )
+                    await retrier(cancel_invitations)

src/ttt/infrastructure/processors/auto_cancel_invitations_to_game_processor.py

@@ -5,13 +5,13 @@
 
 from ttt.application.user.game.matchmake import Matchmake
 from ttt.infrastructure.dishka.next_container import NextContainer
+from ttt.infrastructure.processors.processor import Processor
 from ttt.infrastructure.retrier import Retrier
 from ttt.infrastructure.structlog.logger import unexpected_error_log
-from ttt.presentation.tasks.task import Task
 
 
 @dataclass
-class MatchmakeTasks(Task):
+class MatchmakeProcessor(Processor):
     _max_workers: int
     _worker_creation_interval_seconds: float
     _logger: FilteringBoundLogger

src/ttt/presentation/tasks/matchmake_tasks.py renamed to src/ttt/infrastructure/processors/matchmake_processor.py

@@ -3,5 +3,5 @@
 from ttt.infrastructure.dishka.next_container import NextContainer
 
 
-class Task(Protocol):
+class Processor(Protocol):
     async def __call__(self, container: NextContainer, /) -> Any: ...  # noqa: ANN401

src/ttt/presentation/tasks/task.py renamed to src/ttt/infrastructure/processors/processor.py

@@ -0,0 +1,14 @@
+from asyncio import gather
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from ttt.infrastructure.dishka.next_container import NextContainer
+from ttt.infrastructure.processors.processor import Processor
+
+
+@dataclass(frozen=True, unsafe_hash=False)
+class Processors(Processor):
+    _processors: Sequence[Processor]
+
+    async def __call__(self, container: NextContainer) -> None:
+        await gather(*(processor(container) for processor in self._processors))

src/ttt/infrastructure/processors/processors.py

@@ -1,4 +1,6 @@
 import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 from typing import cast
 
 import structlog
@@ -49,3 +51,13 @@ async def unexpected_error_log(
     error: Exception,
 ) -> None:
     await logger.aexception("unexpected_error", exc_info=error)
+
+
+@asynccontextmanager
+async def unexpected_error_logging(
+    logger: FilteringBoundLogger,
+) -> AsyncIterator[None]:
+    try:
+        yield
+    except Exception as error:  # noqa: BLE001
+        await unexpected_error_log(logger, error)

src/ttt/infrastructure/structlog/logger.py

@@ -1,7 +1,8 @@
 from asyncio import Queue
 from collections.abc import AsyncIterator
+from typing import Annotated
 
-from dishka import Provider, Scope, provide
+from dishka import FromComponent, Provider, Scope, provide
 from nats import connect as connect_to_nats
 from nats.aio.client import Client as Nats
 from nats.js import JetStreamContext
@@ -11,6 +12,7 @@
     AsyncSession,
     create_async_engine,
 )
+from structlog.types import FilteringBoundLogger
 from taskiq.receiver import Receiver
 
 from ttt.application.common.errors.serialization_error import SerializationError
@@ -105,6 +107,11 @@
 from ttt.infrastructure.adapters.users import InPostgresUsers
 from ttt.infrastructure.adapters.uuids import UUIDv4s
 from ttt.infrastructure.openai.gemini import Gemini, gemini
+from ttt.infrastructure.processors.auto_cancel_invitations_to_game_processor import (  # noqa: E501
+    AutoCancelInvitationsToGameProcessor,
+)
+from ttt.infrastructure.processors.matchmake_processor import MatchmakeProcessor
+from ttt.infrastructure.processors.processors import Processors
 from ttt.infrastructure.pydantic_settings.envs import Envs
 from ttt.infrastructure.pydantic_settings.secrets import Secrets
 from ttt.infrastructure.retrier import Retrier
@@ -358,3 +365,43 @@ def provide_retrier(self, envs: Envs) -> Retrier:
         })
 
     provide_retry = provide(RetrierRetry, provides=Retry, scope=Scope.REQUEST)
+
+    @provide(scope=Scope.APP)
+    def provide_auto_cancel_invitations_to_game_task(
+        self,
+        envs: Envs,
+        logger: Annotated[FilteringBoundLogger, FromComponent("app")],
+    ) -> AutoCancelInvitationsToGameProcessor:
+        return AutoCancelInvitationsToGameProcessor(
+            _interval_seconds=(
+                envs.auto_cancel_invitations_to_game_interval_seconds
+            ),
+            _logger=logger,
+        )
+
+    @provide(scope=Scope.APP)
+    def provide_matchmake_processor(
+        self,
+        envs: Envs,
+        logger: Annotated[FilteringBoundLogger, FromComponent("app")],
+    ) -> MatchmakeProcessor:
+        return MatchmakeProcessor(
+            _max_workers=envs.matchmaking_max_workers,
+            _worker_creation_interval_seconds=(
+                envs.matchmaking_worker_creation_interval_seconds
+            ),
+            _logger=logger,
+        )
+
+    @provide(scope=Scope.APP)
+    async def processors(
+        self,
+        auto_cancel_invitations_to_game_processor: (
+            AutoCancelInvitationsToGameProcessor
+        ),
+        matchmake_processor: MatchmakeProcessor,
+    ) -> Processors:
+        return Processors((
+            auto_cancel_invitations_to_game_processor,
+            matchmake_processor,
+        ))

src/ttt/main/common/di.py

@@ -1,6 +1,6 @@
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
-from typing import Annotated, cast
+from typing import cast
 
 from aiogram import Bot, Dispatcher
 from aiogram.fsm.storage.base import BaseStorage, DefaultKeyBuilder
@@ -15,15 +15,13 @@
 from aiogram_dialog.manager.bg_manager import BgManagerFactoryImpl
 from aiogram_dialog.manager.manager import ManagerImpl
 from dishka import (
-    FromComponent,
     Provider,
     Scope,
     from_context,
     provide,
 )
 from dishka.integrations.aiogram import AiogramMiddlewareData
 from redis.asyncio import Redis
-from structlog.types import FilteringBoundLogger
 
 from ttt.application.common.ports.emojis import Emojis
 from ttt.application.game.game.cancel_game import CancelGame
@@ -120,7 +118,6 @@
 from ttt.application.user.view_other_user import ViewOtherUser
 from ttt.application.user.view_user import ViewUser
 from ttt.application.user.view_user_emojis import ViewUserEmojis
-from ttt.infrastructure.pydantic_settings.envs import Envs
 from ttt.infrastructure.pydantic_settings.secrets import Secrets
 from ttt.presentation.adapters.emojis import PictographsAsEmojis
 from ttt.presentation.adapters.game_views import (
@@ -151,12 +148,6 @@
 )
 from ttt.presentation.aiogram_dialog.main_dialog import main_dialog
 from ttt.presentation.result_buffer import ResultBuffer
-from ttt.presentation.tasks.auto_cancel_invitations_to_game_task import (
-    AutoCancelInvitationsToGameTask,
-)
-from ttt.presentation.tasks.matchmake_tasks import MatchmakeTasks
-from ttt.presentation.tasks.unkillable_tasks import UnkillableTasks
-from ttt.presentation.unkillable_task_group import UnkillableTaskGroup
 
 
 @dataclass
@@ -253,50 +244,6 @@ async def provide_bot(self, secrets: Secrets) -> AsyncIterator[Bot]:
     def provide_result_buffer(self) -> ResultBuffer:
         return ResultBuffer()
 
-    @provide(scope=Scope.APP)
-    def provide_auto_cancel_invitations_to_game_task(
-        self, envs: Envs,
-    ) -> AutoCancelInvitationsToGameTask:
-        return AutoCancelInvitationsToGameTask(
-            _interval_seconds=(
-                envs.auto_cancel_invitations_to_game_interval_seconds
-            ),
-        )
-
-    @provide(scope=Scope.APP)
-    def provide_matchmake_tasks(
-        self,
-        envs: Envs,
-        logger: Annotated[FilteringBoundLogger, FromComponent("app")],
-    ) -> MatchmakeTasks:
-        return MatchmakeTasks(
-            _max_workers=envs.matchmaking_max_workers,
-            _worker_creation_interval_seconds=(
-                envs.matchmaking_worker_creation_interval_seconds
-            ),
-            _logger=logger,
-        )
-
-    @provide(scope=Scope.APP)
-    async def unkillable_task_group(
-        self, logger: Annotated[FilteringBoundLogger, FromComponent("app")],
-    ) -> AsyncIterator[UnkillableTaskGroup]:
-        async with UnkillableTaskGroup(logger) as group:
-            yield group
-
-    @provide(scope=Scope.APP)
-    async def unkillable_tasks(
-        self,
-        task_group: UnkillableTaskGroup,
-        auto_cancel_invitations_to_game_task: AutoCancelInvitationsToGameTask,
-        matchmake_tasks: MatchmakeTasks,
-    ) -> UnkillableTasks:
-        tasks = (
-            auto_cancel_invitations_to_game_task,
-            matchmake_tasks,
-        )
-        return UnkillableTasks(tasks, task_group)
-
     @provide(scope=Scope.APP)
     def provide_dp(self, storage: BaseStorage) -> Dispatcher:
         dp = Dispatcher(name="main", storage=storage)

src/ttt/main/tg_bot/di.py

@@ -1,21 +1,21 @@
 import logging
+from asyncio import CancelledError, TaskGroup
+from contextlib import suppress
 
 from aiogram import Bot, Dispatcher
 from aiogram.types import TelegramObject
 from dishka import AsyncContainer
-from dishka.integrations.aiogram import (
-    AiogramMiddlewareData,
-)
+from dishka.integrations.aiogram import AiogramMiddlewareData
 from taskiq import TaskiqMessage
 
+from ttt.infrastructure.processors.processors import Processors
 from ttt.infrastructure.taskiq.broker import NatsBroker
 from ttt.infrastructure.taskiq.middlewares import TaskiqNextContainerMiddleware
 from ttt.infrastructure.taskiq.worker import TaskiqBgWorker
 from ttt.main.common.next_container import NextContainerWithFilledContext
 from ttt.presentation.aiogram.common.middlewares import (
     AiogramNextContainerMiddleware,
 )
-from ttt.presentation.tasks.unkillable_tasks import UnkillableTasks
 
 
 async def start_tg_bot(container: AsyncContainer) -> None:
@@ -33,14 +33,17 @@ async def start_tg_bot(container: AsyncContainer) -> None:
     nats_broker.add_middlewares(TaskiqNextContainerMiddleware(next_container))
 
     taskiq_bg_worker = await container.get(TaskiqBgWorker)
-    tasks = await container.get(UnkillableTasks)
+    processors = await container.get(Processors)
     bot = await container.get(Bot)
 
     logging.basicConfig(level=logging.INFO)
 
     try:
-        await tasks(next_container)
-        await taskiq_bg_worker()
-        await dp.start_polling(bot)
+        with suppress(CancelledError):
+            async with TaskGroup() as tasks:
+                tasks.create_task(processors(next_container))
+                await taskiq_bg_worker()
+                await dp.start_polling(bot)
+                raise CancelledError
     finally:
         await container.close()