Commit cf3784d9 authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

feat: changes so fm_client and nats_publisher are only created once during startup

parent 65422a07

src/srm/api/databus/dependencies.py

+2 −2
Original line number Diff line number Diff line
@@ -83,10 +83,10 @@ def build_network_capability_activate_use_case( 


def get_network_capability_activate_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

    publisher: DataBusPublisher,

) -> NetworkCapabilityActivateCommandCoordinator:

    return NetworkCapabilityActivateCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_network_capability_activate_use_case,

        publisher=NatsPublisher(connection_manager),

        publisher=publisher,

    )

src/srm/api/databus/nats_subscriber.py

+4 −7
Original line number Diff line number Diff line
@@ -7,14 +7,13 @@ from pydantic import ValidationError 
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.adapters.federation_manager.fm_client import FederationManager

from srm.api.databus.dependencies import (

    get_deploy_service_command_coordinator,

    get_network_capability_activate_command_coordinator,

)

from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

from srm.config import FederationManagerSettings

from srm.domain.ports.databus.publisher import DataBusPublisher



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -48,10 +47,9 @@ COMMAND_SUBJECTS = ( 
async def subscribe_to_subjects(

    connection_manager: NatsConnectionManager,

    session_maker: async_sessionmaker[AsyncSession],

    fm_settings: FederationManagerSettings,

    publisher: DataBusPublisher,

    fm_client: FederationManager,

) -> list["NatsSubscriber"]:

    publisher = NatsPublisher(connection_manager=connection_manager)

    fm_client = FederationManager(settings=fm_settings, publisher=publisher)



    deploy_coordinator = get_deploy_service_command_coordinator(

        session_maker,
@@ -59,8 +57,7 @@ async def subscribe_to_subjects( 
        fm_client=fm_client,

    )

    network_activate_coordinator = get_network_capability_activate_command_coordinator(

        session_maker,

        connection_manager,

        session_maker, publisher=publisher

    )



    command_handlers = {

src/srm/main.py

+9 −5
Original line number Diff line number Diff line
@@ -84,11 +84,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: 
        await engine.dispose()

        raise



    publisher = NatsPublisher(databus_manager)

    fm_client = FederationManager(

        settings=settings.federation_manager_settings, publisher=publisher

    )



    try:

        databus_subscribers: list[NatsSubscriber] = await subscribe_to_subjects(

            connection_manager=databus_manager,

            session_maker=session_maker,

            fm_settings=settings.federation_manager_settings,

            publisher=publisher,

            fm_client=fm_client,

        )

    except Exception as e:

        logger.error("Databus subscription failed!", error=str(e))
@@ -103,10 +109,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: 
    app.state.session_maker = session_maker

    app.state.databus_connection_manager = databus_manager

    app.state.databus_subscribers = databus_subscribers

    app.state.publisher = NatsPublisher(databus_manager)

    app.state.fm_client = FederationManager(

        settings=settings.federation_manager_settings, publisher=app.state.publisher

    )

    app.state.publisher = publisher

    app.state.fm_client = fm_client



    yield

tests/api/databus/test_nats_subscriber.py

+13 −7
Original line number Diff line number Diff line 
from __future__ import annotations



from dataclasses import dataclass, field

from unittest.mock import ANY, AsyncMock, MagicMock, patch

from unittest.mock import AsyncMock, MagicMock, patch



import pytest

import structlog.testing
@@ -13,7 +13,6 @@ from srm.api.databus.nats_subscriber import ( 
    subscribe_to_subjects,

)

from srm.api.databus.schemas import AnswerableCommandV1

from srm.config import FederationManagerSettings





@dataclass
@@ -262,11 +261,14 @@ async def test_deploy_subscription_callback_delegates_to_command_coordinator( 
        coordinator = MagicMock()

        coordinator.handle = AsyncMock()

        coordinator_factory.return_value = coordinator

        publisher = AsyncMock()

        fm_client = AsyncMock()



        subscribers = await subscribe_to_subjects(

            connection_manager,

            session_maker,  # type: ignore[arg-type]

            FederationManagerSettings(base_url="http://test", timeout=0.5),

            publisher,

            fm_client,

        )

        deploy_subscriber = next(

            subscriber
@@ -275,7 +277,7 @@ async def test_deploy_subscription_callback_delegates_to_command_coordinator( 
        )

        await deploy_subscriber._callback(envelope)



    coordinator_factory.assert_called_once_with(session_maker, pub=ANY, fm_client=ANY)

    coordinator_factory.assert_called_once_with(session_maker, pub=publisher, fm_client=fm_client)

    coordinator.handle.assert_awaited_once_with(envelope)

    assert session_maker.contexts == []
@@ -310,11 +312,14 @@ async def test_network_activate_subscription_callback_delegates_to_command_coord 
        coordinator = MagicMock()

        coordinator.handle = AsyncMock()

        coordinator_factory.return_value = coordinator

        publisher = AsyncMock()

        fm_client = AsyncMock()



        subscribers = await subscribe_to_subjects(

            connection_manager,

            session_maker,  # type: ignore[arg-type]

            FederationManagerSettings(base_url="http://test", timeout=0.5),

            publisher,

            fm_client,

        )

        activate_subscriber = next(

            subscriber
@@ -323,7 +328,7 @@ async def test_network_activate_subscription_callback_delegates_to_command_coord 
        )

        await activate_subscriber._callback(envelope)



    coordinator_factory.assert_called_once_with(session_maker, connection_manager)

    coordinator_factory.assert_called_once_with(session_maker, publisher=publisher)

    coordinator.handle.assert_awaited_once_with(envelope)

    assert session_maker.contexts == []
@@ -334,7 +339,8 @@ async def test_subscribe_to_subjects_registers_all_command_subjects( 
    subscribers = await subscribe_to_subjects(

        connection_manager,

        FakeSessionMaker(),  # type: ignore[arg-type]

        FederationManagerSettings(base_url="http://test", timeout=0.5),

        AsyncMock(),

        AsyncMock(),

    )



    assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)

tests/integration/test_databus.py

+5 −2
Original line number Diff line number Diff line
@@ -13,7 +13,7 @@ from srm.adapters.databus.nats_connection_manager import NatsConnectionManager 
from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects

from srm.api.databus.schemas import AnswerableCommandV1

from srm.config import FederationManagerSettings, NatsSettings

from srm.config import NatsSettings

from tests.api.databus.test_nats_subscriber import EXPECTED_COMMAND_SUBJECTS
@@ -132,10 +132,13 @@ async def test_subscribe_to_subjects_registers_all_command_subjects( 
        def __call__(self) -> FakeSessionContext:

            return FakeSessionContext()



    publisher = AsyncMock()

    fm_client = AsyncMock()

    subscribers = await subscribe_to_subjects(

        connection_manager,

        FakeSessionMaker(),  # type: ignore[arg-type]

        FederationManagerSettings(base_url="http://test", timeout=0.5),

        publisher,

        fm_client,

    )



    assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)