14440395301

Committed 14 Apr 2025 07:55AM UTC coverage: 75.452% (-1.2%) from 76.67%

Build # 14440395301

Build Type

Pull #221

github

Committed by

web-flow

Commit Message

Merge 39d5ed8b5 into 89c260aaf

Pull Request Pull Request #221: Pixel_based_queries_rewrite

Run Details

2837 of 3760 relevant lines covered (75.45%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.81

geoengine/error.py

'''
Package errors and backend mapped error types
'''

from typing import Any, Dict, Union
import json
import xml.etree.ElementTree as ET
from requests import Response, HTTPError
import geoengine_openapi_client


class GeoEngineException(Exception):
    '''
    Base class for exceptions from the backend
    '''

    error: str
    message: str

    def __init__(self, response: Union[geoengine_openapi_client.ApiException, Dict[str, str]]) -> None:
        super().__init__()

        if isinstance(response, geoengine_openapi_client.ApiException):
            obj = json.loads(response.body) if response.body else {'error': 'unknown', 'message': 'unknown'}
        else:
            obj = response

        self.error = obj['error'] if 'error' in obj else '?'
        self.message = obj['message'] if 'message' in obj else '?'

    def __str__(self) -> str:
        return f"{self.error}: {self.message}"


class InputException(Exception):
    '''
    Exception that is thrown on wrong inputs
    '''

    __message: str

    def __init__(self, message: str) -> None:
        super().__init__()

        self.__message = message

    def __str__(self) -> str:
        return f"{self.__message}"


class UninitializedException(Exception):
    '''
    Exception that is thrown when there is no connection to the backend but methods on the backend are called
    '''

    def __str__(self) -> str:
        return "You have to call `initialize` before using other functionality"


class TypeException(Exception):
    '''
    Exception on wrong types of input
    '''

    __message: str

    def __init__(self, message: str) -> None:
        super().__init__()

        self.__message = message

    def __str__(self) -> str:
        return f"{self.__message}"


class ModificationNotOnLayerDbException(Exception):
    '''
    Exception that is when trying to modify layers that are not part of the layerdb
    '''

    __message: str

    def __init__(self, message: str) -> None:
        super().__init__()

        self.__message = message

    def __str__(self) -> str:
        return f"{self.__message}"

# TODO: remove methods and forbid calling methods in the first place


class MethodNotCalledOnRasterException(Exception):
    '''
    Exception for calling a raster method on a, e.g., vector layer
    '''

    def __str__(self) -> str:
        return "Only allowed to call method on raster result"


# TODO: remove methods and forbid calling methods in the first place
class MethodNotCalledOnVectorException(Exception):
    '''
    Exception for calling a vector method on a, e.g., raster layer
    '''

    def __str__(self) -> str:
        return "Only allowed to call method on vector result"


# TODO: remove methods and forbid calling methods in the first place
class MethodNotCalledOnPlotException(Exception):
    '''
    Exception for calling a plot method on a, e.g., vector layer
    '''

    def __str__(self) -> str:
        return "Only allowed to call method on plot result"


class SpatialReferenceMismatchException(Exception):
    '''
    Exception for calling a method on a workflow with a query rectangle that has a different spatial reference
    '''

    def __init__(self, spatial_reference_a: str, spatial_reference_b: str) -> None:
        super().__init__()

        self.__spatial_reference_a = spatial_reference_a
        self.__spatial_reference_b = spatial_reference_b

    def __str__(self) -> str:
        return f"Spatial reference mismatch {self.__spatial_reference_a} != {self.__spatial_reference_b}"


class InvalidUrlException(Exception):
    '''
    Exception for when no valid url is provided
    '''

    def __init__(self, msg: str) -> None:
        super().__init__()

        self.__msg = msg

    def __str__(self) -> str:
        return f"{self.__msg}"


class MissingFieldInResponseException(Exception):
    '''
    Exception for when a field is missing in a response
    '''

    missing_field: str
    response: Any

    def __init__(self, missing_field: str, response: Any) -> None:
        super().__init__()

        self.missing_field = missing_field
        self.response = response

    def __str__(self) -> str:
        return f"Missing field '{self.missing_field}' in response: {self.response}"


def check_response_for_error(response: Response) -> None:
    '''
    Checks a `Response` for an error and raises it if there is one.
    '''

    try:
        response.raise_for_status()

        return  # no error
    except HTTPError as http_error:
        exception = http_error

    # try to parse it as a Geo Engine error
    try:
        response_json = response.json()
    except Exception:  # pylint: disable=broad-except
        pass  # ignore errors, it seemed not to be JSON
    else:
        # if parsing was successful, raise the appropriate exception
        if 'error' in response_json:
            raise GeoEngineException(response_json)

    # raise `HTTPError` if `GeoEngineException` or any other was not thrown
    raise exception


class MethodOnlyAvailableInGeoEnginePro(Exception):
    '''
    Exception when trying to use a method that is only available in Geo Engine Pro
    '''

    __message: str

    def __init__(self, message: str) -> None:
        super().__init__()

        self.__message = message

    def __str__(self) -> str:
        return f"Method is only available in Geo Engine Pro: {self.__message}"


class OGCXMLError(Exception):
    '''
    Exception when an OGC XML error is returned
    '''

    __xml: ET.Element

    def __init__(self, xml: bytearray) -> None:
        super().__init__()

        self.__xml = ET.fromstring(xml[1:])

    def __str__(self) -> str:
        service_exception = ''
        for e in self.__xml:
            if 'ServiceException' in e.tag and e.text is not None:
                service_exception = e.text
                break

        return f'OGC API error: {service_exception}'

    @classmethod
    def is_ogc_error(cls, xml: bytearray) -> bool:
        return xml.startswith(b"\n<?xml")

1	'''
2	Package errors and backend mapped error types
3	'''
4
5	from typing import Any, Dict, Union	1✔
6	import json	1✔
7	import xml.etree.ElementTree as ET	1✔
8	from requests import Response, HTTPError	1✔
9	import geoengine_openapi_client	1✔
10
11
12	class GeoEngineException(Exception):	1✔
13	'''
14	Base class for exceptions from the backend
15	'''
16
17	error: str	1✔
18	message: str	1✔
19
20	def __init__(self, response: Union[geoengine_openapi_client.ApiException, Dict[str, str]]) -> None:	1✔
21	super().__init__()	1✔
22
23	if isinstance(response, geoengine_openapi_client.ApiException):	1✔
24	obj = json.loads(response.body) if response.body else {'error': 'unknown', 'message': 'unknown'}	×
25	else:
26	obj = response	1✔
27
28	self.error = obj['error'] if 'error' in obj else '?'	1✔
29	self.message = obj['message'] if 'message' in obj else '?'	1✔
30
31	def __str__(self) -> str:	1✔
32	return f"{self.error}: {self.message}"	×
33
34
35	class InputException(Exception):	1✔
36	'''
37	Exception that is thrown on wrong inputs
38	'''
39
40	__message: str	1✔
41
42	def __init__(self, message: str) -> None:	1✔
43	super().__init__()	1✔
44
45	self.__message = message	1✔
46
47	def __str__(self) -> str:	1✔
48	return f"{self.__message}"	1✔
49
50
51	class UninitializedException(Exception):	1✔
52	'''
53	Exception that is thrown when there is no connection to the backend but methods on the backend are called
54	'''
55
56	def __str__(self) -> str:	1✔
57	return "You have to call `initialize` before using other functionality"	1✔
58
59
60	class TypeException(Exception):	1✔
61	'''
62	Exception on wrong types of input
63	'''
64
65	__message: str	1✔
66
67	def __init__(self, message: str) -> None:	1✔
68	super().__init__()	×
69
70	self.__message = message	×
71
72	def __str__(self) -> str:	1✔
73	return f"{self.__message}"	×
74
75
76	class ModificationNotOnLayerDbException(Exception):	1✔
77	'''
78	Exception that is when trying to modify layers that are not part of the layerdb
79	'''
80
81	__message: str	1✔
82
83	def __init__(self, message: str) -> None:	1✔
84	super().__init__()	×
85
86	self.__message = message	×
87
88	def __str__(self) -> str:	1✔
89	return f"{self.__message}"	×
90
91	# TODO: remove methods and forbid calling methods in the first place
92
93
94	class MethodNotCalledOnRasterException(Exception):	1✔
95	'''
96	Exception for calling a raster method on a, e.g., vector layer
97	'''
98
99	def __str__(self) -> str:	1✔
100	return "Only allowed to call method on raster result"	×
101
102
103	# TODO: remove methods and forbid calling methods in the first place
104	class MethodNotCalledOnVectorException(Exception):	1✔
105	'''
106	Exception for calling a vector method on a, e.g., raster layer
107	'''
108
109	def __str__(self) -> str:	1✔
110	return "Only allowed to call method on vector result"	×
111
112
113	# TODO: remove methods and forbid calling methods in the first place
114	class MethodNotCalledOnPlotException(Exception):	1✔
115	'''
116	Exception for calling a plot method on a, e.g., vector layer
117	'''
118
119	def __str__(self) -> str:	1✔
120	return "Only allowed to call method on plot result"	×
121
122
123	class SpatialReferenceMismatchException(Exception):	1✔
124	'''
125	Exception for calling a method on a workflow with a query rectangle that has a different spatial reference
126	'''
127
128	def __init__(self, spatial_reference_a: str, spatial_reference_b: str) -> None:	1✔
129	super().__init__()	×
130
131	self.__spatial_reference_a = spatial_reference_a	×
132	self.__spatial_reference_b = spatial_reference_b	×
133
134	def __str__(self) -> str:	1✔
135	return f"Spatial reference mismatch {self.__spatial_reference_a} != {self.__spatial_reference_b}"	×
136
137
138	class InvalidUrlException(Exception):	1✔
139	'''
140	Exception for when no valid url is provided
141	'''
142
143	def __init__(self, msg: str) -> None:	1✔
144	super().__init__()	×
145
146	self.__msg = msg	×
147
148	def __str__(self) -> str:	1✔
149	return f"{self.__msg}"	×
150
151
152	class MissingFieldInResponseException(Exception):	1✔
153	'''
154	Exception for when a field is missing in a response
155	'''
156
157	missing_field: str	1✔
158	response: Any	1✔
159
160	def __init__(self, missing_field: str, response: Any) -> None:	1✔
161	super().__init__()	×
162
163	self.missing_field = missing_field	×
164	self.response = response	×
165
166	def __str__(self) -> str:	1✔
167	return f"Missing field '{self.missing_field}' in response: {self.response}"	×
168
169
170	def check_response_for_error(response: Response) -> None:	1✔
171	'''
172	Checks a `Response` for an error and raises it if there is one.
173	'''
174
175	try:	×
176	response.raise_for_status()	×
177
178	return # no error	×
179	except HTTPError as http_error:	×
180	exception = http_error	×
181
182	# try to parse it as a Geo Engine error
183	try:	×
184	response_json = response.json()	×
185	except Exception: # pylint: disable=broad-except	×
186	pass # ignore errors, it seemed not to be JSON	×
187	else:
188	# if parsing was successful, raise the appropriate exception
189	if 'error' in response_json:	×
190	raise GeoEngineException(response_json)	×
191
192	# raise `HTTPError` if `GeoEngineException` or any other was not thrown
193	raise exception	×
194
195
196	class MethodOnlyAvailableInGeoEnginePro(Exception):	1✔
197	'''
198	Exception when trying to use a method that is only available in Geo Engine Pro
199	'''
200
201	__message: str	1✔
202
203	def __init__(self, message: str) -> None:	1✔
204	super().__init__()	×
205
206	self.__message = message	×
207
208	def __str__(self) -> str:	1✔
209	return f"Method is only available in Geo Engine Pro: {self.__message}"	×
210
211
212	class OGCXMLError(Exception):	1✔
213	'''
214	Exception when an OGC XML error is returned
215	'''
216
217	__xml: ET.Element	1✔
218
219	def __init__(self, xml: bytearray) -> None:	1✔
220	super().__init__()	1✔
221
222	self.__xml = ET.fromstring(xml[1:])	1✔
223
224	def __str__(self) -> str:	1✔
225	service_exception = ''	×
226	for e in self.__xml:	×
227	if 'ServiceException' in e.tag and e.text is not None:	×
228	service_exception = e.text	×
229	break	×
230
231	return f'OGC API error: {service_exception}'	×
232
233	@classmethod	1✔
234	def is_ogc_error(cls, xml: bytearray) -> bool:	1✔
235	return xml.startswith(b"\n<?xml")	1✔

geo-engine / geoengine-python / 14440395301

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous