terra.api.api_responses

  1#  Copyright 2022 Terra Enabling Developers Limited
  2#
  3#  Licensed under the Apache License, Version 2.0 (the "License");
  4#  you may not use this file except in compliance with the License.
  5#  You may obtain a copy of the License at
  6#
  7#      http://www.apache.org/licenses/LICENSE-2.0
  8#
  9#  Unless required by applicable law or agreed to in writing, software
 10#  distributed under the License is distributed on an "AS IS" BASIS,
 11#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12#  See the License for the specific language governing permissions and
 13#  limitations under the License.
 14from __future__ import annotations
 15
 16__all__ = [
 17    "NoDataReturned",
 18    "DataReturned",
 19    "AuthenticationFailed",
 20    "TerraParsedApiResponse",
 21    "TerraWebhookResponse",
 22    "TerraApiResponse",
 23    "GenericMessage",
 24    "GoogleNoDataSourceHookResponse",
 25    "WidgetSession",
 26    "SentToWebhook",
 27    "SubscribedUsers",
 28    "UserDeauthResp",
 29    "AuthHookResponse",
 30    "AccessRevokedHookResponse",
 31    "UserAuthUrl",
 32    "UserDeauthHookResponse",
 33    "UserReauthHookResponse",
 34    "ConnectionDegraded",
 35    "ProvidersResponse",
 36    "RequestProcessingHookResponse",
 37    "ConnectionErrorHookResponse",
 38    "HookResponse",
 39    "RequestCompletedHookResponse",
 40    "UserInfo",
 41]
 42
 43import dataclasses
 44import json.decoder
 45import typing
 46
 47import requests
 48
 49from terra import exceptions
 50from terra import models
 51from terra.models import base_model
 52from terra.models import user as user_
 53
 54if typing.TYPE_CHECKING:
 55    from terra.base_client import Terra
 56
 57# TODO - should use a mixin/trait here instead of a redundant subclass probably
 58class TerraParsedApiResponse(base_model.TerraDataModel):
 59    pass
 60
 61
 62def _parse_api_body(
 63    dtype: typing.Optional[str],
 64    body: typing.Optional[typing.Dict[str, typing.Any]],
 65    user: typing.Optional[models.user.User],
 66) -> TerraParsedApiResponse:
 67
 68    if not body:
 69        raise exceptions.NoBodyException
 70
 71    Auser = user
 72    if "user" in body:
 73        Auser = models.user.User.from_dict(body["user"])
 74
 75    if ("status" in body) and (body["status"] in STATUS.keys()):
 76        response = STATUS[body["status"]]().from_dict(body)
 77
 78    elif dtype in USER_DATATYPES:
 79
 80        if not dtype:
 81            raise exceptions.NoDtypeException
 82
 83        return DataReturned(
 84            user=Auser,
 85            data=[MODEL_MAPPING[dtype]().from_dict(item) for item in body["data"]]
 86            if body.get("data") or body.get("data") == []
 87            else [],
 88            type=dtype,
 89        )
 90    elif dtype in DTYPE_TO_RESPONSE.keys():
 91
 92        if not dtype:
 93            raise exceptions.NoDtypeException
 94
 95        response = DTYPE_TO_RESPONSE[dtype]().from_dict(body)
 96
 97    elif dtype in HOOK_RESPONSE.keys():
 98
 99        if not dtype:
