✨ Adding catalog client to dynamic-scheduler ⚠️ (#7162)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

services/director-v2/src/simcore_service_director_v2/modules/catalog.py

@@ -83,7 +83,7 @@ async def get_service(
     ) -> dict[str, Any]:
         resp = await self.request(
             "GET",
-            f"/services/{urllib.parse.quote( service_key, safe='')}/{service_version}",
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}",
             params={"user_id": user_id},
             headers={"X-Simcore-Products-Name": product_name},
         )
@@ -98,7 +98,7 @@ async def get_service_resources(
     ) -> ServiceResourcesDict:
         resp = await self.request(
             "GET",
-            f"/services/{urllib.parse.quote( service_key, safe='')}/{service_version}/resources",
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}/resources",
             params={"user_id": user_id},
         )
         resp.raise_for_status()
@@ -114,7 +114,7 @@ async def get_service_labels(
     ) -> SimcoreServiceLabels:
         resp = await self.request(
             "GET",
-            f"/services/{urllib.parse.quote( service_key, safe='')}/{service_version}/labels",
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}/labels",
         )
         resp.raise_for_status()
         if resp.status_code == status.HTTP_200_OK:
@@ -137,7 +137,7 @@ async def get_service_specifications(
     ) -> dict[str, Any]:
         resp = await self.request(
             "GET",
-            f"/services/{urllib.parse.quote( service_key, safe='')}/{service_version}/specifications",
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}/specifications",
             params={"user_id": user_id},
         )
         resp.raise_for_status()

services/docker-compose.yml

@@ -563,6 +563,8 @@ services:
       REDIS_SECURE: ${REDIS_SECURE}
       REDIS_USER: ${REDIS_USER}
       REDIS_PASSWORD: ${REDIS_PASSWORD}
+      CATALOG_HOST: ${CATALOG_HOST}
+      CATALOG_PORT: ${CATALOG_PORT}
       DIRECTOR_V2_HOST: ${DIRECTOR_V2_HOST}
       DIRECTOR_V2_PORT: ${DIRECTOR_V2_PORT}
       DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER: ${DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER}

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

@@ -24,6 +24,7 @@
 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
@@ -48,6 +49,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
     lifespans: list[LifespanGenerator] = [
         lifespan_director_v2,
         lifespan_director_v0,
+        lifespan_catalog,
         lifespan_rabbitmq,
         lifespan_rpc_api_routes,
         lifespan_redis,

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

@@ -5,6 +5,7 @@
 from servicelib.logging_utils_filtering import LoggerName, MessageSubstring
 from settings_library.application import BaseApplicationSettings
 from settings_library.basic_types import LogLevel, VersionTag
+from settings_library.catalog import CatalogSettings
 from settings_library.director_v0 import DirectorV0Settings
 from settings_library.director_v2 import DirectorV2Settings
 from settings_library.docker_api_proxy import DockerApiProxysettings
@@ -145,6 +146,11 @@ class ApplicationSettings(_BaseApplicationSettings):
         description="settings for director-v2 service",
     )
 
+    DYNAMIC_SCHEDULER_CATALOG_SETTINGS: CatalogSettings = Field(
+        json_schema_extra={"auto_default_from_env": True},
+        description="settings for catalog service",
+    )
+
     DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED: bool = True
 
     DYNAMIC_SCHEDULER_PROFILING: bool = False

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/catalog/__init__.py

@@ -0,0 +1,7 @@
+from ._public_client import CatalogPublicClient
+from ._setup import lifespan_catalog
+
+__all__: tuple[str, ...] = (
+    "CatalogPublicClient",
+    "lifespan_catalog",
+)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/catalog/_public_client.py

