chore: setup nats connection

.env

@@ -2,3 +2,6 @@
 POSTGRES_USER=buber_dinner
 POSTGRES_PASSWORD=buber_dinner
 POSTGRES_DB=buber_dinner
+
+NATS_USER=buber_dinner
+NATS_PASSWORD=buber_dinner

README.md

@@ -10,6 +10,10 @@
 - Pydantic
 - PyJWT
 - Structlog
+- SQLAlchemy
+- Alembic
+- orjson
+- nats-py
 
 ## Test
 

config_dist/config.prod.toml

@@ -5,8 +5,14 @@ expiry_minutes = 60
 [db]
 sa_db = "postgresql"
 sa_engine = "asyncpg"
-host = "localhost"
+host = "db"
 port = 5432
 database = "buber_dinner"
 user = "buber_dinner"
 password = "buber_dinner"
+
+[broker]
+host = "nats"
+port = 4222
+login = "buber_dinner"
+password = "buber_dinner"

config_dist/config.toml

@@ -10,3 +10,9 @@ port = 15432
 database = "buber_dinner"
 user = "buber_dinner"
 password = "buber_dinner"
+
+[broker]
+host = "nats"
+port = 14222
+login = "buber_dinner"
+password = "buber_dinner"

docker-compose.yml

@@ -19,9 +19,20 @@ services:
     depends_on:
       db:
         condition: service_healthy
+      nats:
+        condition: service_started
     environment:
       - BUBER_DINNER_CONFIG_PATH=config_dist/config.prod.toml
 
+  nats:
+    container_name: nats
+    image: nats:alpine3.19
+    command: nats-server --user ${NATS_USER:?err} --pass ${NATS_PASSWORD:?err}
+    ports:
+      - 14222:4222
+    env_file:
+      - ./.env
+
   db:
     container_name: db
     image: postgres:16

pyproject.toml

@@ -18,6 +18,7 @@ adaptix = "^3.0.0b3"
 guardpost = "^1.0.2"
 orjson = "^3.10.2"
 nats-py = "^2.7.2"
+asgiref = "^3.8.1"
 
 [tool.poetry.group.db.dependencies]
 asyncpg = "^0.29.0"

src/api/auth/handler.py

@@ -10,7 +10,8 @@ async def authenticate(self, context: Request) -> Identity | None:
         header_value = context.get_first_header(b"Authorization")
 
         if header_value:
-            context.identity = Identity({"name": "Abobe"}, "MOCK")
+            # FIXME: add real identity
+            context.identity = Identity({"name": "test"}, "MOCK")
         else:
             context.identity = None
 

src/infrastructure/config/broker.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class BrokerConfig:
+    host: str = "localhost"
+    port: int = 4222
+    login: str = "admin"
+    password: str = "admin"
+
+    @property
+    def full_url(self) -> str:
+        return f"nats://{self.login}:{self.password}@{self.host}:{self.port}"

src/infrastructure/config/config.py

@@ -1,6 +1,6 @@
-
 from pydantic import BaseModel
 
+from src.infrastructure.config.broker import BrokerConfig
 from src.infrastructure.config.db import DBConfig
 from src.infrastructure.config.jwt import JWTConfig
 from src.infrastructure.config_loader import read_toml
@@ -9,11 +9,13 @@
 class Config(BaseModel):
     jwt_config: JWTConfig
     db_config: DBConfig
+    broker_config: BrokerConfig
 
 
 def create_config_obj() -> Config:
     config_data = read_toml()
     return Config(
         jwt_config=config_data["jwt"],
         db_config=config_data["db"],
+        broker_config=config_data["broker"],
     )

src/infrastructure/config/jwt.py

@@ -1,6 +1,7 @@
-from pydantic import BaseModel
+from dataclasses import dataclass
 
 
-class JWTConfig(BaseModel):
+@dataclass
+class JWTConfig:
     jwt_secret: str
     expiry_minutes: int

src/infrastructure/di/config.py

@@ -1,11 +1,14 @@
 from rodi import Container
 
-from src.infrastructure.config.config import create_config_obj
+from src.infrastructure.config.config import Config, create_config_obj
 
 
-def setup_config_di(di_container: Container) -> None:
+def setup_config_di(di_container: Container) -> Config:
     config = create_config_obj()
 
     di_container.add_instance(config)
     di_container.add_instance(config.jwt_config)
     di_container.add_instance(config.db_config)
+    di_container.add_instance(config.broker_config)
+
+    return config

src/infrastructure/di/main.py

@@ -9,6 +9,7 @@
 from .extra import setup_extra_di
 from .mapper import setup_mapper_di
 from .mediatr import setup_mediatr_di
+from .message_queue import setup_message_queue_di
 from .persistence import setup_persistence_di
 
 
@@ -19,10 +20,11 @@ def build_application_container() -> Container:
     container.add_instance(logging.getLogger(__name__), logging.Logger)
     container.add_instance(setup_retort())
 
-    setup_mapper_di(container)
+    setup_config_di(container)
     setup_extra_di(container)
+    setup_mapper_di(container)
     setup_mediatr_di(container)
+    setup_message_queue_di(container)
     setup_persistence_di(container)
-    setup_config_di(container)
 
     return container

src/infrastructure/di/mediatr.py

@@ -15,6 +15,7 @@
     LoginQueryValidationBehavior,
     RegisterCommandValidationBehavior,
 )
+from src.application.dinners.events.menu_create_handler import MenuCreateHandler
 from src.application.menu.commands.create_menu.handler import CreateMenuCommandHandler
 from src.infrastructure.mediator.main import setup_mediatr
 
@@ -40,3 +41,4 @@ def setup_auth_mediatr(container: Container) -> None:
 
 def setup_menu_mediatr(container: Container) -> None:
     container.register(CreateMenuCommandHandler)
+    container.register(MenuCreateHandler)

src/infrastructure/di/broker.py → src/infrastructure/di/message_queue.py

@@ -1,13 +1,17 @@
+from nats import NATS
 from rodi import Container
 
 from src.application.common.events.event_bus import EventBus
 from src.infrastructure.event_bus.event_bus import EventBusImpl
 from src.infrastructure.message_broker.interface import MessageBroker
+from src.infrastructure.message_broker.main import make_broker_connection
 from src.infrastructure.message_broker.message_broker import MessageBrokerImpl
 
 
-def setup_message_queue(container: Container) -> None:
+def setup_message_queue_di(container: Container) -> None:
     setup_events_di(container)
+
+    container.add_singleton_by_factory(make_broker_connection, NATS)
     container.add_transient(MessageBroker, MessageBrokerImpl)
 
 

src/infrastructure/message_broker/main.py

@@ -0,0 +1,11 @@
+from typing import AsyncGenerator
+
+import nats
+
+from src.infrastructure.config.broker import BrokerConfig
+
+
+async def make_broker_connection(config: BrokerConfig) -> AsyncGenerator[nats.NATS, None]:
+    conn = await nats.connect(config.full_url)
+    yield conn
+    await conn.close()

tests/integration/menu/test_repository.py

@@ -12,7 +12,7 @@ async def test_get_all_menu(menu_repo: MenuRepository) -> None:
     assert resp == []
 
 
-@pytest.mark.skip("Add message broker config and connection")
+@pytest.mark.skip("Error with IoC container. https://github.com/edpyt/buber_dinner/issues/14")
 async def test_add_menu(menu_repo: MenuRepository) -> None:
     menu = Menu.create(
         name="test",