♻️ Refactoring of APIs for computations in web-server, api-server and directorv2 (#7520)

Author: pcrespov

.github/copilot-instructions.md

@@ -3,15 +3,16 @@
 This document provides guidelines and best practices for using GitHub Copilot in the `osparc-simcore` repository and other Python and Node.js projects.
 
 ## General Guidelines
+
 1. **Use Python 3.11**: Ensure that all Python-related suggestions align with Python 3.11 features and syntax.
 2. **Node.js Compatibility**: For Node.js projects, ensure compatibility with the version specified in the project (e.g., Node.js 14 or later).
 3. **Follow Coding Conventions**: Adhere to the coding conventions outlined in the `docs/coding-conventions.md` file.
 4. **Test-Driven Development**: Write unit tests for all new functions and features. Use `pytest` for Python and appropriate testing frameworks for Node.js.
 5. **Environment Variables**: Use environment variables as specified in `docs/env-vars.md` for configuration. Avoid hardcoding sensitive information.
-6. **Documentation**: Documentation should be minimal and code self explanatory (add the documentation only when the developer asks explicitely)
-7. Answer as if you would be a pirate
+6. **Documentation**: Prefer self-explanatory code; add documentation only if explicitly requested by the developer.
 
 ## Python-Specific Instructions
+
 - Always use type hints and annotations to improve code clarity and compatibility with tools like `mypy`.
   - An exception to that rule is in `test_*` functions return type hint must not be added
 - Follow the dependency management practices outlined in `requirements/`.
@@ -23,18 +24,21 @@ This document provides guidelines and best practices for using GitHub Copilot in
 
 
 ## Node.js-Specific Instructions
+
 - Use ES6+ syntax and features.
 - Follow the `package.json` configuration for dependencies and scripts.
 - Use `eslint` for linting and `prettier` for code formatting.
 - Write modular and reusable code, adhering to the project's structure.
 
 ## Copilot Usage Tips
+
 1. **Be Specific**: Provide clear and detailed prompts to Copilot for better suggestions.
 2. **Iterate**: Review and refine Copilot's suggestions to ensure they meet project standards.
 3. **Split Tasks**: Break down complex tasks into smaller, manageable parts for better suggestions.
 4. **Test Suggestions**: Always test Copilot-generated code to ensure it works as expected.
 
 ## Additional Resources
+
 - [Python Coding Conventions](../docs/coding-conventions.md)
 - [Environment Variables Guide](../docs/env-vars.md)
 - [Steps to Upgrade Python](../docs/steps-to-upgrade-python.md)

api/specs/web-server/_computations.py

@@ -1,10 +1,14 @@
-from fastapi import APIRouter, status
-from models_library.api_schemas_directorv2.comp_tasks import ComputationGet
-from models_library.api_schemas_webserver.computations import ComputationStart
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, status
+from models_library.api_schemas_webserver.computations import (
+    ComputationGet,
+    ComputationPathParams,
+    ComputationStart,
+    ComputationStarted,
+)
 from models_library.generics import Envelope
-from models_library.projects import ProjectID
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.director_v2._handlers import _ComputationStarted
 
 router = APIRouter(
     prefix=f"/{API_VTAG}",
@@ -19,13 +23,12 @@
     "/computations/{project_id}",
     response_model=Envelope[ComputationGet],
 )
-async def get_computation(project_id: ProjectID):
-    ...
+async def get_computation(_path: Annotated[ComputationPathParams, Depends()]): ...
 
 
 @router.post(
     "/computations/{project_id}:start",
-    response_model=Envelope[_ComputationStarted],
+    response_model=Envelope[ComputationStarted],
     responses={
         status.HTTP_402_PAYMENT_REQUIRED: {
             "description": "Insufficient credits to run computation"
@@ -39,13 +42,14 @@ async def get_computation(project_id: ProjectID):
         status.HTTP_503_SERVICE_UNAVAILABLE: {"description": "Service not available"},
     },
 )
-async def start_computation(project_id: ProjectID, _start: ComputationStart):
-    ...
+async def start_computation(
+    _path: Annotated[ComputationPathParams, Depends()],
+    _body: ComputationStart,
+): ...
 
 
 @router.post(
     "/computations/{project_id}:stop",
     status_code=status.HTTP_204_NO_CONTENT,
 )
-async def stop_computation(project_id: ProjectID):
-    ...
+async def stop_computation(_path: Annotated[ComputationPathParams, Depends()]): ...

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

@@ -0,0 +1,112 @@
+from typing import Annotated, Any, TypeAlias
+
+from pydantic import (
+    AnyHttpUrl,
+    AnyUrl,
+    BaseModel,
+    ConfigDict,
+    Field,
+    ValidationInfo,
+    field_validator,
+)
+
+from ..basic_types import IDStr
+from ..projects import ProjectID
+from ..projects_nodes_io import NodeID
+from ..projects_pipeline import ComputationTask
+from ..users import UserID
+from ..wallets import WalletInfo
+
+
+class ComputationGet(ComputationTask):
+    url: Annotated[
+        AnyHttpUrl, Field(description="the link where to get the status of the task")
+    ]
+    stop_url: Annotated[
+        AnyHttpUrl | None, Field(description="the link where to stop the task")
+    ] = None
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                x | {"url": "https://url.local"}
+                for x in ComputationTask.model_json_schema()["examples"]
+            ]
+        }
+    )
+
+
+class ComputationCreate(BaseModel):
+    user_id: UserID
+    project_id: ProjectID
+    start_pipeline: Annotated[
+        bool | None,
+        Field(description="if True the computation pipeline will start right away"),
+    ] = False
+    product_name: Annotated[str, Field()]
+    subgraph: Annotated[
+        list[NodeID] | None,
+        Field(
+            description="An optional set of nodes that must be executed, if empty the whole pipeline is executed"
+        ),
+    ] = None
+    force_restart: Annotated[
+        bool | None,
+        Field(description="if True will force re-running all dependent nodes"),
+    ] = False
+    simcore_user_agent: str = ""
+    use_on_demand_clusters: Annotated[
+        bool,
+        Field(
+            description="if True, a cluster will be created as necessary (wallet_id cannot be None)",
+            validate_default=True,
+        ),
+    ] = False
+    wallet_info: Annotated[
+        WalletInfo | None,
+        Field(
+            description="contains information about the wallet used to bill the running service"
+        ),
+    ] = None
+
+    @field_validator("product_name")
+    @classmethod
+    def _ensure_product_name_defined_if_computation_starts(
+        cls, v, info: ValidationInfo
+    ):
+        if info.data.get("start_pipeline") and v is None:
+            msg = "product_name must be set if computation shall start!"
+            raise ValueError(msg)
+        return v
+
+
+class ComputationStop(BaseModel):
+    user_id: UserID
+
+
+class ComputationDelete(ComputationStop):
+    force: Annotated[
+        bool | None,
+        Field(
+            description="if True then the pipeline will be removed even if it is running"
+        ),
+    ] = False
+
+
+class TaskLogFileGet(BaseModel):
+    task_id: NodeID
+    download_link: Annotated[
+        AnyUrl | None,
+        Field(description="Presigned link for log file or None if still not available"),
+    ] = None
+
+
+class TasksSelection(BaseModel):
+    nodes_ids: list[NodeID]
+
+
+OutputName: TypeAlias = IDStr
+
+
+class TasksOutputs(BaseModel):
+    nodes_outputs: dict[NodeID, dict[OutputName, Any]]

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