@@ -0,0 +1,34 @@
+from fastapi import FastAPI
+from models_library.api_schemas_catalog.services_specifications import (
+    ServiceSpecifications,
+)
+from models_library.service_settings_labels import SimcoreServiceLabels
+from models_library.services import ServiceKey, ServiceVersion
+from models_library.users import UserID
+from pydantic import TypeAdapter
+from servicelib.fastapi.app_state import SingletonInAppStateMixin
+
+from ._thin_client import CatalogThinClient
+
+
+class CatalogPublicClient(SingletonInAppStateMixin):
+    app_state_name: str = "catalog_public_client"
+
+    def __init__(self, app: FastAPI) -> None:
+        self.app = app
+
+    async def get_services_labels(
+        self, service_key: ServiceKey, service_version: ServiceVersion
+    ) -> SimcoreServiceLabels:
+        response = await CatalogThinClient.get_from_app_state(
+            self.app
+        ).get_services_labels(service_key, service_version)
+        return TypeAdapter(SimcoreServiceLabels).validate_python(response.json())
+
+    async def get_services_specifications(
+        self, user_id: UserID, service_key: ServiceKey, service_version: ServiceVersion
+    ) -> ServiceSpecifications:
+        response = await CatalogThinClient.get_from_app_state(
+            self.app
+        ).get_services_specifications(user_id, service_key, service_version)
+        return TypeAdapter(ServiceSpecifications).validate_python(response.json())

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/catalog/_setup.py

@@ -0,0 +1,21 @@
+from collections.abc import AsyncIterator
+
+from fastapi import FastAPI
+from fastapi_lifespan_manager import State
+
+from ._public_client import CatalogPublicClient
+from ._thin_client import CatalogThinClient
+
+
+async def lifespan_catalog(app: FastAPI) -> AsyncIterator[State]:
+    thin_client = CatalogThinClient(app)
+    thin_client.set_to_app_state(app)
+    thin_client.attach_lifespan_to(app)
+
+    public_client = CatalogPublicClient(app)
+    public_client.set_to_app_state(app)
+
+    yield {}
+
+    CatalogPublicClient.pop_from_app_state(app)
+    CatalogThinClient.pop_from_app_state(app)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/catalog/_thin_client.py

@@ -0,0 +1,57 @@
+import urllib.parse
+
+from fastapi import FastAPI, status
+from httpx import Response
+from models_library.services import ServiceKey, ServiceVersion
+from models_library.users import UserID
+from servicelib.fastapi.app_state import SingletonInAppStateMixin
+from servicelib.fastapi.http_client import AttachLifespanMixin
+from servicelib.fastapi.http_client_thin import (
+    BaseThinClient,
+    expect_status,
+    retry_on_errors,
+)
+from yarl import URL
+
+from ...core.settings import ApplicationSettings
+
+
+class CatalogThinClient(SingletonInAppStateMixin, BaseThinClient, AttachLifespanMixin):
+    app_state_name: str = "catalog_thin_client"
+
+    def __init__(self, app: FastAPI) -> None:
+        settings: ApplicationSettings = app.state.settings
+        assert settings.CLIENT_REQUEST.HTTP_CLIENT_REQUEST_TOTAL_TIMEOUT  # nosec
+
+        super().__init__(
+            total_retry_interval=int(
+                settings.CLIENT_REQUEST.HTTP_CLIENT_REQUEST_TOTAL_TIMEOUT
+            ),
+            extra_allowed_method_names={
+                "attach_lifespan_to",
+                "get_from_app_state",
+                "pop_from_app_state",
+                "set_to_app_state",
+            },
+            base_url=settings.DYNAMIC_SCHEDULER_CATALOG_SETTINGS.api_base_url,
+            tracing_settings=settings.DYNAMIC_SCHEDULER_TRACING,
+        )
+
+    @retry_on_errors()
+    @expect_status(status.HTTP_200_OK)
+    async def get_services_labels(
+        self, service_key: ServiceKey, service_version: ServiceVersion
+    ) -> Response:
+        return await self.client.get(
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}/labels"
+        )
+
+    @retry_on_errors()
+    @expect_status(status.HTTP_200_OK)
+    async def get_services_specifications(
+        self, user_id: UserID, service_key: ServiceKey, service_version: ServiceVersion
+    ) -> Response:
+        request_url = URL(
+            f"/services/{urllib.parse.quote(service_key, safe='')}/{service_version}/specifications",
+        ).with_query(user_id=user_id)
+        return await self.client.get(f"{request_url}")

services/dynamic-scheduler/tests/conftest.py

