feat: add poller and rabbitmq queue containers

app/app/app.py → app/app.py

@@ -6,9 +6,9 @@
 
 
 class Application(AiohttpApplication):
-    database: Database | None = None
-    config: Config | None = None
-    store: Store | None = None
+    database: Database
+    config: Config
+    store: Store
 
 
 app = Application()

app/core/config.py

@@ -6,7 +6,7 @@
 from sqlalchemy.engine.url import URL
 
 if typing.TYPE_CHECKING:
-    from app.app.app import Application
+    from app.app import Application
 
 
 class BotConfig(BaseModel):
@@ -32,12 +32,27 @@ def url(self) -> URL:
         )
 
 
+class RabbitmqConfig(BaseModel):
+    host: str = "localhost"
+    port: int = 5672
+    user: str = "guest"
+    password: str = "guest"
+    input_queue: str = "input_queue"
+    output_queue: str = "output_queue"
+
+    @cached_property
+    def url(self) -> str:
+        return f"amqp://{self.user}:{self.password}@{self.host}:{self.port}"
+
+
 class Config(BaseSettings):
-    bot: BotConfig | None = None
-    database: DatabaseConfig | None = None
+    bot: BotConfig
+    database: DatabaseConfig
+    rabbitmq: RabbitmqConfig
 
     model_config = SettingsConfigDict(
-        env_file=(".env", "../../../../.env"),  # Если main / если миграции
+        # Если main / если poller / если миграции
+        env_file=(".env", "../../.env", "../../../../.env"),
         env_nested_delimiter="__",
     )
 

app/core/database/database.py

@@ -12,7 +12,7 @@
 from app.core.database.sqlalchemy_base import BaseModel
 
 if TYPE_CHECKING:
-    from app.app.app import Application
+    from app.app import Application
 
 
 class Database:

app/core/database/migrations/env.py

@@ -9,7 +9,7 @@
 from sqlalchemy.ext.asyncio import async_engine_from_config
 
 import app
-from app.app.app import app as application, setup_app
+from app.app import app as application, setup_app
 from app.core.database.sqlalchemy_base import BaseModel
 
 setup_app()

app/core/manager.py

@@ -0,0 +1,39 @@
+from collections.abc import Awaitable, Callable
+
+import aio_pika
+from aio_pika import Message
+from aio_pika.abc import AbstractIncomingMessage
+
+
+class RabbitMQManager:
+    def __init__(self, amqp_url: str, queue_name: str):
+        self._url = amqp_url
+        self._queue_name = queue_name
+        self._connection: aio_pika.abc.AbstractRobustConnection | None = None
+        self._channel: aio_pika.abc.AbstractChannel | None = None
+        self._queue: aio_pika.abc.AbstractQueue | None = None
+
+    async def connect(self) -> None:
+        self._connection = await aio_pika.connect_robust(self._url)
+        self._channel = await self._connection.channel()
+        self._queue = await self._channel.declare_queue(self._queue_name, durable=True)
+
+    async def send(self, body: bytes) -> None:
+        if self._channel is None:
+            raise RuntimeError("RabbitMQManager is not connected")
+        await self._channel.default_exchange.publish(
+            Message(body=body),
+            routing_key=self._queue_name,
+        )
+
+    async def consume(
+        self,
+        handler: Callable[[AbstractIncomingMessage], Awaitable[None]],
+    ) -> None:
+        if self._queue is None:
+            raise RuntimeError("RabbitMQManager is not connected")
+        await self._queue.consume(handler)
+
+    async def close(self) -> None:
+        if self._connection:
+            await self._connection.close()

app/core/store.py

@@ -1,7 +1,7 @@
 import typing
 
 if typing.TYPE_CHECKING:
-    from app.app.app import Application
+    from app.app import Application
 
 
 class Store:  # noqa: B903

app/poller/Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.12-alpine
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /project
+COPY pyproject.toml uv.lock ./
+
+RUN uv sync --compile-bytecode --no-cache --no-dev
+ENV PATH="/project/.venv/bin:$PATH"
+
+COPY . .
+
+WORKDIR /project
+CMD python -m app.poller.poller

