drafts catalog service

Author: pcrespov

services/api-server/src/simcore_service_api_server/services_http/catalog.py

@@ -9,7 +9,9 @@
 
 from fastapi import FastAPI, status
 from models_library.emails import LowerCaseEmailStr
+from models_library.products import ProductName
 from models_library.services import ServiceMetaDataPublished, ServiceType
+from models_library.users import UserID
 from pydantic import ConfigDict, TypeAdapter, ValidationError
 from settings_library.catalog import CatalogSettings
 from settings_library.tracing import TracingSettings
@@ -70,9 +72,9 @@ def to_solver(self) -> Solver:
 
 _exception_mapper = partial(service_exception_mapper, service_name="Catalog")
 
-TruncatedCatalogServiceOutAdapter: Final[
-    TypeAdapter[TruncatedCatalogServiceOut]
-] = TypeAdapter(TruncatedCatalogServiceOut)
+TruncatedCatalogServiceOutAdapter: Final[TypeAdapter[TruncatedCatalogServiceOut]] = (
+    TypeAdapter(TruncatedCatalogServiceOut)
+)
 TruncatedCatalogServiceOutListAdapter: Final[
     TypeAdapter[list[TruncatedCatalogServiceOut]]
 ] = TypeAdapter(list[TruncatedCatalogServiceOut])
@@ -97,8 +99,8 @@ class CatalogApi(BaseServiceClientApi):
     async def list_solvers(
         self,
         *,
-        user_id: int,
-        product_name: str,
+        user_id: UserID,
+        product_name: ProductName,
         predicate: Callable[[Solver], bool] | None = None,
     ) -> list[Solver]:
 
@@ -140,7 +142,12 @@ async def list_solvers(
         http_status_map={status.HTTP_404_NOT_FOUND: SolverOrStudyNotFoundError}
     )
     async def get_service(
-        self, *, user_id: int, name: SolverKeyId, version: VersionStr, product_name: str
+        self,
+        *,
+        user_id: UserID,
+        name: SolverKeyId,
+        version: VersionStr,
+        product_name: ProductName,
     ) -> Solver:
 
         assert version != LATEST_VERSION  # nosec
@@ -171,8 +178,13 @@ async def get_service(
         http_status_map={status.HTTP_404_NOT_FOUND: SolverOrStudyNotFoundError}
     )
     async def get_service_ports(
-        self, *, user_id: int, name: SolverKeyId, version: VersionStr, product_name: str
-    ):
+        self,
+        *,
+        user_id: UserID,
+        name: SolverKeyId,
+        version: VersionStr,
+        product_name: ProductName,
+    ) -> list[SolverPort]:
 
         assert version != LATEST_VERSION  # nosec
 
