Source code for weaver.wps_restapi.processes.utils

import logging
import math
from typing import TYPE_CHECKING

import colander
from pyramid.httpexceptions import HTTPBadRequest
from pyramid.settings import asbool

from weaver.compat import InvalidVersion
from weaver.config import WeaverFeature, get_weaver_configuration
from weaver.database import get_db
from weaver.formats import ContentType
from import StoreProcesses
from weaver.utils import get_path_kvp, get_settings, get_weaver_url
from weaver.visibility import Visibility
from weaver.wps_restapi import swagger_definitions as sd

    from typing import Dict, List, Optional, Tuple

    from weaver.datatype import Process, Service
    from weaver.typedefs import JSON, PyramidRequest

[docs]LOGGER = logging.getLogger(__name__)
[docs]def resolve_process_tag(request, process_query=False): # type: (PyramidRequest, bool) -> str """ Obtain the tagged :term:`Process` reference from request path and/or query according to available information. Whether the :term:`Process` is specified by path or query, another ``version`` query can be provided to specify the desired revision by itself. This ``version`` query is considered only if another version indication is not already specified in the :term:`Process` reference using the tagged semantic. When ``process_query = False``, possible combinations are as follows: - ``/processes/{processID}:{version}`` - ``/processes/{processID}?version={version}`` When ``process_query = True``, possible combinations are as follows: - ``/...?process={processID}:{version}`` - ``/...?process={processID}&version={version}`` :param request: Request from which to retrieve the process reference. :param process_query: Whether the process ID reference is located in request path or ``process={id}`` query. """ if process_query: process_id = request.params.get("process") else: process_id = request.matchdict.get("process_id", "") params = sd.LocalProcessQuery().deserialize(request.params) version = params.get("version") if version and ":" not in process_id: # priority to tagged version over query if specified process_id = f"{process_id}:{version}" process_id = sd.ProcessIdentifierTag(name="ProcessID").deserialize(process_id) return process_id
[docs]def get_processes_filtered_by_valid_schemas(request, detail=True): # type: (PyramidRequest, bool) -> Tuple[List[JSON], List[str], Dict[str, Optional[int]], bool, int] """ Validates the processes summary schemas and returns them into valid/invalid lists. :returns: List of valid process and invalid processes IDs for manual cleanup, along with filtering parameters. """ settings = get_settings(request) with_providers = False if get_weaver_configuration(settings) in WeaverFeature.REMOTE: with_providers = asbool(request.params.get("providers", False)) revisions_param = sd.ProcessRevisionsQuery(unknown="ignore").deserialize(request.params) with_revisions = revisions_param.get("revisions") with_links = asbool(request.params.get("links", False)) and detail paging_query = sd.ProcessPagingQuery() paging_value = { param.default for param in paging_query.children} paging_names = set(paging_value) paging_param = paging_query.deserialize(request.params) if with_providers and any(value != paging_value[param] for param, value in paging_param.items()): raise HTTPBadRequest(json={ "description": "Cannot combine paging/sorting parameters with providers full listing query.", "error": "ListingInvalidParameter", "value": list(paging_names.intersection(request.params)) }) store = get_db(request).get_store(StoreProcesses) processes, total_local_processes = store.list_processes( visibility=Visibility.PUBLIC, total=True, **revisions_param, **paging_param ) valid_processes = [] invalid_processes_ids = [] for process in processes: # type: Process try: try: valid_processes.append(process.summary(revision=with_revisions, links=with_links, container=request)) except (InvalidVersion, ValueError) as exc: raise colander.Invalid(sd.ProcessSummary, value=None, msg=str(exc)) except colander.Invalid as invalid: process_ref = process.tag if with_revisions else process.identifier LOGGER.debug("Invalid process [%s] because:\n%s", process_ref, invalid) invalid_processes_ids.append(process.identifier) return valid_processes, invalid_processes_ids, paging_param, with_providers, total_local_processes