weaver.processes.utils ====================== .. py:module:: weaver.processes.utils Module Contents --------------- .. py:data:: LOGGER .. py:data:: UpdateFieldListMethod .. py:function:: get_process(process_id: Optional[str] = None, request: Optional[weaver.typedefs.PyramidRequest] = None, settings: Optional[weaver.typedefs.SettingsType] = None, store: Optional[weaver.store.base.StoreProcesses] = None, revision: bool = True) -> weaver.datatype.Process Obtain the specified process and validate information, returning appropriate HTTP error if invalid. Process identifier must be provided from either the request path definition or literal ID. Database must be retrievable from either the request, underlying settings, or direct store reference. .. versionchanged:: 4.20 Process identifier can also be an 'id:version' tag. Also, the request query parameter 'version' can be used. If using the :paramref:`process_id` explicitly instead of the request, a versioned :term:`Process` reference MUST employ the tagged representation to resolve the appropriate :term:`Process` revision. Different parameter combinations are intended to be used as needed or more appropriate, such that redundant operations can be reduced where some objects are already fetched from previous operations. :param process_id: Explicit :term:`Process` identifier to employ for lookup. :param request: When no explicit ID specified, try to find information from the request. :param settings: Application settings for database connection. Can be guessed from local thread or request object if not given. :param store: Database process store reference. :param revision: When parsing the :term:`Process` ID (either explicit or from request), indicate if any tagged revision specifier should be used or dropped. .. py:function:: map_progress(progress: weaver.typedefs.Number, range_min: weaver.typedefs.Number, range_max: weaver.typedefs.Number) -> weaver.typedefs.Number Calculates the relative progression of the percentage process within min/max values. .. py:function:: get_process_information(process_description: weaver.typedefs.JSON) -> weaver.typedefs.JSON Obtain the details for the process within its description considering various supported formats. .. py:function:: _check_deploy(payload: weaver.typedefs.JSON) -> Union[weaver.typedefs.ProcessDeployment, weaver.typedefs.CWL] Validate minimum deploy payload field requirements with exception handling. .. py:function:: _validate_deploy_process_info(process_info: weaver.typedefs.JSON, reference: Optional[str], package: Optional[weaver.typedefs.CWL], settings: weaver.typedefs.SettingsType, headers: Optional[weaver.typedefs.AnyHeadersContainer]) -> weaver.typedefs.JSON Obtain the process definition from deploy payload with exception handling. .. seealso:: - :func:`weaver.processes.wps_package.get_process_definition` .. py:function:: resolve_cwl_graph(package: weaver.typedefs.CWL) -> weaver.typedefs.CWL .. py:function:: parse_process_deploy_content(request: Optional[weaver.typedefs.AnyRequestType] = None, content: Optional[Union[weaver.typedefs.JSON, str]] = None, content_schema: Optional[colander.SchemaNode] = None, content_type: Optional[weaver.formats.ContentType] = sd.RequestContentTypeHeader.default, content_type_schema: Optional[colander.SchemaNode] = sd.RequestContentTypeHeader) -> Union[weaver.typedefs.JSON, weaver.typedefs.CWL] Load the request content with validation of expected content type and their schema. .. py:function:: deploy_process_from_payload(payload: Union[weaver.typedefs.JSON, str], container: Union[weaver.typedefs.AnySettingsContainer, weaver.typedefs.AnyRequestType], overwrite: Union[bool, weaver.datatype.Process] = False) -> pyramid.httpexceptions.HTTPException Deploy the process after resolution of all references and validation of the parameters from payload definition. Adds a :class:`weaver.datatype.Process` instance to storage using the provided JSON ``payload`` matching :class:`weaver.wps_restapi.swagger_definitions.ProcessDescription`. :param payload: JSON payload that was specified during the process deployment request. :param container: Container to retrieve application settings. If it is a ``request``-like object, additional parameters may be used to identify the payload schema. :param overwrite: In case of a pure deployment (from scratch), indicates (using :class:`bool`) whether to allow override of an existing process definition if conflict occurs. No versioning is applied in this case (full replacement). In case of an update deployment (from previous), indicates which process to be replaced with updated version. The new version should not conflict with another existing process version. If payload doesn't provide a new version, the following `MAJOR` version from the specified overwrite process is used to define the new revision. :returns: HTTPOk if the process registration was successful. :raises HTTPException: for any invalid process deployment step. .. py:function:: _save_deploy_process(process: weaver.datatype.Process, override: bool, container: weaver.typedefs.AnySettingsContainer) -> weaver.typedefs.JSON Store the :class:`Process` to database with error handling and appropriate message reporting the problem. .. py:function:: _update_deploy_process_version(process: weaver.datatype.Process, process_overwrite: weaver.datatype.Process, update_level: weaver.utils.VersionLevel, container: Optional[weaver.typedefs.AnySettingsContainer] = None) -> weaver.typedefs.JSON Handle all necessary update operations of a :term:`Process` definition. Validate that any specified version for :term:`Process` deployment is valid against any other existing versions. Perform any necessary database adjustments to replace the old :term:`Process` references for the creation of the updated :term:`Process` to ensure all versions and links remain valid against their original references. :param process: Desired new process definition. :param process_overwrite: Old process from which update of the definition in database could be required. :param update_level: Minimum semantic version level required for this update operation. If the new :term:`Process` definition did not provide a version explicitly, this level will be used to automatically infer the following revision number based on the old :term:`Process` reference. :param container: Any container to retrieve a database connection. :returns: Process summary with definition retrieved from storage (saved) after all operations were applied. :raises HTTPException: Relevant error is raised in the even of any erroneous process definition (old and new). .. py:function:: _bump_process_version(version: weaver.typedefs.AnyVersion, update_level: weaver.utils.VersionLevel) -> weaver.typedefs.AnyVersion Obtain the relevant version with specified level incremented by one. .. py:function:: _apply_process_metadata(process: weaver.datatype.Process, update_data: weaver.typedefs.JSON) -> weaver.utils.VersionLevel Apply requested changes for update of the :term:`Process`. Assumes that update data was pre-validated with appropriate schema validation to guarantee relevant typings and formats are applied for expected fields. Validation of fields metadata with their specific conditions is accomplished when attempting to apply changes. .. seealso:: Schema :class:`sd.PatchProcessBodySchema` describes specific field handling based on unspecified value, null or empty-list. Corresponding update levels required for fields are also provided in this schema definition. :param process: Process to modify. Can be the latest or a previously tagged version. :param update_data: Fields with updated data to apply to the process. :return: Applicable update level based on updates to be applied. .. py:function:: update_process_metadata(request: weaver.typedefs.AnyRequestType) -> pyramid.httpexceptions.HTTPException Update only MINOR or PATCH level :term:`Process` metadata. Desired new version can be eiter specified explicitly in request payload, or will be guessed accordingly to detected changes to be applied. .. py:function:: parse_wps_process_config(config_entry: Union[weaver.typedefs.JSON, str]) -> Tuple[str, str, List[str], bool] Parses the available WPS provider or process entry to retrieve its relevant information. :return: WPS provider name, WPS service URL, and list of process identifier(s). :raise ValueError: if the entry cannot be parsed correctly. .. py:function:: register_wps_processes_static(service_url: str, service_name: str, service_visibility: bool, service_processes: List[str], container: weaver.typedefs.AnyRegistryContainer) -> None Register WPS-1 :term:`Process` under a service :term:`Provider` as static references. For a given WPS provider endpoint, either iterates over all available processes under it to register them one by one, or limit itself only to those of the reduced set specified by :paramref:`service_processes`. The registered `WPS-1` processes generate a **static** reference, meaning that metadata of each process as well as any other modifications to the real remote reference will not be tracked, including validation of even their actual existence, or modifications to inputs/outputs. The :term:`Application Package` will only point to it assuming it remains valid. Each of the deployed processes using *static* reference will be accessible directly under `Weaver` endpoints:: /processes/_ The service is **NOT** deployed as :term:`Provider` since the processes are registered directly. .. seealso:: - :func:`register_wps_processes_dynamic` :param service_url: WPS-1 service location (where ``GetCapabilities`` and ``DescribeProcess`` requests can be made). :param service_name: Identifier to employ for generating the full process identifier. :param service_visibility: Visibility flag of the provider. :param service_processes: process IDs under the service to be registered, or all if empty. :param container: settings to retrieve required configuration settings. .. py:function:: register_wps_processes_dynamic(service_name: str, service_url: str, service_visibility: bool, container: weaver.typedefs.AnyRegistryContainer) -> None Register a WPS service ``provider`` such that ``processes`` under it are dynamically accessible on demand. The registered `WPS-1` provider generates a **dynamic** reference to processes under it. Only the :term:`Provider` reference itself is actually registered. No :term:`Process` are directly registered following this operation. When information about the offered processes, descriptions of those processes or their execution are requested, `Weaver` will query the referenced :term:`Provider` for details and convert the corresponding :term:`Process` dynamically. This means that latest metadata of the :term:`Process`, and any modification to it on the remote service will be immediately reflected on `Weaver` without any need to re-deploy processes. Each of the deployed processes using *dynamic* reference will be accessible under `Weaver` endpoints:: /providers//processes/ The processes are **NOT** deployed locally since the processes are retrieved from the :term:`Provider` itself. .. seealso:: - :func:`register_wps_processes_static` :param service_url: WPS-1 service location (where ``GetCapabilities`` and ``DescribeProcess`` requests can be made). :param service_name: Identifier to employ for registering the provider identifier. :param service_visibility: Visibility flag of the provider. :param container: settings to retrieve required configuration settings. .. py:function:: register_wps_processes_from_config(container: weaver.typedefs.AnySettingsContainer, wps_processes_file_path: Optional[weaver.typedefs.FileSystemPathType] = None) -> None Registers remote :term:`WPS` providers and/or processes as specified from the configuration file. Loads a ``wps_processes.yml`` file and registers processes under :ref:`proc_wps_12` providers to the current `Weaver` instance as equivalent :term:`OGC API - Processes` instances. References listed under ``processes`` are registered statically (by themselves, unchanging snapshot). References listed under ``providers``, the :term:`WPS` themselves are registered, making each :term:`Process` listed in their ``GetCapabilities`` available. In this case, registered processes are defined dynamically, meaning they will be fetched on the provider each time a request refers to them, keeping their definition up-to-date with the remote server. .. versionadded:: 1.14 When references are specified using ``providers`` section instead of ``processes``, the registration only saves the remote WPS provider endpoint to dynamically populate :term:`WPS` processes on demand. Previous behavior was to register each :term:`WPS` process individually with ID ``[service]_[process]``. .. versionchanged:: 4.19 Parameter position are inverted. If :paramref:`wps_processes_file_path` is explicitly provided, it is used directly without considering settings. Otherwise, automatically employ the definition in setting: ``weaver.wps_processes_file``. .. seealso:: - `weaver.wps_processes.yml.example` for additional file format details. .. note:: Settings with an explicit empty ``weaver.wps_processes_file`` entry will be considered as *nothing to load*. If the entry is omitted, default location :data:`WEAVER_DEFAULT_WPS_PROCESSES_CONFIG` is attempted instead. :param container: Registry container to obtain database reference as well as application settings. :param wps_processes_file_path: Override file path to employ instead of default settings definition. .. py:function:: _check_package_file(cwl_file_path_or_url: str) -> str Validates that the specified :term:`CWL` file path or URL points to an existing and allowed file format. :param cwl_file_path_or_url: one of allowed file types path on disk, or an URL pointing to one served somewhere. :returns: validated absolute path or URL of the file reference. :raises PackageRegistrationError: in case of missing file, invalid format or invalid HTTP status code. .. py:function:: is_cwl_package(package: Any) -> bool Perform minimal validation of a :term:`CWL` package definition. .. py:function:: load_package_file(file_path: str) -> weaver.typedefs.CWL Loads the package in :term:`YAML`/:term:`JSON` format specified by the file path. .. py:function:: register_cwl_processes_from_config(container: weaver.typedefs.AnySettingsContainer) -> int Load multiple :term:`CWL` definitions from a directory to register corresponding :term:`Process`. .. versionadded:: 4.19 Each individual :term:`CWL` definition must fully describe a :term:`Process` by itself. Therefore, an ``id`` must be available in the file to indicate the target deployment reference. In case of conflict, the existing database :term:`Process` will be overridden to ensure file updates are applied. Files are loaded in alphabetical order. If a :term:`Workflow` needs to refer to other processes, they should be named in way that dependencies will be resolvable prior to the registration of the :term:`Workflow` :term:`Process`. The resolved directory to search for :term:`CWL` will be traversed recursively. This, along with the name of :term:`CWL` files themselves, can be used to resolve order-dependent loading cases. Only ``.cwl`` extensions are considered to avoid invalid parsing of other files that could be defined in the shared configuration directory. .. note:: Settings with an explicit empty ``weaver.cwl_processes_dir`` entry will be considered as *nothing to load*. If the entry is omitted, default location :data:`WEAVER_CONFIG_DIR` is used to search for :term:`CWL` files. :param container: Registry container to obtain database reference as well as application settings. :returns: Number of successfully registered processes from found :term:`CWL` files. .. py:function:: pull_docker(docker_auth: weaver.datatype.DockerAuthentication, logger: weaver.utils.LoggerHandler = LOGGER) -> Optional[docker.client.DockerClient] Pulls the referenced Docker image to local cache from an optionally secured registry. If the Docker image is already available locally, simply validates it. Authentication are applied as necessary using the provided parameters. .. warning:: Logging calls must employ the :param docker_auth: Docker reference with optional authentication parameters. :param logger: Alternative logger reference to log status messages about the operation. :returns: Docker client to perform further operations with the retrieved or validated image. None if failed.