100            raise exceptions.NoDtypeException
101        response = HOOK_RESPONSE[dtype]().from_dict(body)
102
103    else:
104
105        response = GenericMessage().from_dict(body)
106
107    try:
108        setattr(response, "user", Auser)
109
110    finally:
111        try:
112            if "old_user" in body:
113                setattr(
114                    response,
115                    "old_user",
116                    models.user.User.from_dict(body["old_user"]),
117                )
118            if "new_user" in body:
119                setattr(
120                    response,
121                    "new_user",
122                    models.user.User.from_dict(body["new_user"]),
123                )
124        finally:
125            return typing.cast(TerraParsedApiResponse, response)
126
127
128class TerraApiResponse(TerraParsedApiResponse):
129    def __init__(
130        self,
131        resp: requests.Response,
132        dtype: str,
133        user: typing.Optional[user_.User] = None,
134        client: typing.Optional[Terra] = None,
135    ) -> None:
136        self.response_code: int = resp.status_code
137        self.raw_body: typing.Optional[str] = resp.text
138        self.json: typing.Optional[typing.Dict[str, typing.Any]] = None
139        self.dtype: typing.Optional[str] = dtype
140        try:
141            body: typing.Dict[str, typing.Any] = resp.json()
142            self.json = body
143            self.dtype = body.get("type", dtype)
144            self.parsed_response: TerraParsedApiResponse = _parse_api_body(self.dtype, body, user)
145            if client:
146
147                for e in typing.cast(SubscribedUsers, self.parsed_response).users:
148                    e._client = client
149        except json.decoder.JSONDecodeError:
150
151            resp.raise_for_status()
152
153    def get_parsed_response(self) -> TerraParsedApiResponse:
154        """
155        Get the parsed representation of the api response.
156
157        Returns:
158            :obj:`TerraParsedApiResponse`: Parsed representation of the data model.
159        """
160        return self.parsed_response
161
162    def get_json(self) -> typing.Optional[typing.Dict[str, typing.Any]]:
163        """
164        Get the dictionary (json) representation of the api response.
165
166        Returns:
167            :obj:`dict`: Dictionary representation of the api response.
168        """
169        return self.json
170
171
172class TerraWebhookResponse(TerraParsedApiResponse):
173    def __init__(
174        self,
175        resp: typing.Union[typing.Dict[str, typing.Any], requests.Response],
176        user: typing.Optional[user_.User] = None,
177        dtype: typing.Optional[str] = None,
178    ) -> None:
179        self.json: typing.Dict[str, typing.Any] = resp.json() if isinstance(resp, requests.Response) else resp
180        self.dtype: typing.Optional[str] = self.json.get("type", dtype)
181        self.parsed_response: TerraParsedApiResponse = _parse_api_body(self.dtype, self.json, user)
182
183    def get_parsed_response(self) -> TerraParsedApiResponse:
184        """
185        Get the parsed representation of the api response.
186
187        Returns:
188            :obj:`TerraParsedApiResponse`: Parsed representation of the data model.
189        """
190        return self.parsed_response
191
192    def get_json(self) -> typing.Dict[str, typing.Any]:
193        """
194        Get the dictionary (json) representation of the api response.
195
196        Returns:
197            :obj:`dict`: Dictionary representation of the api response.
198        """
199        return self.json
200
201
202@dataclasses.dataclass
203class GenericMessage(TerraParsedApiResponse):
204    message: typing.Optional[str] = dataclasses.field(default=None)
205    status: typing.Optional[str] = dataclasses.field(default=None)
206
207
208@dataclasses.dataclass
209class WidgetSession(TerraParsedApiResponse):
210    expires_in: int = dataclasses.field(default=900)
211    status: typing.Optional[str] = dataclasses.field(default=None)
212    session_id: typing.Optional[str] = dataclasses.field(default=None)
213    url: typing.Optional[str] = dataclasses.field(default=None)
214
215
216@dataclasses.dataclass
217class UserInfo(TerraParsedApiResponse):
218    status: typing.Optional[str] = dataclasses.field(default=None)
219    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
220    is_authenticated: bool = dataclasses.field(default=True)
221
222
223@dataclasses.dataclass
224class UserDeauthResp(TerraParsedApiResponse):
225    status: typing.Optional[str] = dataclasses.field(default="success")
226
227
228@dataclasses.dataclass
229class HookResponse(TerraParsedApiResponse):
230    status: typing.Optional[str] = dataclasses.field(default="success")
231    type: typing.Optional[str] = dataclasses.field(default=None)
232
233
234@dataclasses.dataclass
235class AuthHookResponse(HookResponse):
236    reference_id: typing.Optional[str] = dataclasses.field(default=None)
237    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
238    widget_session_id: typing.Optional[str] = dataclasses.field(default=None)
239
240
241@dataclasses.dataclass
242class UserReauthHookResponse(HookResponse):
243    message: typing.Optional[str] = dataclasses.field(default=None)
244    old_user: typing.Optional[models.user.User] = dataclasses.field(default=None)
245    new_user: typing.Optional[models.user.User] = dataclasses.field(default=None)
246
247
248@dataclasses.dataclass
249class UserDeauthHookResponse(HookResponse):
250    message: typing.Optional[str] = dataclasses.field(default=None)
251    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
252
253
254@dataclasses.dataclass
255class AccessRevokedHookResponse(HookResponse):
256    message: typing.Optional[str] = dataclasses.field(default=None)
257    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
258
259
260@dataclasses.dataclass
261class GoogleNoDataSourceHookResponse(HookResponse):
262    message: typing.Optional[str] = dataclasses.field(default=None)
263    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
264
265
266@dataclasses.dataclass
267class ConnectionErrorHookResponse(HookResponse):
268    message: typing.Optional[str] = dataclasses.field(default=None)
269    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
270
271
272@dataclasses.dataclass
273class RequestProcessingHookResponse(HookResponse):
274    message: typing.Optional[str] = dataclasses.field(default=None)
275    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
276
277
278@dataclasses.dataclass
279class RequestCompletedHookResponse(HookResponse):
280    message: typing.Optional[str] = dataclasses.field(default=None)
281    reference: typing.Optional[str] = dataclasses.field(default=None)
282    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
283
284
285@dataclasses.dataclass
286class SubscribedUsers(TerraParsedApiResponse):
287    users: typing.List[models.user.User] = dataclasses.field(default_factory=list)
288
289
290@dataclasses.dataclass
291class UserAuthUrl(TerraParsedApiResponse):
292    status: typing.Optional[str] = dataclasses.field(default=None)
293    expires_in: int = dataclasses.field(default=900)
294    auth_url: typing.Optional[str] = dataclasses.field(default=None)
295    user_id: typing.Optional[str] = dataclasses.field(default=None)
296
297
298@dataclasses.dataclass
299class NoDataReturned(TerraParsedApiResponse):
300    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
301    status: typing.Optional[str] = dataclasses.field(default="not_available")
302    message: typing.Optional[str] = dataclasses.field(default="Data type requested not available from provider")
303
304
305@dataclasses.dataclass
306class DataReturned(TerraParsedApiResponse):
307    user: typing.Optional[models.user.User] = dataclasses.field(default=None)
308    type: typing.Optional[str] = dataclasses.field(default=None)
309    data: typing.List[TerraParsedApiResponse] = dataclasses.field(default_factory=list)
310
311
312@dataclasses.dataclass
313class AuthenticationFailed(TerraParsedApiResponse):
314    status: str = dataclasses.field(default="error")
315    widget_session_id: typing.Optional[str] = dataclasses.field(default=None)
316    reference_id: typing.Optional[str] = dataclasses.field(default=None)
317    message: typing.Optional[str] = dataclasses.field(default="User failed to authenticate and has been deleted")
318    type: str = dataclasses.field(default="auth")
319    reason: str = dataclasses.field(default="auth_cancelled")
320
321
322@dataclasses.dataclass
323class ConnectionDegraded(TerraParsedApiResponse):
324    status: typing.Optional[str] = dataclasses.field(default="warning")
325    message: typing.Optional[str] = dataclasses.field(default="User connection degraded")
326    type: typing.Optional[str] = dataclasses.field(default="connection_error")
327
328
329@dataclasses.dataclass
330class ProvidersResponse(TerraParsedApiResponse):
331    status: typing.Optional[str] = dataclasses.field(default="warning")
332    providers: typing.Optional[typing.List[str]] = dataclasses.field(default=None)
333    sdk_providers: typing.Optional[typing.List[str]] = dataclasses.field(default=None)
334
335
336@dataclasses.dataclass
337class SentToWebhook(TerraParsedApiResponse):
338    status: typing.Optional[str] = dataclasses.field(default=None)
339    message: typing.Optional[str] = dataclasses.field(default=None)
340
341
342USER_DATATYPES = [
343    "activity",
344    "athlete",
345    "body",
346    "daily",
347    "menstruation",
348    "sleep",
349    "nutrition",
350]
351MODEL_MAPPING = {
352    "activity": models.v2.activity.Activity,
353    "body": models.v2.body.Body,
354    "daily": models.v2.daily.Daily,
355    "sleep": models.v2.sleep.Sleep,
356    "menstruation": models.v2.menstruation.Menstruation,
357    "athlete": models.v2.athlete.Athlete,
358    "nutrition": models.v2.nutrition.Nutrition,
359}
360
361
362DTYPE_TO_RESPONSE = {
363    "widget_session": WidgetSession,
364    "auth_url": UserAuthUrl,
365    "user_info": UserInfo,
366    "subscriptions": SubscribedUsers,
367    "providers": ProvidersResponse,
368    "sent_to_webhook": SentToWebhook,
369}
370
371HOOK_TYPES = {
372    "auth",
373    "user_reauth",
374    "access_revoked",
375    "deauth",
376    "google_no_datasource",
377    "connexion_error",
378    "request_processing",
379    "request_completed",
380}
381
382HOOK_RESPONSE = {
383    "auth": AuthHookResponse,
384    "user_reauth": UserReauthHookResponse,
385    "access_revoked": AccessRevokedHookResponse,
386    "deauth": UserDeauthHookResponse,
387    "google_no_datasource": GoogleNoDataSourceHookResponse,
388    "connexion_error": ConnectionErrorHookResponse,
389    "request_completed": RequestCompletedHookResponse,
390    "request_processing": RequestProcessingHookResponse,
391}
392
393STATUS = {
394    "not_available": NoDataReturned,
395    "warning": ConnectionDegraded,
396    "error": ConnectionDegraded,
397}