intro

Author: matusdrobuliak66

api/specs/web-server/_computations.py

@@ -1,5 +1,6 @@
 from typing import Annotated
 
+from _common import as_query
 from fastapi import APIRouter, Depends, status
 from models_library.api_schemas_webserver.computations import (
     ComputationGet,
@@ -9,6 +10,11 @@
 )
 from models_library.generics import Envelope
 from simcore_service_webserver._meta import API_VTAG
+from simcore_service_webserver.director_v2._computations_rest_schema import (
+    ComputationRunListQueryParams,
+    ComputationTaskListQueryParams,
+    ComputationTaskPathParams,
+)
 
 router = APIRouter(
     prefix=f"/{API_VTAG}",
@@ -53,3 +59,26 @@ async def start_computation(
     status_code=status.HTTP_204_NO_CONTENT,
 )
 async def stop_computation(_path: Annotated[ComputationPathParams, Depends()]): ...
+
+
+@router.get(
+    "/computations/-/iterations/latest",
+    response_model=Envelope[list[ComputationGet]],
+    name="list_computations_latest_iteration",
+    description="Lists the latest iteration of computations",
+)
+async def list_computations_latest_iteration(
+    _query: Annotated[as_query(ComputationRunListQueryParams), Depends()],
+): ...
+
+
+@router.get(
+    "/computations/{project_id}/iterations/latest/tasks",
+    response_model=Envelope[list[ComputationGet]],
+    name="list_computations_latest_iteration_tasks",
+    description="Lists the latest iteration tasks for a computation",
+)
+async def list_computations_latest_iteration_tasks(
+    _query: Annotated[as_query(ComputationTaskListQueryParams), Depends()],
+    _path: Annotated[ComputationTaskPathParams, Depends()],
+): ...

packages/models-library/src/models_library/api_schemas_directorv2/__init__.py

@@ -1,8 +1,19 @@
+from typing import Final
+
+from pydantic import TypeAdapter
+
+from ..rabbitmq_basic_types import RPCNamespace
 from . import clusters, dynamic_services
 
 assert clusters  # nosec
 assert dynamic_services  # nosec
 
+
+DIRECTOR_V2_RPC_NAMESPACE: Final[RPCNamespace] = TypeAdapter(
+    RPCNamespace
+).validate_python("director-v2")
+
+
 __all__: tuple[str, ...] = (
     "clusters",
     "dynamic_services",

packages/models-library/src/models_library/api_schemas_directorv2/comp_runs.py

@@ -0,0 +1,41 @@
+from datetime import datetime
+from typing import Any, NamedTuple
+
+from pydantic import (
+    BaseModel,
+    PositiveInt,
+)
+
+from ..projects import ProjectID
+from ..projects_nodes_io import NodeID
+from ..projects_state import RunningState
+
+
+class ComputationRunRpcGet(BaseModel):
+    project_uuid: ProjectID
+    iteration: int
+    state: RunningState
+    info: dict[str, Any]
+    submitted_at: datetime
+    started_at: datetime | None
+    ended_at: datetime | None
+
+
+class ComputationRunRpcGetPage(NamedTuple):
+    items: list[ComputationRunRpcGet]
+    total: PositiveInt
+
+
+class ComputationTaskRpcGet(BaseModel):
+    project_uuid: ProjectID
+    node_id: NodeID
+    state: RunningState
+    progress: float
+    image: dict[str, Any]
+    started_at: datetime | None
+    ended_at: datetime | None
+
+
+class ComputationTaskRpcGetPage(NamedTuple):
+    items: list[ComputationTaskRpcGet]
+    total: PositiveInt

packages/models-library/src/models_library/api_schemas_webserver/computations.py

@@ -1,12 +1,19 @@
-from typing import Annotated
+from datetime import datetime
+from typing import Annotated, Any, NamedTuple
 
 from common_library.basic_types import DEFAULT_FACTORY
-from pydantic import BaseModel, Field
+from pydantic import (
+    BaseModel,
+    Field,
+    PositiveInt,
+)
 
 from ..api_schemas_directorv2.computations import (
     ComputationGet as _DirectorV2ComputationGet,
 )
 from ..projects import CommitID, ProjectID
+from ..projects_nodes_io import NodeID
+from ..projects_state import RunningState
 from ._base import InputSchemaWithoutCamelCase, OutputSchemaWithoutCamelCase
 
 
@@ -41,3 +48,33 @@ class ComputationStarted(OutputSchemaWithoutCamelCase):
             json_schema_extra={"default": []},
         ),
     ] = DEFAULT_FACTORY
