utils/whisp_api_client.py

# 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'})