Source code for weaver.wps_restapi.providers.providers

import logging
import warnings

from owslib.wps import WebProcessingService
from pyramid.httpexceptions import HTTPCreated, HTTPNoContent, HTTPNotFound, HTTPOk

from utils import get_cookie_headers
from weaver.database import get_db
from weaver.datatype import Service
from weaver.exceptions import ServiceNotFound, log_unhandled_exceptions
from weaver.owsexceptions import OWSMissingParameterValue, OWSNotImplemented
from weaver.processes.types import PROCESS_WPS
from weaver.store.base import StoreServices
from weaver.utils import get_any_id, get_settings
from weaver.warning import NonBreakingExceptionWarning
from weaver.wps_restapi import swagger_definitions as sd
from weaver.wps_restapi.utils import OUTPUT_FORMAT_JSON, get_wps_restapi_base_url

[docs]LOGGER = logging.getLogger(__name__)
@sd.providers_service.get(tags=[sd.TAG_PROVIDERS], renderer=OUTPUT_FORMAT_JSON, schema=sd.GetProviders(), response_schemas=sd.get_providers_list_responses) @log_unhandled_exceptions(logger=LOGGER, message=sd.InternalServerErrorGetProvidersListResponse.description)
[docs]def get_providers(request): """ Lists registered providers. """ store = get_db(request).get_store(StoreServices) providers = [] for service in store.list_services(request=request): try: if service.type.lower() != "wps": continue wps = WebProcessingService(url=service.url, headers=get_cookie_headers(request.headers)) providers.append(dict( id=service.name, title=getattr(wps.identification, "title", ""), abstract=getattr(wps.identification, "abstract", ""), url="{base_url}/providers/{provider_id}".format( base_url=get_wps_restapi_base_url(get_settings(request)), provider_id=service.name), public=service.public)) except Exception as exc: warnings.warn("Exception occurred while fetching wps {0} : {1!r}".format(service.url, exc), NonBreakingExceptionWarning) return HTTPOk(json=providers)
[docs]def get_capabilities(service, request): """ GetCapabilities of a wps provider. """ wps = WebProcessingService(url=service.url, headers=get_cookie_headers(request.headers)) settings = get_settings(request) return dict( id=service.name, title=wps.identification.title, abstract=wps.identification.abstract, url="{base_url}/providers/{provider_id}".format( base_url=get_wps_restapi_base_url(settings), provider_id=service.name), processes="{base_url}/providers/{provider_id}/processes".format( base_url=get_wps_restapi_base_url(settings), provider_id=service.name), type=PROCESS_WPS, contact=wps.provider.contact.name)
[docs]def get_service(request): """ Get the request service using provider_id from the service store. """ store = get_db(request).get_store(StoreServices) provider_id = request.matchdict.get("provider_id") try: service = store.fetch_by_name(provider_id, request=request) except ServiceNotFound: raise HTTPNotFound("Provider {0} cannot be found.".format(provider_id)) return service, store
@sd.providers_service.post(tags=[sd.TAG_PROVIDERS], renderer=OUTPUT_FORMAT_JSON, schema=sd.PostProvider(), response_schemas=sd.post_provider_responses) @log_unhandled_exceptions(logger=LOGGER, message=sd.InternalServerErrorPostProviderResponse.description)
[docs]def add_provider(request): """ Add a provider. """ store = get_db(request).get_store(StoreServices) try: new_service = Service(url=request.json["url"], name=get_any_id(request.json)) except KeyError as exc: raise OWSMissingParameterValue("Missing json parameter '{!s}'.".format(exc), value=exc) if "public" in request.json: new_service["public"] = request.json["public"] if "auth" in request.json: new_service["auth"] = request.json["auth"] try: store.save_service(new_service, request=request) except NotImplementedError: raise OWSNotImplemented(sd.NotImplementedPostProviderResponse.description, value=new_service) return HTTPCreated(json=get_capabilities(new_service, request))
@sd.provider_service.delete(tags=[sd.TAG_PROVIDERS], renderer=OUTPUT_FORMAT_JSON, schema=sd.ProviderEndpoint(), response_schemas=sd.delete_provider_responses) @log_unhandled_exceptions(logger=LOGGER, message=sd.InternalServerErrorDeleteProviderResponse.description)
[docs]def remove_provider(request): """ Remove a provider. """ service, store = get_service(request) try: store.delete_service(service.name, request=request) except NotImplementedError: raise OWSNotImplemented(sd.NotImplementedDeleteProviderResponse.description) return HTTPNoContent(json={})
@sd.provider_service.get(tags=[sd.TAG_PROVIDERS], renderer=OUTPUT_FORMAT_JSON, schema=sd.ProviderEndpoint(), response_schemas=sd.get_provider_responses) @log_unhandled_exceptions(logger=LOGGER, message=sd.InternalServerErrorGetProviderCapabilitiesResponse.description)
[docs]def get_provider(request): """ Get a provider definition (GetCapabilities). """ service, _ = get_service(request) return HTTPOk(json=get_capabilities(service, request))