Source code for weaver.wps_restapi.utils

import inspect
import logging
from typing import TYPE_CHECKING

from colander import SchemaNode

from weaver.utils import get_settings, get_weaver_url
from weaver.wps_restapi import swagger_definitions as sd

    from typing import Dict

    from weaver.typedefs import AnySettingsContainer

[docs]LOGGER = logging.getLogger(__name__)
[docs]def wps_restapi_base_path(container): # type: (AnySettingsContainer) -> str settings = get_settings(container) restapi_path = settings.get("weaver.wps_restapi_path", "").rstrip("/").strip() return restapi_path
[docs]def get_wps_restapi_base_url(container): # type: (AnySettingsContainer) -> str settings = get_settings(container) weaver_rest_url = settings.get("weaver.wps_restapi_url") if not weaver_rest_url: weaver_url = get_weaver_url(settings) restapi_path = wps_restapi_base_path(settings) weaver_rest_url = weaver_url + restapi_path return weaver_rest_url.rstrip("/").strip()
[docs]def get_schema_ref(schema, container, ref_type="$schema", ref_name=True): # type: (SchemaNode, AnySettingsContainer, str, True) -> Dict[str, str] """ Generates the JSON OpenAPI schema reference relative to the current `Weaver` instance. The provided schema should be one of the items listed in ``#/definitions`` of the ``/json`` endpoint. No validation is accomplished to avoid long processing of all references. If setting ``weaver.schema_url`` is set, this value will be used direct as fully-defined base URL. This could be used to refer to a static endpoint where schemas are hosted. Otherwise, the current Web Application resolved location is employed with JSON OpenAPI path. :param schema: schema-node instance or type for which to generate the OpenAPI reference. :param container: application settings to retrieve the base URL of the schema location. :param ref_type: key employed to form the reference (e.g.: "$schema", "$ref", "@schema", etc.) :param ref_name: indicate if the plain name should also be included under field ``"schema"``. :return: OpenAPI schema reference """ is_instance = isinstance(schema, SchemaNode) assert is_instance or (inspect.isclass(schema) and issubclass(schema, SchemaNode)) if is_instance: schema = type(schema) schema_name = schema.__name__ settings = get_settings(container) weaver_schema_url = settings.get("weaver.schema_url") if not weaver_schema_url: restapi_path = get_wps_restapi_base_url(container) weaver_schema_url = "{}{}#/definitions".format(restapi_path, sd.openapi_json_service.path) weaver_schema_url = weaver_schema_url.rstrip("/").strip() schema_ref = {ref_type: "{}/{}".format(weaver_schema_url, schema_name)} if ref_name: schema_ref.update({"schema": schema_name}) return schema_ref