Coverage for pygeoapi/process/base.py: 93%

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

2# 

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

4# Francesco Martinelli <francesco.martinelli@ingv.it> 

5# 

6# Copyright (c) 2026 Tom Kralidis 

7# Copyright (c) 2024 Francesco Martinelli 

8# 

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

10# obtaining a copy of this software and associated documentation 

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

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

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

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

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

16# conditions: 

17# 

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

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

20# 

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

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

23# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

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

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

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

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

28# OTHER DEALINGS IN THE SOFTWARE. 

29# 

30# ================================================================= 

31 

32import logging 

33from typing import Any, Tuple, Optional 

34 

35from pygeoapi.error import GenericError 

36 

37LOGGER = logging.getLogger(__name__) 

38 

39 

40class BaseProcessor: 

41 """generic Processor ABC. Processes are inherited from this class""" 

42 

43 def __init__(self, processor_def: dict, process_metadata: dict): 

44 """ 

45 Initialize object 

46 

47 :param processor_def: processor definition 

48 :param process_metadata: process metadata `dict` 

49 

50 :returns: pygeoapi.processor.base.BaseProvider 

51 """ 

52 

53 self.name = processor_def['name'] 

54 self.metadata = process_metadata 

55 self.supports_outputs = False 

56 self.allow_internal_requests = processor_def.get( 

57 'allow_internal_requests', False) 

58 

59 def set_job_id(self, job_id: str) -> None: 

60 """ 

61 Set the job_id within the processor 

62 To be implemented by derived classes where required. 

63 

64 :param job_id: the job_id assigned to the request by the Manager. 

65 The function should be called by the Manager upon 

66 assigning the job_id. The job_id is intended to be used 

67 by derived classes, e.g. to write temporary files where 

68 filenames contains the string job_id. 

69 

70 :returns: `None` 

71 """ 

72 

73 pass 

74 

75 def execute(self, data: dict, outputs: Optional[dict] = None 

76 ) -> Tuple[str, Any]: 

77 """ 

78 execute the process 

79 

80 :param data: Dict with the input data that the process needs in order 

81 to execute 

82 :param outputs: `dict` or `list` to optionally specify the subset of 

83 required outputs - defaults to all outputs. 

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

85 property `transmissionMode` - defaults to `value`. 

86 :returns: tuple of MIME type and process response 

87 (string, bytes, list or dict) 

88 """ 

89 

90 raise NotImplementedError() 

91 

92 def __repr__(self): 

93 return f'<BaseProcessor> {self.name}' 

94 

95 

96class ProcessorGenericError(GenericError): 

97 """processor generic error""" 

98 pass 

99 

100 

101class ProcessorExecuteError(ProcessorGenericError): 

102 """query / backend error""" 

103 default_msg = "generic error (check logs)" 

104 

105 

106class JobError(Exception): 

107 pass 

108 

109 

110class JobNotFoundError(JobError): 

111 pass 

112 

113 

114class JobResultNotFoundError(JobError): 

115 pass 

116 

117 

118class ProcessError(Exception): 

119 pass 

120 

121 

122class UnknownProcessError(ProcessError): 

123 pass