extend and improve capturing mechanism

.env-devel

@@ -229,7 +229,7 @@ WEBSERVER_VERSION_CONTROL=1
 #
 # AIODEBUG_SLOW_DURATION_SECS=0.25
 # API_SERVER_DEV_FEATURES_ENABLED=1
-# API_SERVER_DEV_HTTP_CALLS_LOGS_PATH=captures.ignore.keep.log
+# API_SERVER_DEV_HTTP_CALLS_LOGS_PATH=captures.ignore.keep.json
 # PYTHONASYNCIODEBUG=1
 # PYTHONTRACEMALLOC=1
 #

services/api-server/src/simcore_service_api_server/api/routes/solvers_jobs.py

@@ -15,7 +15,6 @@
 from models_library.api_schemas_webserver.wallets import WalletGet
 from models_library.clusters import ClusterID
 from models_library.projects_nodes_io import BaseFileLink
-from pydantic import ValidationError, parse_obj_as
 from pydantic.types import PositiveInt
 from servicelib.logging_utils import log_context
 
@@ -80,13 +79,6 @@ def _raise_if_job_not_associated_with_solver(
         )
 
 
-def _get_pricing_plan_and_unit(request: Request) -> JobPricingSpecification | None:
-    try:
-        return parse_obj_as(JobPricingSpecification, request.headers)
-    except ValidationError:
-        return None
-
-
 # JOBS ---------------
 #
 # - Similar to docker container's API design (container = job and image = solver)
@@ -314,7 +306,7 @@ async def start_job(
     job_name = _compose_job_resource_name(solver_key, version, job_id)
     _logger.debug("Start Job '%s'", job_name)
 
-    if pricing_spec := _get_pricing_plan_and_unit(request):
+    if pricing_spec := JobPricingSpecification.create_from_headers(request.headers):
         with log_context(_logger, logging.DEBUG, "Set pricing plan and unit"):
             project: ProjectGet = await webserver_api.get_project(project_id=job_id)
             _raise_if_job_not_associated_with_solver(solver_key, version, project)

services/api-server/src/simcore_service_api_server/models/schemas/jobs.py

@@ -14,8 +14,11 @@
     StrictBool,
     StrictFloat,
     StrictInt,
+    ValidationError,
+    parse_obj_as,
     validator,
 )
+from starlette.datastructures import Headers
 
 from ...models.schemas.files import File
 from ...models.schemas.solvers import Solver
@@ -286,3 +289,10 @@ class JobPricingSpecification(BaseModel):
 
     class Config:
         extra = Extra.ignore
+
+    @classmethod
+    def create_from_headers(cls, headers: Headers) -> "JobPricingSpecification | None":
+        try:
+            return parse_obj_as(JobPricingSpecification, headers)
+        except ValidationError:
+            return None

services/api-server/src/simcore_service_api_server/utils/client_base.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 from contextlib import suppress
@@ -6,13 +7,16 @@
 
 import httpx
 from fastapi import FastAPI
+from fastapi.encoders import jsonable_encoder
 from httpx._types import URLTypes
 from jsonschema import ValidationError
+from pydantic import parse_file_as
+from simcore_service_api_server.utils.http_calls_capture import HttpApiCallCaptureModel
 
 from .app_data import AppDataMixin
 
 if os.environ.get("API_SERVER_DEV_HTTP_CALLS_LOGS_PATH"):
-    from .http_calls_capture import get_captured_as_json
+    from .http_calls_capture import get_captured
     from .http_calls_capture_processing import CaptureProcessingException
 
 _logger = logging.getLogger(__name__)
