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.request import Request
from pyramid.settings import asbool

from weaver.config import WeaverFeature, get_weaver_configuration
from weaver.database import get_db
from weaver.formats import ContentType
from weaver.store.base 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

if TYPE_CHECKING:
    from typing import Dict, List, Optional, Tuple

    from weaver.datatype import Service
    from weaver.typedefs import JSON

[docs]LOGGER = logging.getLogger(__name__)
[docs]def get_processes_filtered_by_valid_schemas(request): # type: (Request) -> 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)) paging_query = sd.ProcessPagingQuery() paging_value = {param.name: 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, **paging_param) valid_processes = list() invalid_processes_ids = list() for process in processes: try: valid_processes.append(process.summary()) except colander.Invalid as invalid: LOGGER.debug("Invalid process [%s] because:\n%s", process.identifier, invalid) invalid_processes_ids.append(process.identifier) return valid_processes, invalid_processes_ids, paging_param, with_providers, total_local_processes