# 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": , < ... >}]}, "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": , <...>}]}, "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": , <...>}]}, "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'})