@@ -49,16 +53,36 @@ class _AsyncClientForDevelopmentOnly(httpx.AsyncClient):
     Adds captures mechanism
     """
 
+    def __init__(self, capture_file: Path, **async_clint_kwargs):
+        super().__init__(**async_clint_kwargs)
+        assert capture_file.name.endswith(
+            ".json"
+        ), "The capture file should be a json file"
+        self._capture_file: Path = capture_file
+
     async def request(self, method: str, url: URLTypes, **kwargs):
         response: httpx.Response = await super().request(method, url, **kwargs)
 
         capture_name = f"{method} {url}"
         _logger.info("Capturing %s ... [might be slow]", capture_name)
         try:
-            capture_json = get_captured_as_json(name=capture_name, response=response)
-            _capture_logger.info("%s,", capture_json)
+            capture: HttpApiCallCaptureModel = get_captured(
+                name=capture_name, response=response
+            )
+            if (
+                not self._capture_file.is_file()
+                or self._capture_file.read_text().strip() == ""
+            ):
+                self._capture_file.write_text("[]")
+            serialized_captures: list[HttpApiCallCaptureModel] = parse_file_as(
+                list[HttpApiCallCaptureModel], self._capture_file
+            )
+            serialized_captures.append(capture)
+            self._capture_file.write_text(
+                json.dumps(jsonable_encoder(serialized_captures), indent=1)
+            )
         except (CaptureProcessingException, ValidationError, httpx.RequestError):
-            _capture_logger.exception(
+            _logger.exception(
                 "Unexpected failure with %s",
                 capture_name,
                 exc_info=True,
@@ -69,24 +93,6 @@ async def request(self, method: str, url: URLTypes, **kwargs):
 
 # HELPERS -------------------------------------------------------------
 
-_capture_logger = logging.getLogger(f"{__name__}.capture")
-
-
-def _setup_capture_logger_once(capture_path: Path) -> None:
-    """NOTE: this is only to capture during development"""
-
-    if not any(
-        isinstance(hnd, logging.FileHandler) for hnd in _capture_logger.handlers
-    ):
-        file_handler = logging.FileHandler(filename=f"{capture_path}")
-        file_handler.setLevel(logging.INFO)
-
-        formatter = logging.Formatter("%(message)s")
-        file_handler.setFormatter(formatter)
-
-        _capture_logger.addHandler(file_handler)
-        _logger.info("Setup capture logger at %s", capture_path)
-
 
 def setup_client_instance(
     app: FastAPI,
@@ -100,20 +106,23 @@ def setup_client_instance(
     assert issubclass(api_cls, BaseServiceClientApi)  # nosec
 
     # Http client class
-    client_class: type = httpx.AsyncClient
+    client: httpx.AsyncClient | _AsyncClientForDevelopmentOnly = httpx.AsyncClient(
+        base_url=api_baseurl
+    )
     with suppress(AttributeError):
         # NOTE that this is a general function with no guarantees as when is going to be used.
         # Here, 'AttributeError' might be raied when app.state.settings is still not initialized
         if capture_path := app.state.settings.API_SERVER_DEV_HTTP_CALLS_LOGS_PATH:
-            _setup_capture_logger_once(capture_path)
-            client_class = _AsyncClientForDevelopmentOnly
+            client = _AsyncClientForDevelopmentOnly(
+                capture_file=capture_path, base_url=api_baseurl
+            )
 
     # events
     def _create_instance() -> None:
-        _logger.debug("Creating %s for %s", f"{client_class=}", f"{api_baseurl=}")
+        _logger.debug("Creating %s for %s", f"{type(client)=}", f"{api_baseurl=}")
         api_cls.create_once(
             app,
-            client=client_class(base_url=api_baseurl),
+            client=client,
             service_name=service_name,
             **extra_fields,
         )

services/api-server/src/simcore_service_api_server/utils/http_calls_capture.py

@@ -66,8 +66,5 @@ def as_response(self) -> httpx.Response:
         return httpx.Response(status_code=self.status_code, json=self.response_body)
 
 
-def get_captured_as_json(name: str, response: httpx.Response) -> str:
-    capture_json: str = HttpApiCallCaptureModel.create_from_response(
-        response, name=name
-    ).json(indent=1)
-    return f"{capture_json}"
+def get_captured(name: str, response: httpx.Response) -> HttpApiCallCaptureModel:
+    return HttpApiCallCaptureModel.create_from_response(response, name=name)

services/api-server/src/simcore_service_api_server/utils/http_calls_capture_processing.py

@@ -7,11 +7,12 @@
 from pydantic import BaseModel, Field, parse_obj_as, root_validator, validator
 from simcore_service_api_server.core.settings import (
     CatalogSettings,
+    DirectorV2Settings,
     StorageSettings,
     WebServerSettings,
 )
 
-service_hosts = Literal["storage", "catalog", "webserver"]
+service_hosts = Literal["storage", "catalog", "webserver", "director-v2"]
 
 
 class CapturedParameterSchema(BaseModel):
@@ -170,6 +171,9 @@ def _get_openapi_specs(host: service_hosts) -> dict[str, Any]:
     elif host == "webserver":
         settings = WebServerSettings()
         url = settings.base_url + "/dev/doc/swagger.json"
+    elif host == "director-v2":
+        settings = DirectorV2Settings()
+        url = settings.base_url + "/api/v2/openapi.json"
     else:
         raise OpenApiSpecIssue(
             f"{host=} has not been added yet to the testing system. Please do so yourself"

services/api-server/tests/mocks/start_job_no_payment.json

@@ -0,0 +1,47 @@
+[
+ {
+  "name": "POST /v2/computations",
+  "description": "<Request('POST', 'http://director-v2:8000/v2/computations')>",
+  "method": "POST",
+  "host": "director-v2",
+  "path": {
+   "path": "/v2/computations",
+   "path_parameters": []
+  },
+  "query": null,
+  "request_payload": {
+   "user_id": 1,
+   "project_id": "6e52228c-6edd-4505-9131-e901fdad5b17",
+   "start_pipeline": true,
+   "product_name": "osparc",
+   "use_on_demand_clusters": false
+  },
+  "response_body": {
+   "id": "6e52228c-6edd-4505-9131-e901fdad5b17",
+   "state": "PUBLISHED",
+   "result": null,
+   "pipeline_details": {
+    "adjacency_list": {
+     "97792753-cbda-56a7-b8c0-576f990b21c0": []
+    },
+    "progress": 0.0,
+    "node_states": {
+     "97792753-cbda-56a7-b8c0-576f990b21c0": {
+      "modified": true,
+      "dependencies": [],
+      "currentStatus": "PUBLISHED",
+      "progress": null
+     }
+    }
+   },
+   "iteration": 9,
+   "cluster_id": 0,
+   "started": null,
+   "stopped": "2023-10-19T12:59:28.719336+00:00",
+   "submitted": "2023-10-19T12:59:50.999473+00:00",
+   "url": "http://director-v2:8000/v2/computations/6e52228c-6edd-4505-9131-e901fdad5b17?user_id=1",
+   "stop_url": "http://director-v2:8000/v2/computations/6e52228c-6edd-4505-9131-e901fdad5b17:stop?user_id=1"
+  },
+  "status_code": 201
+ }
+]