added catalog client to dynamic-scheduler

Author: Andrei Neagu

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()
@@ -109,12 +109,12 @@ async def get_service_resources(
             return json_response
         raise HTTPException(status_code=resp.status_code, detail=resp.content)
 
-    async def get_service_labels(
+    async def get_service_labels(  # to add
         self, service_key: ServiceKey, service_version: ServiceVersion
     ) -> 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:
@@ -132,12 +132,12 @@ async def get_service_extras(
             return ServiceExtras.model_validate(resp.json())
         raise HTTPException(status_code=resp.status_code, detail=resp.content)
 
-    async def get_service_specifications(
+    async def get_service_specifications(  # to add
         self, user_id: UserID, service_key: ServiceKey, service_version: ServiceVersion
     ) -> 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/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/application.py

@@ -18,6 +18,7 @@
 from ..api.frontend import setup_frontend
 from ..api.rest.routes import setup_rest_api
 from ..api.rpc.routes import setup_rpc_api_routes
+from ..services.catalog import setup_catalog
 from ..services.deferred_manager import setup_deferred_manager
 from ..services.director_v0 import setup_director_v0
 from ..services.director_v2 import setup_director_v2
@@ -64,6 +65,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
 
     setup_director_v2(app)
     setup_director_v0(app)
+    setup_catalog(app)
 
     setup_rabbitmq(app)
     setup_rpc_api_routes(app)

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.http_client_request import ClientRequestSettings
@@ -144,6 +145,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 setup_catalog
+
+__all__: tuple[str, ...] = (
+    "CatalogPublicClient",
+    "setup_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 fastapi import FastAPI
+
+from ._public_client import CatalogPublicClient
+from ._thin_client import CatalogThinClient
+
+
+def setup_catalog(app: FastAPI) -> None:
+    async def _on_startup() -> None:
+        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)
+
+    async def _on_shutdown() -> None:
+        CatalogPublicClient.pop_from_app_state(app)
+        CatalogThinClient.pop_from_app_state(app)
+
+    app.add_event_handler("startup", _on_startup)
+    app.add_event_handler("shutdown", _on_shutdown)

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/unit/test_services_catalog.py

@@ -0,0 +1,125 @@
+# 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_setup: None,
+    disable_rabbitmq_setup: None,
+    disable_service_tracker_setup: None,
+    disable_deferred_manager_setup: None,
+    disable_notifier_setup: None,
+    disable_status_monitor_setup: 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 user_id() -> UserID:
+    return 1
+
+
+@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"),
+        )
+
+        print("Respx mock paths:", [route.pattern for route in respx_mock.routes])
+
+        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"
+    )

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

@@ -10,7 +10,6 @@
 from models_library.api_schemas_directorv2.dynamic_services_service import (
     RunningDynamicServiceDetails,
 )
-from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from pydantic import TypeAdapter
 from pytest_simcore.helpers.typing_env import EnvVarsDict
@@ -33,9 +32,7 @@ def app_environment(
 
 
 @pytest.fixture
-def legacy_service_details(
-    node_id: NodeID, project_id: ProjectID
-) -> RunningDynamicServiceDetails:
+def legacy_service_details() -> RunningDynamicServiceDetails:
     return TypeAdapter(RunningDynamicServiceDetails).validate_python(
         RunningDynamicServiceDetails.model_json_schema()["examples"][0]
     )