+
+
+class ComputationRunRestGet(BaseModel):
+    project_uuid: ProjectID
+    iteration: int
+    state: RunningState
+    info: dict[str, Any]
+    submitted_at: datetime
+    started_at: datetime | None
+    ended_at: datetime | None
+
+
+class ComputationRunRestGetPage(NamedTuple):
+    items: list[ComputationRunRestGet]
+    total: PositiveInt
+
+
+class ComputationTaskRestGet(BaseModel):
+    project_uuid: ProjectID
+    node_id: NodeID
+    state: RunningState
+    progress: float
+    image: dict[str, Any]
+    started_at: datetime | None
+    ended_at: datetime | None
+
+
+class ComputationTaskRestGetPage(NamedTuple):
+    items: list[ComputationTaskRestGet]
+    total: PositiveInt

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/director_v2/__init__.py

@@ -0,0 +1,85 @@
+# pylint: disable=too-many-arguments
+import logging
+from typing import Final
+
+from models_library.api_schemas_directorv2 import (
+    DIRECTOR_V2_RPC_NAMESPACE,
+)
+from models_library.api_schemas_directorv2.comp_runs import (
+    ComputationRunRpcGetPage,
+    ComputationTaskRpcGetPage,
+)
+from models_library.products import ProductName
+from models_library.projects import ProjectID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.rest_ordering import OrderBy
+from models_library.users import UserID
+from pydantic import NonNegativeInt, TypeAdapter
+
+from ....logging_utils import log_decorator
+from ... import RabbitMQRPCClient
+
+_logger = logging.getLogger(__name__)
+
+
+_DEFAULT_TIMEOUT_S: Final[NonNegativeInt] = 20
+
+_RPC_METHOD_NAME_ADAPTER: TypeAdapter[RPCMethodName] = TypeAdapter(RPCMethodName)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def list_computations_latest_iteration_page(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    # pagination
+    offset: int = 0,
+    limit: int = 20,
+    # ordering
+    order_by: OrderBy | None = None,
+) -> ComputationRunRpcGetPage:
+    result = await rabbitmq_rpc_client.request(
+        DIRECTOR_V2_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python(
+            "list_computations_latest_iteration_page"
+        ),
+        product_name=product_name,
+        user_id=user_id,
+        offset=offset,
+        limit=limit,
+        order_by=order_by,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, ComputationRunRpcGetPage)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def list_computations_latest_iteration_tasks_page(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    project_id: ProjectID,
+    # pagination
+    offset: int = 0,
+    limit: int = 20,
+    # ordering
+    order_by: OrderBy | None = None,
+) -> ComputationTaskRpcGetPage:
+    result = await rabbitmq_rpc_client.request(
+        DIRECTOR_V2_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python(
+            "list_computations_latest_iteration_tasks_page"
+        ),
+        product_name=product_name,
+        user_id=user_id,
+        project_id=project_id,
+        offset=offset,
+        limit=limit,
+        order_by=order_by,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, ComputationTaskRpcGetPage)  # nosec
+    return result

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/director_v2/computations.py

@@ -3,7 +3,7 @@
 from typing import TypeVar, cast
 
 from aiopg.sa import Engine
-from fastapi import Depends
+from fastapi import Depends, FastAPI
 from fastapi.requests import Request
 
 from ...modules.db.repositories import BaseRepository
