refactor

Author: Andrei Neagu

packages/service-library/src/servicelib/fastapi/docker.py

@@ -2,7 +2,7 @@
 import logging
 from collections.abc import AsyncIterator
 from contextlib import AsyncExitStack
-from typing import Final
+from typing import Any, Final
 
 import aiodocker
 import aiohttp
@@ -58,6 +58,52 @@ async def _(app: FastAPI) -> AsyncIterator[State]:
     return _
 
 
+_DOCKER_API_PROXY_SETTINGS: Final[str] = "docker_api_proxy_settings"
+
+
+def get_remote_docker_client_main_lifespan(
+    settings: DockerApiProxysettings,
+) -> dict[str, Any]:
+    return {_DOCKER_API_PROXY_SETTINGS: settings}
+
+
+async def lifespan_remote_docker_client(
+    app: FastAPI, state: State
+) -> AsyncIterator[State]:
+    settings: DockerApiProxysettings = state[_DOCKER_API_PROXY_SETTINGS]
+
+    session: ClientSession | None = None
+    if settings.DOCKER_API_PROXY_USER and settings.DOCKER_API_PROXY_PASSWORD:
+        session = ClientSession(
+            auth=aiohttp.BasicAuth(
+                login=settings.DOCKER_API_PROXY_USER,
+                password=settings.DOCKER_API_PROXY_PASSWORD.get_secret_value(),
+            )
+        )
+
+    async with AsyncExitStack() as exit_stack:
+        if settings.DOCKER_API_PROXY_USER and settings.DOCKER_API_PROXY_PASSWORD:
+            await exit_stack.enter_async_context(
+                ClientSession(
+                    auth=aiohttp.BasicAuth(
+                        login=settings.DOCKER_API_PROXY_USER,
+                        password=settings.DOCKER_API_PROXY_PASSWORD.get_secret_value(),
+                    )
+                )
+            )
+
+        client = await exit_stack.enter_async_context(
+            aiodocker.Docker(url=settings.base_url, session=session)
+        )
+
+        app.state.remote_docker_client = client
+
+        await wait_till_docker_api_proxy_is_responsive(app)
+
+        # NOTE this has to be inside exit_stack scope
+        yield {}
+
+
 @tenacity.retry(
     wait=tenacity.wait_fixed(5),
     stop=tenacity.stop_after_delay(60),

packages/service-library/src/servicelib/fastapi/postgres_lifespan.py

@@ -23,7 +23,7 @@ class PostgresConfigurationError(LifespanOnStartupError):
     msg_template = "Invalid postgres settings [={settings}] on startup. Note that postgres cannot be disabled using settings"
 
 
-def create_input_state(settings: PostgresSettings) -> State:
+def get_postgres_database_main_lifespan(settings: PostgresSettings) -> State:
     return {PostgresLifespanState.POSTGRES_SETTINGS: settings}
 
 

packages/service-library/src/servicelib/fastapi/prometheus_instrumentation.py

@@ -1,7 +1,7 @@
 # pylint: disable=protected-access
 
-
 from collections.abc import AsyncIterator
+from typing import Any, Final
 
 from fastapi import FastAPI
 from fastapi_lifespan_manager import State
@@ -54,9 +54,25 @@ def _on_shutdown() -> None:
     return get_prometheus_instrumentator(app)
 
 
-async def lifespan_prometheus_instrumentation(app: FastAPI) -> AsyncIterator[State]:
+_PROMETHEUS_INSTRUMENTATION_ENABLED: Final[str] = "prometheus_instrumentation_enabled"
+
+
+def get_prometheus_instrumentationmain_main_lifespan(
+    *, enabled: bool
+) -> dict[str, Any]:
+    return {_PROMETHEUS_INSTRUMENTATION_ENABLED: enabled}
+
+
+async def lifespan_prometheus_instrumentation(
+    app: FastAPI, state: State
+) -> AsyncIterator[State]:
     # NOTE: requires ``initialize_prometheus_instrumentation`` to be called before the
     # lifespan of the applicaiton runs, usually rigth after the ``FastAPI`` instance is created
-    _startup(app)
+
+    instrumentaiton_enabled = state.get(_PROMETHEUS_INSTRUMENTATION_ENABLED, False)
+    if instrumentaiton_enabled:
+
+        _startup(app)
     yield {}
-    _shutdown(app)
+    if instrumentaiton_enabled:
+        _shutdown(app)

services/catalog/src/simcore_service_catalog/core/events.py

@@ -4,10 +4,11 @@
 from fastapi import FastAPI
 from fastapi_lifespan_manager import LifespanManager, State
 from servicelib.fastapi.postgres_lifespan import (
-    PostgresLifespanState,
+    get_postgres_database_main_lifespan,
     postgres_database_lifespan,
 )
 from servicelib.fastapi.prometheus_instrumentation import (
+    get_prometheus_instrumentationmain_main_lifespan,
     lifespan_prometheus_instrumentation,
 )
 
@@ -43,20 +44,14 @@ async def _main_lifespan(app: FastAPI) -> AsyncIterator[State]:
     settings: ApplicationSettings = app.state.settings
 
     yield {
-        PostgresLifespanState.POSTGRES_SETTINGS: settings.CATALOG_POSTGRES,
-        "prometheus_instrumentation_enabled": settings.CATALOG_PROMETHEUS_INSTRUMENTATION_ENABLED,
+        **get_postgres_database_main_lifespan(settings.CATALOG_POSTGRES),
+        **get_prometheus_instrumentationmain_main_lifespan(
+            enabled=settings.CATALOG_PROMETHEUS_INSTRUMENTATION_ENABLED
+        ),
     }
 
 
-async def _prometheus_instrumentation_lifespan(
-    app: FastAPI, state: State
-) -> AsyncIterator[State]:
-    if state.get("prometheus_instrumentation_enabled", False):
-        async for prometheus_state in lifespan_prometheus_instrumentation(app):
-            yield prometheus_state
-
-
-def create_app_lifespan():
+def create_app_lifespan() -> LifespanManager:
     # WARNING: order matters
     app_lifespan = LifespanManager()
     app_lifespan.add(_main_lifespan)
@@ -81,7 +76,7 @@ def create_app_lifespan():
     app_lifespan.add(background_task_lifespan)
 
     # - prometheus instrumentation
-    app_lifespan.add(_prometheus_instrumentation_lifespan)
+    app_lifespan.add(lifespan_prometheus_instrumentation)
 
     app_lifespan.add(_banners_lifespan)
 

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/application.py

@@ -1,70 +1,21 @@
-from collections.abc import AsyncIterator
-
 from fastapi import FastAPI
-from fastapi_lifespan_manager import State
-from servicelib.fastapi.docker import get_lifespan_remote_docker_client
-from servicelib.fastapi.lifespan_utils import LifespanGenerator, combine_lifespans
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
 from servicelib.fastapi.profiler import initialize_profiler
 from servicelib.fastapi.prometheus_instrumentation import (
     initialize_prometheus_instrumentation,
-    lifespan_prometheus_instrumentation,
 )
 from servicelib.fastapi.tracing import initialize_tracing
 
-from .._meta import (
-    API_VERSION,
-    API_VTAG,
-    APP_FINISHED_BANNER_MSG,
-    APP_NAME,
-    APP_STARTED_BANNER_MSG,
-    PROJECT_NAME,
-    SUMMARY,
-)
+from .._meta import API_VERSION, API_VTAG, APP_NAME, PROJECT_NAME, SUMMARY
 from ..api.frontend import initialize_frontend
 from ..api.rest.routes import initialize_rest_api
-from ..api.rpc.routes import lifespan_rpc_api_routes
-from ..services.catalog import lifespan_catalog
-from ..services.deferred_manager import lifespan_deferred_manager
-from ..services.director_v0 import lifespan_director_v0
-from ..services.director_v2 import lifespan_director_v2
-from ..services.notifier import get_lifespans_notifier
-from ..services.rabbitmq import lifespan_rabbitmq
-from ..services.redis import lifespan_redis
-from ..services.service_tracker import lifespan_service_tracker
-from ..services.status_monitor import lifespan_status_monitor
+from . import events
 from .settings import ApplicationSettings
 
 
-async def _lifespan_banner(app: FastAPI) -> AsyncIterator[State]:
-    _ = app
-    print(APP_STARTED_BANNER_MSG, flush=True)  # noqa: T201
-    yield {}
-    print(APP_FINISHED_BANNER_MSG, flush=True)  # noqa: T201
-
-
 def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
     app_settings = settings or ApplicationSettings.create_from_envs()
 
-    lifespans: list[LifespanGenerator] = [
-        lifespan_director_v2,
-        lifespan_director_v0,
-        lifespan_catalog,
-        lifespan_rabbitmq,
-        lifespan_rpc_api_routes,
-        lifespan_redis,
-        *get_lifespans_notifier(),
-        lifespan_service_tracker,
-        lifespan_deferred_manager,
-        lifespan_status_monitor,
-        get_lifespan_remote_docker_client(
-            app_settings.DYNAMIC_SCHEDULER_DOCKER_API_PROXY
-        ),
-    ]
-
-    if app_settings.DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED:
-        lifespans.append(lifespan_prometheus_instrumentation)
-
     app = FastAPI(
         title=f"{PROJECT_NAME} web API",
         description=SUMMARY,
@@ -74,7 +25,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
             "/doc" if app_settings.DYNAMIC_SCHEDULER_SWAGGER_API_DOC_ENABLED else None
         ),
         redoc_url=None,
-        lifespan=combine_lifespans(*lifespans, _lifespan_banner),
+        lifespan=events.create_app_lifespan(),
     )
     override_fastapi_openapi_method(app)
 
