Source code for pywps.response.basic

from abc import abstractmethod
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from pywps import WPSRequest

import os

from jinja2 import Environment, PackageLoader

from pywps.dblog import store_status
from pywps.translations import get_translation

from . import RelEnvironment
from .status import WPS_STATUS


[docs] class WPSResponse(object): def __init__(self, wps_request: 'WPSRequest', uuid=None, version="1.0.0"): self.wps_request = wps_request self.uuid = uuid self.message = '' self.status = WPS_STATUS.ACCEPTED self.status_percentage = 0 self.doc = None self.content_type = None self.version = version self.template_env = RelEnvironment( loader=PackageLoader('pywps', 'templates'), trim_blocks=True, lstrip_blocks=True, autoescape=True, ) self.template_env.globals.update(get_translation=get_translation) def _update_status(self, status, message, status_percentage): """ Update status report of currently running process instance :param str message: Message you need to share with the client :param int status_percentage: Percent done (number between <0-100>) :param pywps.response.status.WPS_STATUS status: process status - user should usually omit this parameter """ self.message = message self.status = status self.status_percentage = status_percentage store_status(self.uuid, self.status, self.message, self.status_percentage) @abstractmethod def _construct_doc(self): ... def get_response_doc(self): try: self.doc, self.content_type = self._construct_doc() except Exception as e: if hasattr(e, "description"): msg = e.description else: msg = e self._update_status(WPS_STATUS.FAILED, msg, 100) raise e else: self._update_status(WPS_STATUS.SUCCEEDED, "Response generated", 100) return self.doc, self.content_type