app/poller/poller.py

@@ -0,0 +1,34 @@
+import asyncio
+import json
+from typing import Any, cast
+
+import aiohttp
+
+from app.app import app, setup_app
+from app.core.manager import RabbitMQManager
+
+
+async def get_updates(offset: int) -> dict[str, Any]:
+    url = f"https://api.telegram.org/bot{app.config.bot.token}/getUpdates"  # поменяю
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url, params={"offset": offset, "timeout": 30}) as resp:
+            return cast(dict[str, Any], await resp.json())
+
+
+async def poll_and_push() -> None:
+    rabbit = RabbitMQManager(
+        amqp_url=app.config.rabbitmq.url,
+        queue_name=app.config.rabbitmq.input_queue,
+    )
+    await rabbit.connect()
+
+    offset = 0
+    while True:
+        data = await get_updates(offset)
+        for update in data.get("result", []):
+            offset = update["update_id"] + 1
+            await rabbit.send(json.dumps(update).encode())
+
+
+setup_app()
+asyncio.run(poll_and_push())

docker-compose.yml

@@ -14,4 +14,35 @@ services:
       interval: 2s
       timeout: 5s
       retries: 10
+    restart: always
     command: -p ${DATABASE__PORT}
+
+  rabbitmq:
+    image: rabbitmq:4.0.8-management-alpine  # временно
+    ports:
+      - ${RABBITMQ__PORT}:${RABBITMQ__PORT}
+      - 15672:15672  # временно
+    env_file:
+      - .env
+    environment:
+      - RABBITMQ_DEFAULT_USER=${RABBITMQ__USER}
+      - RABBITMQ_DEFAULT_PASS=${RABBITMQ__PASSWORD}
+    healthcheck:
+      test: ["CMD", "rabbitmq-diagnostics", "ping"]
+      interval: 2s
+      timeout: 5s
+      retries: 10
+    restart: always
+
+  poller:
+    build:
+      context: .
+      dockerfile: app/poller/Dockerfile
+    depends_on:
+      postgres:
+        condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
+    env_file:
+      - .env
+    restart: always

main.py

@@ -1,6 +1,6 @@
 from aiohttp.web import run_app
 
-from app.app.app import setup_app
+from app.app import setup_app
 
 if __name__ == "__main__":
     run_app(setup_app())

pyproject.toml

@@ -31,7 +31,8 @@ dependencies = [
     "pycparser==2.22",
     "typing_extensions==4.13.2",
     "webargs==8.6.0",
-    "yarl==1.19.0"
+    "yarl==1.19.0",
+    "aio-pika==9.5.5",
 ]
 
 [dependency-groups]
@@ -95,6 +96,7 @@ extend-select = [
 extend-ignore = [
     "D1",
     "CPY001",
+    "SIM117",
     # По рекомендации https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules
     "W191",
     "E111",
@@ -138,4 +140,4 @@ mark-parentheses = false
 [tool.mypy]
 strict = true  # Ещё строже!
 python_version = "3.12"
-plugins = ['pydantic.mypy']
+plugins = ['pydantic.mypy']

requirements.txt

@@ -2,6 +2,7 @@ aiohttp==3.11.16
 aiohttp_apispec==3.0.0b2
 aiohttp_cors==0.8.1
 aiohttp_session==2.12.1
+aio-pika==9.5.5
 alembic==1.15.2
 asyncpg==0.30.0
 cryptography==44.0.1

template.env

@@ -0,0 +1,12 @@
+BOT__TOKEN=0123456789:a1b2c3d4e5f6g7h8i9jklmnopqrstuvwxyz
+
+DATABASE__HOST=localhost
+DATABASE__PORT=5432
+DATABASE__USER=postgres
+DATABASE__PASSWORD=postgres
+DATABASE__DATABASE=postgres
+
+RABBITMQ__HOST=localhost
+RABBITMQ__PORT=5672
+RABBITMQ__USER=guest
+RABBITMQ__PASSWORD=guest