@@ -47,3 +47,11 @@ async def _get_repo(
         yield get_base_repository(engine=engine, repo_type=repo_type)
 
     return _get_repo
+
+
+def get_repository_instance(app: FastAPI, repo_type: type[RepoType]) -> RepoType:
+    """
+    Retrieves an instance of the specified repository type using the database engine from the FastAPI app.
+    """
+    engine = cast(Engine, app.state.engine)
+    return get_base_repository(engine=engine, repo_type=repo_type)

services/director-v2/src/simcore_service_director_v2/api/dependencies/database.py

@@ -0,0 +1,74 @@
+# pylint: disable=too-many-arguments
+from fastapi import FastAPI
+from models_library.api_schemas_directorv2.comp_runs import (
+    ComputationRunRpcGetPage,
+    ComputationTaskRpcGetPage,
+)
+from models_library.products import ProductName
+from models_library.projects import ProjectID
+from models_library.rest_ordering import OrderBy
+from models_library.users import UserID
+from servicelib.rabbitmq import RPCRouter
+
+from ...modules.db.repositories.comp_runs import CompRunsRepository
+from ...modules.db.repositories.comp_tasks import CompTasksRepository
+from ..dependencies.database import get_repository_instance
+
+router = RPCRouter()
+
+
+@router.expose(reraise_if_error_type=())
+async def list_computations_latest_iteration_page(
+    app: FastAPI,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    # pagination
+    offset: int = 0,
+    limit: int = 20,
+    # ordering
+    order_by: OrderBy | None = None,
+) -> ComputationRunRpcGetPage:
+    comp_runs_repo = get_repository_instance(app, CompRunsRepository)
+    total, comp_runs = await comp_runs_repo.list_for_user__only_latest_iterations(
+        product_name=product_name,
+        user_id=user_id,
+        offset=offset,
+        limit=limit,
+        order_by=order_by,
+    )
+    return ComputationRunRpcGetPage(
+        items=comp_runs,
+        total=total,
+    )
+
+
+@router.expose(reraise_if_error_type=())
+async def list_computations_latest_iteration_tasks_page(
+    app: FastAPI,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    project_id: ProjectID,
+    # pagination
+    offset: int = 0,
+    limit: int = 20,
+    # ordering
+    order_by: OrderBy | None = None,
+) -> ComputationTaskRpcGetPage:
+    assert product_name  # nosec  NOTE: Whether project_id belong to the product_name was checked in the webserver
+    assert user_id  # nosec  NOTE: Whether user_id has access to the project was checked in the webserver
+
+    comp_tasks_repo = get_repository_instance(app, CompTasksRepository)
+    total, comp_runs = (
+        await comp_tasks_repo.list_computational_tasks_for_frontend_client(
+            project_id=project_id,
+            offset=offset,
+            limit=limit,
+            order_by=order_by,
+        )
+    )
+    return ComputationTaskRpcGetPage(
+        items=comp_runs,
+        total=total,
+    )

services/director-v2/src/simcore_service_director_v2/api/rpc/__init__.py

@@ -0,0 +1,34 @@
+import logging
+
+from fastapi import FastAPI
+from models_library.api_schemas_directorv2 import (
+    DIRECTOR_V2_RPC_NAMESPACE,
+)
+from servicelib.logging_utils import log_context
+from servicelib.rabbitmq import RPCRouter
+
+from ...modules.rabbitmq import get_rabbitmq_rpc_server
+from . import (
+    _computations,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+ROUTERS: list[RPCRouter] = [
+    _computations.router,
+]
+
+
+def setup_rpc_api_routes(app: FastAPI) -> None:
+    async def startup() -> None:
+        with log_context(
+            _logger,
+            logging.INFO,
+            msg="Director-v2 startup RPC API Routes",
+        ):
+            rpc_server = get_rabbitmq_rpc_server(app)
+            for router in ROUTERS:
+                await rpc_server.register_router(router, DIRECTOR_V2_RPC_NAMESPACE, app)
+
+    app.add_event_handler("startup", startup)