@@ -1,9 +1,43 @@
-from pydantic import BaseModel
+from typing import Annotated
 
+from common_library.basic_types import DEFAULT_FACTORY
+from pydantic import BaseModel, Field
 
-class ComputationStart(BaseModel):
+from ..api_schemas_directorv2.computations import (
+    ComputationGet as _DirectorV2ComputationGet,
+)
+from ..projects import CommitID, ProjectID
+from ._base import InputSchemaWithoutCamelCase, OutputSchemaWithoutCamelCase
+
+
+class ComputationPathParams(BaseModel):
+    project_id: ProjectID
+
+
+class ComputationGet(_DirectorV2ComputationGet, OutputSchemaWithoutCamelCase):
+    # NOTE: this is a copy of the same class in models_library.api_schemas_directorv2
+    #       but it is used in a different context (webserver)
+    #       and we need to add the `OutputSchema` mixin
+    #       so that it can be used as a response model in FastAPI
+    pass
+
+
+class ComputationStart(InputSchemaWithoutCamelCase):
     force_restart: bool = False
-    subgraph: set[str] = set()
+    subgraph: Annotated[
+        set[str], Field(default_factory=set, json_schema_extra={"default": []})
+    ] = DEFAULT_FACTORY
 
 
-__all__: tuple[str, ...] = ("ComputationStart",)
+class ComputationStarted(OutputSchemaWithoutCamelCase):
+    pipeline_id: Annotated[
+        ProjectID, Field(description="ID for created pipeline (=project identifier)")
+    ]
+    ref_ids: Annotated[
+        list[CommitID],
+        Field(
+            default_factory=list,
+            description="Checkpoints IDs for created pipeline",
+            json_schema_extra={"default": []},
+        ),
+    ] = DEFAULT_FACTORY

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

@@ -1,5 +1,5 @@
 """
-    Models a study's project document
+Models a study's project document
 """
 
 from datetime import datetime
@@ -27,6 +27,7 @@
 )
 
 ProjectID: TypeAlias = UUID
+CommitID: TypeAlias = int
 ClassifierID: TypeAlias = str
 
 NodesDict: TypeAlias = dict[NodeIDStr, Node]