spothole/webserver/handlers/api/compatibility/v2_compatibility.py

from webserver.handlers.api.addspot import APISpotHandler
from webserver.handlers.api.alerts import APIAlertsHandler, APIAlertsStreamHandler
from webserver.handlers.api.compatibility.compatibility import (
    CompatibilityWrapper,
    RequestCompatibilityWrapper,
    StreamCompatibilityWrapper,
    rename_keys_and_values,
    rename_query_params,
    translate_json_body,
)
from webserver.handlers.api.dxstats import APIDxStatsHandler
from webserver.handlers.api.lookups import APILookupActivityRefHandler, APILookupCallHandler, APILookupGridHandler
from webserver.handlers.api.options import APIOptionsHandler
from webserver.handlers.api.solar_conditions import APISolarConditionsHandler
from webserver.handlers.api.spots import APISpotsHandler, APISpotsStreamHandler
from webserver.handlers.api.status import APIStatusHandler

# Query parameters renamed in v3
_V2_TO_V3_QUERY_PARAMS = {
    "sig": "activity",
    "needs_sig": "needs_activity",
    "needs_sig_ref": "needs_activity_ref",
}
# Values of query parameters renamed in v3
_V2_TO_V3_QUERY_VALUES = {
    "activity": {"NO_SIG": "NO_ACTIVITY"},
    "fields": {"sig": "activities", "sig_refs": "activity_refs"},
}
# Keys of JSON objects in API responses renamed in v3
_V3_TO_V2_RESPONSE_KEYS = {
    "activity": "sig",
    "activity_refs": "sig_refs",
    "activity_type": "sig_type",
    "activities": "sigs",
    "activity_ref_data_providers": "sig_ref_data_providers",
    "activity_name": "sig_name",
}
# Values of JSON objects in API responses renamed in v3
_V3_TO_V2_RESPONSE_VALUES = {
    "dx_location_source": {"ACTIVITY REF LOOKUP": "SIG REF LOOKUP"},
}


class V2CompatibilityWrapper(CompatibilityWrapper):
    """Translates v2 requests to v3 on the way in, and v3 responses to v2 on the way out. This logic captures the
    majority of changes needed for each endpoint."""

    def translate_request(self):
        rename_query_params(self.request, _V2_TO_V3_QUERY_PARAMS, _V2_TO_V3_QUERY_VALUES)
        super().translate_request()

    def translate_response_object(self, obj):
        obj = super().translate_response_object(obj)
        return rename_keys_and_values(obj, _V3_TO_V2_RESPONSE_KEYS, _V3_TO_V2_RESPONSE_VALUES)


class V2SpotsAlertsCompatibilityWrapper(V2CompatibilityWrapper):
    """Extra translation for spots and alerts. In v3 these have a list of "activities" rather than a single activity,
    so for v2 we collapse this back down to a single value using the first activity in the list. This must happen
    before the generic key renaming, which would otherwise rename "activities" to "sigs" as it does for /options."""

    def translate_response_object(self, obj):
        return super().translate_response_object(collapse_activities(obj))


class V2APISpotsHandler(V2SpotsAlertsCompatibilityWrapper, RequestCompatibilityWrapper, APISpotsHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APISpotsStreamHandler(V2SpotsAlertsCompatibilityWrapper, StreamCompatibilityWrapper, APISpotsStreamHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APIAlertsHandler(V2SpotsAlertsCompatibilityWrapper, RequestCompatibilityWrapper, APIAlertsHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APIAlertsStreamHandler(V2SpotsAlertsCompatibilityWrapper, StreamCompatibilityWrapper, APIAlertsStreamHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APISolarConditionsHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APISolarConditionsHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APIDxStatsHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APIDxStatsHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APIOptionsHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APIOptionsHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APIStatusHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APIStatusHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APILookupCallHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APILookupCallHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APILookupSigRefHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APILookupActivityRefHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APILookupGridHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APILookupGridHandler):
    """No special handling for this, the compatibility wrapper will handle translation to and from the later version."""


class V2APISpotHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APISpotHandler):
    """Some special handling for add spot, because unlike the other calls we have to deal with a request body.
    Because the definition of a spot changed (e.g. sig to activity) we need to apply the same translation to
    spots that are coming in via the add spot call."""

    def translate_request(self):
        def translate(body):
            if isinstance(body, dict) and isinstance(body.get("spot"), dict):
                body["spot"] = self._translate_v2_spot(body["spot"])
            return body

        translate_json_body(self.request, translate)
        super().translate_request()


    def _translate_v2_spot(self, spot_data):
        """Translate a spot provided by a client calling the add spot method in v2 format into v3 format"""

        spot_data = dict(spot_data)
        if "sig" in spot_data:
            sig = spot_data.pop("sig")
            spot_data["activities"] = [sig] if sig else []
        if "sig_refs" in spot_data:
            spot_data["activity_refs"] = spot_data.pop("sig_refs")
        if isinstance(spot_data.get("activity_refs"), list):
            refs = []
            for ref in spot_data["activity_refs"]:
                if isinstance(ref, dict) and "sig" in ref:
                    ref = dict(ref)
                    ref["activity"] = ref.pop("sig")
                refs.append(ref)
            spot_data["activity_refs"] = refs
        return spot_data


def collapse_activities(obj):
    """Utility method to replace the "activities" list in a spot or alert JSON object with a single "activity" value,
    being the first activity in the list, or None if there are none. The object can be a single spot/alert dict, or a
    list of them. Anything else is returned untouched. Used to translate v3's list of activities to the single sig
    expected in v2 API calls."""

    if isinstance(obj, list):
        return [collapse_activities(i) for i in obj]
    if isinstance(obj, dict) and "activities" in obj:
        obj = dict(obj)
        activities = obj.pop("activities")
        obj["activity"] = activities[0] if activities else None
    return obj