Spaces:
Running
Running
File size: 7,710 Bytes
5cd41bc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
# imports
from collections.abc import Callable
from requests import post
from requests.exceptions import HTTPError
from returns.context import ReaderIOResult
from returns.io import IOResult
from returns.methods import unwrap_or_failure
from returns.result import Failure, Result, Success
from returns.unsafe import unsafe_perform_io
from typing import NamedTuple, Protocol, TypeAlias, Union
# Setup
HEADER_TEMPLATE: dict = {
"accept": "application/json",
"X-API-KEY": None,
"Content-Type": "application/json"
}
# Define URL
API_URL: str = "https://whisp.openforis.org/api/submit/geojson"
# type aliases
AnyJSON: TypeAlias = Union[
str,
int,
float,
bool,
None,
dict[str, 'AnyJSON'], # Recursive dict key-value pairs
list['AnyJSON'] # Recursive list of JSON values
] # covers all JSON incl. GeoJSON
ApiResponse: TypeAlias = dict[str, Union[int, dict, AnyJSON]]
# settings interface
class _Options(Protocol):
URL: str
HEADER: dict
class _Settings(NamedTuple):
# implements the _Options interface
URL: str
HEADER: dict
# internal helper functions
def _get_api_response(input: AnyJSON) -> ReaderIOResult[ApiResponse, ApiResponse, _Settings]:
def _post_call(settings: _Options) -> IOResult[ApiResponse, ApiResponse]:
try:
response = post(settings.URL, headers=settings.HEADER, json=input)
status = response.status_code
payload = response.json()
response.raise_for_status()
return IOResult.from_value({'status': status, 'payload': payload})
except HTTPError:
return IOResult.from_failure({'status': status, 'payload': payload})
except Exception as e:
return IOResult.from_failure({'status': 499, 'payload': str(e)})
return ReaderIOResult(_post_call)
# whisp request functions
def safe_whisp_request(input: AnyJSON, api_key: str) -> Result[AnyJSON, AnyJSON]:
"""
Safely sends a request to the Whisp API and returns the parsed JSON response or an error.
This function wraps the API interaction inside result containers for safe error and side effect handling.
It prepares the required headers (including the API key), performs the request using
internal helper functions, and safely returns either the successful response payload
or an error message encapsulated in a `Result` container.
Parameters
----------
input : AnyJSON incl. GeoJSON
The data payload (JSON-serializable) to send in the API request.
api_key : str
The authentication token to be included as the "X-API-KEY" header.
Returns
-------
Result[AnyJSON, AnyJSON]
On success: Returns a Success container containing the JSON response from the API.
On failure: Returns a Failure container containing a dictionary with the error message.
Raises
------
Exception
May raise network, serialization/deserialization, or system-level exceptions
if low-level I/O or unexpected errors occur outside of safe handling.
Examples
--------
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": <...>}]}, "secret-api-key-123")
Success< {'status': 200, 'payload': {"type": "FeatureCollection", "features": [{"type": Feature", "geometry": < ... > }]} >
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": <invalid geometry or CRS>, < ... >}]}, "wrong-key")
Failure< {'status': 499, 'payload': {'error': 'Something went wrong'}} >
"""
_settings = _Settings(API_URL, HEADER_TEMPLATE)
_settings.HEADER.update({"X-API-KEY":api_key})
response = _get_api_response(input)(_settings)
return unsafe_perform_io(response)
def raw_whisp_request(input: AnyJSON, api_key: str) -> AnyJSON:
"""
Sends a request to the Whisp API and returns the result as raw JSON. Raised errors are wrapped in JSON and returned.
This function calls the safe_whisp_request wrapper, but instead of returning
a result container, it unwraps the api call result and returns the raw JSON in case of success and the raised error wrapped in JSON in case of failure.
Use this function when you need a simpler interface which returns JSON as a result rather than explicit error handling.
Parameters
----------
input : AnyJSON incl. GeoJSON
The JSON-serializable payload to be sent in the API request.
api_key : str
The authentication token inserted as the "X-API-KEY" header.
Returns
-------
AnyJSON
The JSON-decoded response payload from the API if the request was successful.
If the request fails, the exception is returned wrapped in JSON.
Raises
------
Exception
Raises the contained error or a generic exception if the API request fails
or if low-level issues occur (for example, network or deserialization errors).
Examples
--------
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": < ... >}]}, "secret-api-key-123")
{'status': 200, 'payload': {"type": "FeatureCollection", "features": [{"type": Feature", "geometry": < ... > }]}
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": <invalid geometry or CRS>, <...>}]}, "wrong-key")
{'status': 499, 'payload': {'error': 'Something went wrong'}}
"""
return unwrap_or_failure(safe_whisp_request(input, api_key))
def whisp_request(input: AnyJSON, api_key: str) -> AnyJSON:
"""
Sends a request to the Whisp API and extracts a list of feature properties from the response.
This function wraps the API request and extracts the 'properties' from each feature
inside the 'features' key of the response payload. On success, it returns a dictionary/JSON
containing a list of these properties. On failure, it returns the error message
from the response, if available.
Parameters
----------
input : AnyJSON incl. GeoJSON
The JSON-serializable payload to send in the API request.
api_key : str
The authentication token included as the "X-API-KEY" header.
Returns
-------
AnyJSON
On success: Returns a dictionary/JSON with a single key 'properties', mapping to a list
of properties found in the API response's features.
On failure: Returns the error message wrapped in a dictionary/JSON as returned by the API.
Raises
------
Exception
May raise generic exceptions in the event of I/O, network, or unexpected errors
at lower levels (for example, in safe_whisp_request or during result matching).
Examples
--------
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": <...>}]}, "secret-api-key-123")
{'properties': [{'plotId': '1', 'external_id': < ... > }
>>> safe_whisp_request({"type": "FeatureCollection", "features": [{"type": Feature", "geometry": <invalid geometry or CRS>, <...>}]}, "wrong-key")
{'error': 'Something went wrong'}
"""
response = safe_whisp_request(input, api_key)
match response:
case Success(value):
return {
'properties': \
[feature.get('properties', {'error': 'Properties not available'}) \
for feature in value.get('payload', {}).get('data', {}).get(
'features', {'error': 'Not any properties available'}
)]
}
case Failure(value):
return value.get('payload', {'error':'Error message not available'}) |