Merge branch 'main' into COPDS-2006-format-api-request

Author: mcucchi9

cads_processing_api_service/auth.py

@@ -22,7 +22,7 @@
 import fastapi
 import requests
 
-from . import config, costing, exceptions
+from . import config, costing, exceptions, models
 
 VERIFICATION_ENDPOINT = {
     "PRIVATE-TOKEN": "/account/verification/pat",
@@ -183,17 +183,25 @@ def get_accepted_licences(auth_header: tuple[str, str]) -> set[tuple[str, int]]:
     return accepted_licences
 
 
-def check_licences(
-    required_licences: set[tuple[str, int]], accepted_licences: set[tuple[str, int]]
+def verify_licences(
+    accepted_licences: set[tuple[str, int]] | list[tuple[str, int]],
+    required_licences: set[tuple[str, int]] | list[tuple[str, int]],
+    api_request_url: str,
+    process_id: str,
 ) -> set[tuple[str, int]]:
-    """Check if accepted licences satisfy required ones.
+    """
+    Verify if all the licences required for the process submission have been accepted.
 
     Parameters
     ----------
-    required_licences : set[tuple[str, int]]
-        Required licences.
-    accepted_licences : set[tuple[str, int]]
-        Accepted licences.
+    accepted_licences : set[tuple[str, int]] | list[tuple[str, int]],
+        Licences accepted by a user stored in the Extended Profiles database.
+    required_licences : set[tuple[str, int]] | list[tuple[str, int]],
+        Licences bound to the required process/dataset.
+    api_request_url : str
+        API request URL, required to generate the URL to the dataset licences page.
+    process_id : str
+        Process identifier, required to generate the URL to the dataset licences page.
 
     Returns
     -------
@@ -205,39 +213,30 @@ def check_licences(
     exceptions.PermissionDenied
         Raised if not all required licences have been accepted.
     """
+    if not isinstance(accepted_licences, set):
+        accepted_licences = set(accepted_licences)
+    if not isinstance(required_licences, set):
+        required_licences = set(required_licences)
     missing_licences = required_licences - accepted_licences
     if not len(missing_licences) == 0:
-        missing_licences_detail = [
-            {"id": licence[0], "revision": licence[1]} for licence in missing_licences
-        ]
+        missing_licences_message_template = (
+            config.ensure_settings().missing_licences_message
+        )
+        dataset_licences_url_template = config.ensure_settings().dataset_licences_url
+        parsed_api_request_url = urllib.parse.urlparse(api_request_url)
+        base_url = f"{parsed_api_request_url.scheme}://{parsed_api_request_url.netloc}"
+        dataset_licences_url = dataset_licences_url_template.format(
+            base_url=base_url, process_id=process_id
+        )
+        missing_licences_message = missing_licences_message_template.format(
+            dataset_licences_url=dataset_licences_url
+        )
         raise exceptions.PermissionDenied(
-            title="required licences not accepted",
-            detail=(
-                "required licences not accepted; "
-                "please accept the following licences to proceed: "
-                f"{missing_licences_detail}"
-            ),
+            title="required licences not accepted", detail=missing_licences_message
         )
     return missing_licences
 
 
-def validate_licences(
-    accepted_licences: set[tuple[str, str]],
-    licences: list[tuple[str, int]],
-) -> None:
-    """Validate process execution request's payload in terms of required licences.
-
-    Parameters
-    ----------
-    stored_accepted_licences : set[tuple[str, str]]
-        Licences accepted by a user stored in the Extended Profiles database.
-    licences : list[tuple[str, int]]
-        Licences bound to the required process/dataset.
-    """
-    required_licences = set(licences)
-    check_licences(required_licences, accepted_licences)  # type: ignore
-
-
 def verify_if_disabled(disabled_reason: str | None, user_role: str | None) -> None:
     """Verify if a dataset's disabling reason grant access to the dataset for a specific user role.
 
@@ -262,7 +261,7 @@ def verify_if_disabled(disabled_reason: str | None, user_role: str | None) -> No
 
 
 def verify_cost(
-    request: dict[str, Any], adaptor_properties: dict[str, Any]
+    request: dict[str, Any], adaptor_properties: dict[str, Any], request_origin: str
 ) -> dict[str, float] | None:
     """Verify if the cost of a process execution request is within the allowed limits.
 
@@ -272,6 +271,8 @@ def verify_cost(
         Process execution request.
     adaptor_properties : dict[str, Any]
         Adaptor properties.
+    request_origin : str
+        Origin of the request. Can be either "api" or "ui".
 
     Raises
     ------
@@ -283,9 +284,13 @@ def verify_cost(
     dict[str, float] | None
         Request costs.
     """
-    costing_info = costing.compute_costing(request, adaptor_properties)
-    max_costs_exceeded = costing_info.max_costs_exceeded
-    if max_costs_exceeded:
+    costing_info: models.CostingInfo = costing.compute_costing(
+        request, adaptor_properties, request_origin
+    )
+    highest_cost: models.RequestCost = costing.compute_highest_cost_limit_ratio(
+        costing_info
+    )
+    if highest_cost.cost > highest_cost.limit:
         raise exceptions.PermissionDenied(
             title="cost limits exceeded",
             detail="Your request is too large, please reduce your selection.",

cads_processing_api_service/clients.py

@@ -70,6 +70,16 @@ class DatabaseClient(ogc_api_processes_fastapi.clients.BaseClient):
     process_data_table = cads_catalogue.database.ResourceData
     job_table = cads_broker.database.SystemRequest
 
+    endpoints_description = {
+        "GetProcesses": "Get the list of available processes' summaries.",
+        "GetProcess": "Get the description of the process identified by `process_id`.",
+        "PostProcessExecution": "Request execution of a process.",
+        "GetJobs": "Get the list of submitted jobs, alongside information on their status.",
+        "GetJob": "Get status information for the job identifed by `job_id`.",
+        "GetJobResults": "Get results for the job identifed by `job_id`.",
+        "DeleteJob": "Dismiss the job identifed by `job_id`.",
+    }
+
     def get_processes(
         self,
         limit: int | None = fastapi.Query(10, ge=1, le=10000),
@@ -191,6 +201,8 @@ def post_process_execution(
 
         Parameters
         ----------
+        api_request: fastapi.Request
+            API Request object.
         process_id : str
             Process identifier.
         execution_content : models.Execute
@@ -204,6 +216,7 @@ def post_process_execution(
             Submitted job's status information.
         """
         user_uid, user_role = auth.authenticate_user(auth_header, portal_header)
+        request_origin = auth.REQUEST_ORIGIN[auth_header[0]]
         structlog.contextvars.bind_contextvars(user_uid=user_uid)
         request = execution_content.model_dump()
         catalogue_sessionmaker = db_utils.get_catalogue_sessionmaker(
@@ -231,16 +244,22 @@ def post_process_execution(
                 cads_adaptors.exceptions.InvalidRequest,
             ) as exc:
                 raise exceptions.InvalidRequest(detail=str(exc)) from exc
-        costs = auth.verify_cost(request_inputs, adaptor_properties)
-        licences = adaptor.get_licences(request_inputs)
+        costs = auth.verify_cost(request_inputs, adaptor_properties, request_origin)
+        required_licences = adaptor.get_licences(request_inputs)
         if user_uid != "anonymous":
             accepted_licences = auth.get_accepted_licences(auth_header)
-            auth.validate_licences(accepted_licences, licences)
+            api_request_url = str(api_request.url)
+            _ = auth.verify_licences(
+                accepted_licences, required_licences, api_request_url, process_id
+            )
             job_message = None
         else:
             job_message = config.ensure_settings().anonymous_licences_message.format(
                 licences="; ".join(
-                    [f"{licence[0]} (rev: {licence[1]})" for licence in licences]
+                    [
+                        f"{licence[0]} (rev: {licence[1]})"
+                        for licence in required_licences
+                    ]
                 )
             )
         job_id = str(uuid.uuid4())
@@ -255,7 +274,7 @@ def post_process_execution(
             job = cads_broker.database.create_request(
                 session=compute_session,
                 request_uid=job_id,
-                origin=auth.REQUEST_ORIGIN[auth_header[0]],
+                origin=request_origin,
                 user_uid=user_uid,
                 process_id=process_id,
                 portal=dataset.portal,

cads_processing_api_service/config.py

@@ -51,6 +51,14 @@
     "If you are using cdsapi, please upgrade to the latest version."
 )
 
+
+MISSING_LICENCES_MESSAGE = (
+    "Not all the required licences have been accepted; "
+    "please visit {dataset_licences_url} "
+    "to accept the required licence(s)."
+)
+
+
 general_settings = None
 
 
@@ -76,6 +84,10 @@ class Settings(pydantic_settings.BaseSettings):
     missing_dataset_title: str = "Dataset not available"
     anonymous_licences_message: str = ANONYMOUS_LICENCES_MESSAGE
     deprecation_warning_message: str = DEPRECATION_WARNING_MESSAGE
+    missing_licences_message: str = MISSING_LICENCES_MESSAGE
+    dataset_licences_url: str = (
+        "{base_url}/datasets/{process_id}?tab=download#manage-licences"
+    )
 
     download_nodes_config: str = "/etc/retrieve-api/download-nodes.config"
 

cads_processing_api_service/costing.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License
 
+import enum
 from typing import Any
 
 import cads_adaptors
@@ -22,11 +23,34 @@
 
 from . import adaptors, costing, db_utils, models, utils
 
+COST_THRESHOLDS = {"api": "max_costs", "ui": "max_costs_portal"}
 
-def estimate_costs(
+
+class RequestOrigin(str, enum.Enum):
+    api: str = "api"
+    ui: str = "ui"
+
+
+def estimate_cost(
     process_id: str = fastapi.Path(...),
+    request_origin: RequestOrigin = fastapi.Query("api"),
     execution_content: models.Execute = fastapi.Body(...),
-) -> models.Costing:
+) -> models.RequestCost:
+    """
+    Estimate the cost with the highest cost/limit ratio of the request.
+
+    Parameters
+    ----------
+    process_id : str
+        Process ID.
+    execution_content : models.Execute
+        Request content.
+
+    Returns
+    -------
+    models.RequestCost
+        Info on the cost with the highest cost/limit ratio.
+    """
     request = execution_content.model_dump()
     table = cads_catalogue.database.Resource
     catalogue_sessionmaker = db_utils.get_catalogue_sessionmaker(
@@ -38,28 +62,71 @@ def estimate_costs(
         )
     adaptor_properties = adaptors.get_adaptor_properties(dataset)
     costing_info = costing.compute_costing(
-        request.get("inputs", {}), adaptor_properties
+        request.get("inputs", {}), adaptor_properties, request_origin
     )
-    return costing_info
+    cost = costing.compute_highest_cost_limit_ratio(costing_info)
+    return cost
+
+
+def compute_highest_cost_limit_ratio(
+    costing_info: models.CostingInfo,
+) -> models.RequestCost:
+    """
+    Compute the highest cost/limit ratio of the request.
+
+    Parameters
+    ----------
+    costing_info : models.CostingInfo
+        Costs of the request.
+
+    Returns
+    -------
+    models.RequestCost
+        Info on the cost with the highest cost/limit ratio.
+    """
+    costs = costing_info.costs
+    limits = costing_info.limits
+    highest_cost_limit_ratio = 0.0
+    highest_cost = models.RequestCost()
+    for limit_id, limit in limits.items():
+        cost = costs.get(limit_id, 0.0)
+        cost_limit_ratio = cost / limit if limit > 0 else 1.1
+        if cost_limit_ratio > highest_cost_limit_ratio:
+            highest_cost_limit_ratio = cost_limit_ratio
+            highest_cost = models.RequestCost(cost=cost, limit=limit, id=limit_id)
+    return highest_cost
 
 
 def compute_costing(
     request: dict[str, Any],
     adaptor_properties: dict[str, Any],
-) -> models.Costing:
+    request_origin: str,
+) -> models.CostingInfo:
+    """
+    Compute the costs of the request.
+
+    Parameters
+    ----------
+    request : dict[str, Any]
+        Request to be processed.
+    adaptor_properties : dict[str, Any]
+        Adaptor properties.
+
+    Returns
+    -------
+    models.CostingInfo
+        Costs of the request.
+    """
     adaptor: cads_adaptors.AbstractAdaptor = adaptors.instantiate_adaptor(
         adaptor_properties=adaptor_properties
     )
-    costs: dict[str, float] = adaptor.estimate_costs(request=request)
-    costing_config: dict[str, Any] = adaptor_properties["config"].get("costing", {})
-    max_costs: dict[str, Any] = costing_config.get("max_costs", {})
-    max_costs_exceeded = {}
-    for max_cost_id, max_cost_value in max_costs.items():
-        max_cost_value = float(max_cost_value)
-        if max_cost_id in costs.keys():
-            if costs[max_cost_id] > max_cost_value:
-                max_costs_exceeded[max_cost_id] = max_cost_value
-    costing_info = models.Costing(
-        costs=costs, max_costs=max_costs, max_costs_exceeded=max_costs_exceeded
+    if request_origin not in COST_THRESHOLDS:
+        raise ValueError(f"Invalid request origin: {request_origin}")
+    cost_threshold = COST_THRESHOLDS[request_origin]
+    costs: dict[str, float] = adaptor.estimate_costs(
+        request=request, cost_threshold=cost_threshold
     )
+    costing_config: dict[str, Any] = adaptor_properties["config"].get("costing", {})
+    limits: dict[str, Any] = costing_config.get("max_costs", {})
+    costing_info = models.CostingInfo(costs=costs, limits=limits)
     return costing_info

cads_processing_api_service/main.py

@@ -65,16 +65,19 @@ async def lifespan(application: fastapi.FastAPI) -> AsyncGenerator[Any, None]:
 app.router.add_api_route(
     "/processes/{process_id}/constraints",
     constraints.apply_constraints,
+    description="Apply constraints to the submitted process execution.",
     methods=["POST"],
 )
 app.router.add_api_route(
     "/processes/{process_id}/costing",
-    costing.estimate_costs,
+    costing.estimate_cost,
+    description="Estimate costs of the submitted process execution.",
     methods=["POST"],
 )
 app.router.add_api_route(
     "/processes/{process_id}/api-request",
     translators.get_api_request,
+    description="Get API request equivalent to the submitted prrocess execution json.",
     methods=["POST"],
 )