@@ -190,7 +202,7 @@ async def get_service_ports(
         return TypeAdapter(list[SolverPort]).validate_python(response.json())
 
     async def list_latest_releases(
-        self, *, user_id: int, product_name: str
+        self, *, user_id: UserID, product_name: ProductName
     ) -> list[Solver]:
         solvers: list[Solver] = await self.list_solvers(
             user_id=user_id, product_name=product_name
@@ -205,7 +217,7 @@ async def list_latest_releases(
         return list(latest_releases.values())
 
     async def list_solver_releases(
-        self, *, user_id: int, solver_key: SolverKeyId, product_name: str
+        self, *, user_id: UserID, solver_key: SolverKeyId, product_name: ProductName
     ) -> list[Solver]:
         def _this_solver(solver: Solver) -> bool:
             return solver.id == solver_key
@@ -216,7 +228,7 @@ def _this_solver(solver: Solver) -> bool:
         return releases
 
     async def get_latest_release(
-        self, *, user_id: int, solver_key: SolverKeyId, product_name: str
+        self, *, user_id: UserID, solver_key: SolverKeyId, product_name: ProductName
     ) -> Solver:
         releases = await self.list_solver_releases(
             user_id=user_id, solver_key=solver_key, product_name=product_name

services/api-server/src/simcore_service_api_server/services_rpc/catalog.py

@@ -0,0 +1,83 @@
+from models_library.basic_types import VersionStr
+from models_library.products import ProductName
+from models_library.rest_pagination import (
+    DEFAULT_NUMBER_OF_ITEMS_PER_PAGE,
+    PageLimitInt,
+    PageOffsetInt,
+)
+from models_library.users import UserID
+
+from ..models.schemas.solvers import Solver, SolverKeyId, SolverPort
+
+_FAKE: list[Solver] = [
+    Solver.model_validate(Solver.model_json_schema()["example"]),
+]
+_FAKE2: list[SolverPort] = [
+    SolverPort.model_validate(SolverPort.model_json_schema()["example"]),
+]
+
+
+async def list_latest_releases(
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    offset: PageOffsetInt = 0,
+    limit: PageLimitInt = DEFAULT_NUMBER_OF_ITEMS_PER_PAGE,
+) -> list[Solver]:
+    assert product_name  # nosec
+    assert user_id  # nosec
+
+    return _FAKE[offset : offset + limit]
+
+
+async def list_solver_releases(
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    solver_id: SolverKeyId,
+    offset: PageOffsetInt = 0,
+    limit: PageLimitInt = DEFAULT_NUMBER_OF_ITEMS_PER_PAGE,
+) -> list[Solver]:
+    assert product_name  # nosec
+    assert user_id  # nosec
+    return [solver for solver in _FAKE if solver.id == solver_id][
+        offset : offset + limit
+    ]
+
+
+async def get_solver(
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    solver_id: SolverKeyId,
+    solver_version: VersionStr,
+) -> Solver | None:
+    assert product_name  # nosec
+    assert user_id  # nosec
+
+    return next(
+        (
+            solver
+            for solver in _FAKE
+            if solver.id == solver_id and solver.version == solver_version
+        ),
+        None,
+    )
+
+
+async def get_solver_ports(
+    *,
+    product_name: ProductName,
+    user_id: int,
+    solver_id: SolverKeyId,
+    solver_version: VersionStr,
+) -> list[SolverPort]:
+
+    if await get_solver(
+        product_name=product_name,
+        user_id=user_id,
+        solver_id=solver_id,
+        solver_version=solver_version,
+    ):
+        return _FAKE2
+    return []

services/api-server/tests/unit/test_services_catalog.py

@@ -0,0 +1,52 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+import pytest
+from models_library.products import ProductName
+from models_library.users import UserID
+from simcore_service_api_server.models.schemas.solvers import Solver, SolverPort
+from simcore_service_api_server.services_rpc import catalog as catalog_service
+
+
+@pytest.fixture
+def product_name() -> ProductName:
+    return "osparc"
+
+
+async def test_catalog_service_read_solvers(product_name: ProductName, user_id: UserID):
+    # Step 1: List latest releases in a page
+    solver_releases_page: list[Solver] = await catalog_service.list_latest_releases(
+        product_name=product_name, user_id=user_id
+    )
+    assert solver_releases_page, "Releases page should not be empty"
+
+    # Step 2: Select one release and list solver releases
+    selected_release = solver_releases_page[0]
+    solver_releases = await catalog_service.list_solver_releases(
+        product_name=product_name,
+        user_id=user_id,
+        solver_id=selected_release.id,
+    )
+    assert solver_releases, "Solver releases should not be empty"
+
+    # Step 3: Take the latest solver release and get solver details
+    latest_solver_release = solver_releases[0]
+    solver_details: Solver | None = await catalog_service.get_solver(
+        product_name=product_name,
+        user_id=user_id,
+        solver_id=latest_solver_release.id,
+        solver_version=latest_solver_release.version,
+    )
+    assert solver_details, "Solver details should not be empty"
+
+    # Step 4: Get solver ports
+    solver_ports: list[SolverPort] = await catalog_service.get_solver_ports(
+        product_name=product_name,
+        user_id=user_id,
+        solver_id=latest_solver_release.id,
+        solver_version=latest_solver_release.version,
+    )
+    assert solver_ports, "Solver ports should not be empty"