@@ -84,7 +35,8 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
 
     initialize_rest_api(app)
 
-    initialize_prometheus_instrumentation(app)
+    if app_settings.DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED:
+        initialize_prometheus_instrumentation(app)
 
     initialize_frontend(app)
 

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/events.py

@@ -0,0 +1,72 @@
+from collections.abc import AsyncIterator
+
+from fastapi import FastAPI
+from fastapi_lifespan_manager import LifespanManager, State
+from servicelib.fastapi.docker import (
+    get_remote_docker_client_main_lifespan,
+    lifespan_remote_docker_client,
+)
+from servicelib.fastapi.prometheus_instrumentation import (
+    get_prometheus_instrumentationmain_main_lifespan,
+    lifespan_prometheus_instrumentation,
+)
+
+from .._meta import APP_FINISHED_BANNER_MSG, APP_STARTED_BANNER_MSG
+from ..api.rpc.routes import lifespan_rpc_api_routes
+from ..services.catalog import lifespan_catalog
+from ..services.deferred_manager import lifespan_deferred_manager
+from ..services.director_v0 import lifespan_director_v0
+from ..services.director_v2 import lifespan_director_v2
+from ..services.notifier import get_lifespans_notifier
+from ..services.rabbitmq import lifespan_rabbitmq
+from ..services.redis import lifespan_redis
+from ..services.service_tracker import lifespan_service_tracker
+from ..services.status_monitor import lifespan_status_monitor
+from .settings import ApplicationSettings
+
+
+async def _banner_lifespan(app: FastAPI) -> AsyncIterator[State]:
+    _ = app
+    print(APP_STARTED_BANNER_MSG, flush=True)  # noqa: T201
+    yield {}
+    print(APP_FINISHED_BANNER_MSG, flush=True)  # noqa: T201
+
+
+async def _main_lifespan(app: FastAPI) -> AsyncIterator[State]:
+    settings: ApplicationSettings = app.state.settings
+
+    yield {
+        **get_prometheus_instrumentationmain_main_lifespan(
+            enabled=settings.DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED
+        ),
+        **get_remote_docker_client_main_lifespan(
+            settings.DYNAMIC_SCHEDULER_DOCKER_API_PROXY
+        ),
+    }
+
+
+def create_app_lifespan() -> LifespanManager:
+    app_lifespan = LifespanManager()
+    app_lifespan.add(_main_lifespan)
+
+    app_lifespan.add(lifespan_director_v2)
+    app_lifespan.add(lifespan_director_v0)
+    app_lifespan.add(lifespan_catalog)
+    app_lifespan.add(lifespan_rabbitmq)
+    app_lifespan.add(lifespan_rpc_api_routes)
+    app_lifespan.add(lifespan_redis)
+
+    for lifespan in get_lifespans_notifier():
+        app_lifespan.add(lifespan)
+
+    app_lifespan.add(lifespan_service_tracker)
+    app_lifespan.add(lifespan_deferred_manager)
+    app_lifespan.add(lifespan_status_monitor)
+
+    app_lifespan.add(lifespan_remote_docker_client)
+
+    app_lifespan.add(lifespan_prometheus_instrumentation)
+
+    app_lifespan.add(_banner_lifespan)
+
+    return app_lifespan

services/dynamic-scheduler/tests/conftest.py

@@ -83,7 +83,7 @@ def app_environment(
     )
 
 
-_PATH_APPLICATION: Final[str] = "simcore_service_dynamic_scheduler.core.application"
+_PATH_APPLICATION: Final[str] = "simcore_service_dynamic_scheduler.core.events"
 
 
 @pytest.fixture