@@ -25,6 +25,7 @@
     "pytest_simcore.docker_swarm",
     "pytest_simcore.environment_configs",
     "pytest_simcore.faker_projects_data",
+    "pytest_simcore.faker_users_data",
     "pytest_simcore.rabbit_service",
     "pytest_simcore.redis_service",
     "pytest_simcore.repository_paths",

services/dynamic-scheduler/tests/unit/test_services_catalog.py

@@ -0,0 +1,118 @@
+# pylint:disable=redefined-outer-name
+# pylint:disable=unused-argument
+
+
+import urllib.parse
+from collections.abc import Iterator
+
+import pytest
+import respx
+from fastapi import FastAPI
+from models_library.api_schemas_catalog.services_specifications import (
+    ServiceSpecifications,
+)
+from models_library.service_settings_labels import SimcoreServiceLabels
+from models_library.services import ServiceKey, ServiceVersion
+from models_library.users import UserID
+from pydantic import TypeAdapter
+from pytest_simcore.helpers.typing_env import EnvVarsDict
+from simcore_service_dynamic_scheduler.services.catalog import CatalogPublicClient
+
+
+@pytest.fixture
+def app_environment(
+    disable_redis_lifespan: None,
+    disable_rabbitmq_lifespan: None,
+    disable_service_tracker_lifespan: None,
+    disable_deferred_manager_lifespan: None,
+    disable_notifier_lifespan: None,
+    disable_status_monitor_lifespan: None,
+    app_environment: EnvVarsDict,
+) -> EnvVarsDict:
+    return app_environment
+
+
+@pytest.fixture
+def simcore_service_labels() -> SimcoreServiceLabels:
+    return TypeAdapter(SimcoreServiceLabels).validate_python(
+        SimcoreServiceLabels.model_json_schema()["examples"][1]
+    )
+
+
+@pytest.fixture
+def service_specifications() -> ServiceSpecifications:
+    return TypeAdapter(ServiceSpecifications).validate_python({})
+
+
+@pytest.fixture
+def service_version() -> ServiceVersion:
+    return "1.0.0"
+
+
+@pytest.fixture
+def service_key() -> ServiceKey:
+    return "simcore/services/dynamic/test"
+
+
+@pytest.fixture
+def mock_catalog(
+    app: FastAPI,
+    user_id: UserID,
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+    simcore_service_labels: SimcoreServiceLabels,
+    service_specifications: ServiceSpecifications,
+) -> Iterator[None]:
+    with respx.mock(
+        base_url=app.state.settings.DYNAMIC_SCHEDULER_CATALOG_SETTINGS.api_base_url,
+        assert_all_called=False,
+        assert_all_mocked=True,  # IMPORTANT: KEEP always True!
+    ) as respx_mock:
+        respx_mock.get(
+            f"/services/{urllib.parse.quote_plus(service_key)}/{service_version}/labels",
+            name="service labels",
+        ).respond(
+            status_code=200,
+            json=simcore_service_labels.model_dump(mode="json"),
+        )
+
+        respx_mock.get(
+            f"/services/{urllib.parse.quote_plus(service_key)}/{service_version}/specifications?user_id={user_id}",
+            name="service specifications",
+        ).respond(
+            status_code=200,
+            json=service_specifications.model_dump(mode="json"),
+        )
+
+        yield
+
+
+async def test_get_services_labels(
+    mock_catalog: None,
+    app: FastAPI,
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+    simcore_service_labels: SimcoreServiceLabels,
+):
+    client = CatalogPublicClient.get_from_app_state(app)
+    result = await client.get_services_labels(service_key, service_version)
+    assert result.model_dump(mode="json") == simcore_service_labels.model_dump(
+        mode="json"
+    )
+
+
+async def test_get_services_specifications(
+    mock_catalog: None,
+    app: FastAPI,
+    user_id: UserID,
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+    service_specifications: ServiceSpecifications,
+):
+    client = CatalogPublicClient.get_from_app_state(app)
+    result = await client.get_services_specifications(
+        user_id, service_key, service_version
+    )
+    assert result.model_dump(mode="json") == service_specifications.model_dump(
+        mode="json"
+    )