Coverage for pygeoapi/process/manager/dummy.py: 0%

1# ================================================================= 

2# 

3# Authors: Tom Kralidis <tomkralidis@gmail.com> 

4# 

5# Copyright (c) 2024 Tom Kralidis 

6# 

7# Permission is hereby granted, free of charge, to any person 

8# obtaining a copy of this software and associated documentation 

9# files (the "Software"), to deal in the Software without 

10# restriction, including without limitation the rights to use, 

11# copy, modify, merge, publish, distribute, sublicense, and/or sell 

12# copies of the Software, and to permit persons to whom the 

13# Software is furnished to do so, subject to the following 

14# conditions: 

15# 

16# The above copyright notice and this permission notice shall be 

17# included in all copies or substantial portions of the Software. 

18# 

19# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 

20# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 

21# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

22# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT 

23# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 

24# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 

25# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR 

26# OTHER DEALINGS IN THE SOFTWARE. 

27# 

28# ================================================================= 

29 

30import logging 

31from typing import Any, Dict, Optional, Tuple 

32import uuid 

33 

34from pygeoapi.process.manager.base import BaseManager 

35from pygeoapi.util import ( 

36 JobStatus, 

37 RequestedProcessExecutionMode, 

38 RequestedResponse, 

39 Subscriber 

40) 

41 

42LOGGER = logging.getLogger(__name__) 

43 

44 

45class DummyManager(BaseManager): 

46 """generic Manager ABC""" 

47 

48 def __init__(self, manager_def: dict): 

49 """ 

50 Initialize object 

51 

52 :param manager_def: manager definition 

53 

54 :returns: `pygeoapi.process.manager.base.BaseManager` 

55 """ 

56 

57 super().__init__(manager_def) 

58 

59 def get_jobs(self, status: JobStatus = None, limit=None, offset=None 

60 ) -> dict: 

61 """ 

62 Get process jobs, optionally filtered by status 

63 

64 :param status: job status (accepted, running, successful, 

65 failed, results) (default is all) 

66 :param limit: number of jobs to return 

67 :param offset: pagination offset 

68 

69 :returns: dict of list of jobs (identifier, status, process identifier) 

70 and numberMatched 

71 """ 

72 

73 return {'jobs': [], 'numberMatched': 0} 

74 

75 def execute_process( 

76 self, 

77 process_id: str, 

78 data_dict: dict, 

79 execution_mode: Optional[RequestedProcessExecutionMode] = None, 

80 requested_outputs: Optional[dict] = None, 

81 subscriber: Optional[Subscriber] = None, 

82 requested_response: Optional[RequestedResponse] = RequestedResponse.raw.value # noqa 

83 ) -> Tuple[str, str, Any, JobStatus, Optional[Dict[str, str]]]: 

84 """ 

85 Default process execution handler 

86 

87 :param process_id: process identifier 

88 :param data_dict: `dict` of data parameters 

89 :param execution_mode: requested execution mode 

90 :param requested_outputs: `dict` optionally specify the subset of 

91 required outputs - defaults to all outputs. 

92 The value of any key may be an object and include the property 

93 `transmissionMode` - defaults to `value`. 

94 Note: 'optional' is for backward compatibility. 

95 :param subscriber: `Subscriber` optionally specifying callback urls 

96 :param requested_response: `RequestedResponse` optionally specifying 

97 raw or document (default is `raw`) 

98 

99 :raises UnknownProcessError: if the input process_id does not 

100 correspond to a known process 

101 :returns: tuple of job_id, MIME type, response payload, status and 

102 optionally additional HTTP headers to include in the final 

103 response 

104 """ 

105 

106 jfmt = 'application/json' 

107 

108 response_headers = None 

109 if execution_mode is not None: 

110 response_headers = { 

111 'Preference-Applied': RequestedProcessExecutionMode.wait.value} 

112 if execution_mode == RequestedProcessExecutionMode.respond_async: 

113 LOGGER.debug('Dummy manager does not support asynchronous') 

114 LOGGER.debug('Forcing synchronous execution') 

115 

116 self._send_in_progress_notification(subscriber) 

117 processor = self.get_processor(process_id) 

118 try: 

119 jfmt, outputs = processor.execute( 

120 data_dict, outputs=requested_outputs) 

121 current_status = JobStatus.successful 

122 self._send_success_notification(subscriber, outputs) 

123 except Exception as err: 

124 outputs = { 

125 'code': 'InvalidParameterValue', 

126 'description': f'Error executing process: {err}' 

127 } 

128 current_status = JobStatus.failed 

129 LOGGER.exception(err) 

130 self._send_failed_notification(subscriber) 

131 

132 if requested_response == RequestedResponse.document.value: 

133 outputs = { 

134 'outputs': [outputs] 

135 } 

136 

137 job_id = str(uuid.uuid1()) 

138 return job_id, jfmt, outputs, current_status, response_headers 

139 

140 def __repr__(self): 

141 return f'<DummyManager> {self.name}'