terra

 36class Terra:
 37    """
 38    constructor of the Terra class
 39
 40    Args:
 41        api_key (:obj:`str`) : Your API Key
 42        dev_id (:obj:`str`) : Your dev ID
 43        secret (:obj:`str`) : Your terra secret (for web hooks)
 44
 45    """
 46
 47    def __init__(self, api_key: str, dev_id: str, secret: str) -> None:
 48        self.api_key = api_key
 49        self.dev_id = dev_id
 50        self.secret = secret
 51
 52    @property
 53    def _auth_headers(self) -> typing.Dict[str, str]:
 54        """
 55        Internal method used to fill in authentication headers for all requests to the API
 56
 57        Returns:
 58            :obj:`dict`: Dictionary of required auth headers
 59
 60        """
 61        return {"x-api-key": self.api_key, "dev-id": self.dev_id}
 62
 63    def from_user_id(self, user_id: str) -> user_.User:
 64        """
 65        Creates a User instance out of a UUID corresponding to a registered User on the API
 66
 67        Args:
 68            user_id (:obj:`str`): UUID corresponding to a user currently authenticated on the API
 69
 70        Returns:
 71            :obj:`User`: Created User instance
 72
 73        """
 74
 75        user = user_.User(user_id=user_id, client=self)
 76        user.fill_in_user_info()
 77        return user
 78
 79    def _get_arbitrary_data(self, user: user_.User, dtype: str, **kwargs: typing.Any) -> api_responses.TerraApiResponse:
 80        """
 81        Internal method used to retrieve data for a given User
 82
 83        Args:
 84            user (:obj:`models.user.User`):
 85            dtype (:obj:`str`): datatype to be fetched
 86            **kwargs: optional additional parameters for the request
 87
 88        Returns:
 89            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
 90                response object if no error has occured
 91
 92        """
 93        params = {"user_id": user.user_id}
 94        params = utils.update_if_not_none(params, kwargs)
 95
 96        data_resp = requests.get(
 97            f"{constants.BASE_URL}/{dtype}",
 98            params=params,
 99            headers=self._auth_headers,
100        )
101
102        return api_responses.TerraApiResponse(data_resp, user=user, dtype=dtype)
103
104    def get_activity_for_user(
105        self,
106        user: user_.User,
107        start_date: datetime.datetime,
108        end_date: typing.Optional[datetime.datetime] = None,
109        to_webhook: bool = True,
110    ) -> api_responses.TerraApiResponse:
111        """
112        Retrieves workouts/activity data for a given User object. By default, data will be asynchronously sent to registered
113        webhook URL.
114
115        Args:
116            user (:obj:`models.user.User`): User for whom to fetch data
117            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
118            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set, will
119                default to start_date + 24h according to current API specifications
120            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
121
122        Returns:
123            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
124                response object if no error has occured
125
126        """
127        return user.get_activity(
128            start_date=start_date,
129            end_date=end_date if end_date is not None else None,
130            to_webhook=to_webhook,
131        )
132
133    def get_body_for_user(
134        self,
135        user: user_.User,
136        start_date: datetime.datetime,
137        end_date: typing.Optional[datetime.datetime] = None,
138        to_webhook: bool = True,
139    ) -> api_responses.TerraApiResponse:
140        """
141        Retrieves body metrics data for a given User object. By default, data will be asynchronously sent to registered
142        webhook URL.
143
144        Args:
145            user (:obj:`models.user.User`): User for whom to fetch data
146            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
147            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set, will
148                default to start_date + 24h according to current API specifications
149            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
150
151        Returns:
152            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
153                response object if no error has occured
154
155        """
156        return user.get_body(
157            start_date=start_date,
158            end_date=end_date if end_date is not None else None,
159            to_webhook=to_webhook,
160        )
161
162    def get_daily_for_user(
163        self,
164        user: user_.User,
165        start_date: datetime.datetime,
166        end_date: typing.Optional[datetime.datetime] = None,
167        to_webhook: bool = True,
168    ) -> api_responses.TerraApiResponse:
169        """
170        Retrieves daily summary data for a given User object. By default, data will be asynchronously sent to registered
171        webhook URL.
172
173        Args:
174            user (:obj:`models.user.User`): User for whom to fetch data
175            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
176            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set, will
177                default to start_date + 24h according to current API specifications
178            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
179
180        Returns:
181            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
182                response object if no error has occured
183
184        """
185        return user.get_daily(
186            start_date=start_date,
187            end_date=end_date if end_date is not None else None,
188            to_webhook=to_webhook,
189        )
190
191    def get_sleep_for_user(
192        self,
193        user: user_.User,
194        start_date: datetime.datetime,
195        end_date: typing.Optional[datetime.datetime] = None,
196        to_webhook: bool = True,
197    ) -> api_responses.TerraApiResponse:
198        """
199        Retrieves sleep data for a given User object. By default, data will be asynchronously sent to registered
200        webhook URL.
201
202        Args:
203            user (:obj:`models.user.User`): User for whom to fetch data
204            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
205            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set, will
206                default to start_date + 24h according to current API specifications
207            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
208
209        Returns:
210            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
211                response object if no error has occured
212
213        """
214        return user.get_sleep(
215            start_date=start_date,
216            end_date=end_date if end_date is not None else None,
217            to_webhook=to_webhook,
218        )
219
220    def get_athlete_for_user(
221        self,
222        user: user_.User,
223        to_webhook: bool = True,
224    ) -> api_responses.TerraApiResponse:
225        """
226        Retrieves profile info/athlete data for a given User object. By default, data will be asynchronously sent to
227        registered webhook URL.
228
229        Args:
230            user (:obj:`models.user.User`): User for whom to fetch data
231            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
232
233        Returns:
234            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
235                response object if no error has occured
236
237        """
238        return self._get_arbitrary_data(user, "athlete", to_webhook=to_webhook)
239
240    def get_menstruation_for_user(
241        self,
242        user: user_.User,
243        start_date: datetime.datetime,
244        end_date: typing.Optional[datetime.datetime] = None,
245        to_webhook: bool = True,
246    ) -> api_responses.TerraApiResponse:
247        """
248        Retrieves daily summary data for a given User object. By default, data will be asynchronously sent to registered
249        webhook URL.
250
251        Args:
252            user (:obj:`models.user.User`): User for whom to fetch data
253            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
254            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set,
255                will default to start_date + 24h according to current API specifications
256            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
257
258        Returns:
259            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
260                response object if no error has occured
261
262        """
263
264        return user.get_menstruation(
265            start_date=start_date,
266            end_date=end_date if end_date is not None else None,
267            to_webhook=to_webhook,
268        )
269
270    def get_nutrition_for_user(
271        self,
272        user: user_.User,
273        start_date: datetime.datetime,
274        end_date: typing.Optional[datetime.datetime] = None,
275        to_webhook: bool = True,
276    ) -> api_responses.TerraApiResponse:
277        """
278        Retrieves daily summary data for a given User object. By default, data will be asynchronously sent to registered
279        webhook URL.
280
281        Args:
282            user (:obj:`models.user.User`): User for whom to fetch data
283            start_date (:obj:`datetime.datetime`): Datetime object for which to fetch data
284            end_date:obj (:`datetime.datetime`): Optional end_date for which to fetch data - if not set, will
285                default to start_date + 24h according to current API specifications
286            to_webhook (:obj:`bool`): Whether to send data to registered webhook URL or return as a response body
287
288        Returns:
289            :obj:`models.api_responses.TerraApiResponse`: API response object containing DataReturned parsed
290                response object if no error has occured
291
292        """
293
294        return user.get_nutrition(
295            start_date=start_date,
296            end_date=end_date if end_date is not None else None,
297            to_webhook=to_webhook,
298        )
299
300    def generate_widget_session(
301        self,
302        providers: typing.List[str],
303        auth_success_redirect_url: typing.Optional[str] = None,
304        auth_failure_redirect_url: typing.Optional[str] = None,
305        language: typing.Optional[str] = None,
306        reference_id: typing.Optional[str] = None,
307        **kwargs: typing.Any,
308    ) -> api_responses.TerraApiResponse:
309        """
310        Generates a widget session used to allow an end user to authenticate through the API. Users should be
311        redirected to the given URL in order to complete authentication
312
313        Args:
314            providers (List[:obj:`str`]): Providers to display on widget wearable selection screen, by leaving it empty it will use all default providers
315            auth_success_redirect_url (Optional[:obj:`str`]): URL to redirect to upon successful authentication
316            auth_failure_redirect_url (Optional[:obj:`str`]): URL to redirect to upon unsuccessful authentication
317            language (Optional[:obj:`str`]): Language to display widget in
318            reference_id (Optional[:obj:`str`]): ID of a user in your app, which will be returned at the end of a successful auth
319            **kwargs: Optional additional arguments to be passed in to the body of the request
320
321        Returns:
322            :obj:`models.api_responses.TerraApiResponse`: API response object containing WidgetSession parsed response object if no error has occured
323        """
324        maybe_body_payload = {
325            "providers": ",".join(providers) if providers else None,
326            "auth_success_redirect_url": auth_success_redirect_url,
327            "auth_failure_redirect_url": auth_failure_redirect_url,
328            "language": language,
329            "reference_id": reference_id,
330        }
331        body_payload = utils.update_if_not_none({}, maybe_body_payload)
332        body_payload.update(kwargs)
333
334        widget_resp = requests.post(
335            f"{constants.BASE_URL}/auth/generateWidgetSession",
336            headers=self._auth_headers,
337            json=body_payload,
338        )
339        return api_responses.TerraApiResponse(widget_resp, dtype="widget_session")
340
341    def generate_authentication_url(
342        self,
343        resource: str,
344        auth_success_redirect_url: typing.Optional[str] = None,
345        auth_failure_redirect_url: typing.Optional[str] = None,
346        reference_id: typing.Optional[str] = None,
347        **kwargs: typing.Any,
348    ) -> api_responses.TerraApiResponse:
349        """
350        Generates an authentication URL to allow an end user to authenticate through the API. Users should be
351        redirected to the given URL in order to complete authentication. User ID will be provided in the response
352        for convenience (note that at this stage, said user will have yet to complete the auth flow)
353
354        Args:
355            resource (:obj:`str`): Provider to authenticate user with
356            auth_success_redirect_url (Optional[:obj:`str`]): URL to redirect to upon successful authentication
357            auth_failure_redirect_url (Optional[:obj:`str`]): URL to redirect to upon unsuccessful authentication
358            reference_id (Optional[:obj:`str`]): ID of a user in your app, which will be returned at the
359                end of a successful auth
360            **kwargs: Optional additional arguments to be passed in to the body of the request
361
362        Returns:
363            :obj:`models.api_responses.TerraApiResponse`: API response object containing UserAuthUrl parsed
364                response object if no error has occured
365
366        """
367
368        body_payload = {
369            "resource": resource,
370            "auth_success_redirect_url": auth_success_redirect_url,
371            "auth_failure_redirect_url": auth_failure_redirect_url,
372            "reference_id": reference_id,
373        }
374        body_payload.update(kwargs)
375
376        auth_resp = requests.post(
377            f"{constants.BASE_URL}/auth/authenticateUser",
378            headers=self._auth_headers,
379            json=body_payload,
380        )
381
382        return api_responses.TerraApiResponse(auth_resp, dtype="auth_url")
383
384    def get_user_info(self, user: user_.User) -> api_responses.TerraApiResponse:
385        """
386        Retrieve information on a given User, including is_authenticated status, indicating if the user has
387        successfully completed auth flow, or has yet to do so
388        Note: Also updates information on user object passed as an argument
389
390        Args:
391            user (:obj:`models.user.User`): User to retrieve information for
392
393        Returns:
394            :obj:`models.api_responses.TerraApiResponse`: API response object containing UserInfo parsed
395                response object if no error has occured
396        """
397        user_resp = requests.get(
398            f"{constants.BASE_URL}/userInfo",
399            params={"user_id": user.user_id},
400            headers=self._auth_headers,
401        )
402        return api_responses.TerraApiResponse(user_resp, dtype="user_info")
403
404    def deauthenticate_user(self, user: user_.User) -> api_responses.TerraApiResponse:
405        """
406        Deauthenticates the given User from the Api. If successful, this will trigger a `deauth`
407        webhook event.
408
409        Args:
410            user (:obj:`models.user.User`): User to Deauthenticate from the API
411
412        Returns:
413            :obj:`models.api_responses.TerraApiResponse`: API response object containing UserDeauthResp parsed response object if no error has occured
414        """
415        deauth_resp = requests.delete(
416            f"{constants.BASE_URL}/auth/deauthenticateUser",
417            params={"user_id": user.user_id},
418            headers=self._auth_headers,
419        )
420        deauth_resp.raise_for_status()
421        return api_responses.TerraApiResponse(deauth_resp, dtype="user_deauth")
422
423    def list_users(self) -> api_responses.TerraApiResponse:
424        """
425        Lists all users registered under Client's credentials on the API
426
427        Returns:
428            :obj:`models.api_responses.TerraApiResponse`: API response object containing SubscribedUsers parsed response object if no error has occured
429        """
430        users_resp = requests.get(f"{constants.BASE_URL}/subscriptions", headers=self._auth_headers)
431        return api_responses.TerraApiResponse(users_resp, dtype="subscriptions", client=self)
432
433    def list_providers(self) -> api_responses.TerraApiResponse:
434        """
435        Lists all providers on the API
436
437        Returns:
438            :obj:`models.api_responses.TerraApiResponse`: API response object containing ProvidersResponse parsed response object if no error has occured
439        """
440        providers_resp = requests.get(f"{constants.BASE_URL}/integrations", headers=self._auth_headers)
441        return api_responses.TerraApiResponse(providers_resp, dtype="providers")
442
443    def check_terra_signature(self, body: str, header: str) -> bool:
444        """
445        Function to test if the body of an API response comes from terra using SHA256
446
447        Args:
448            body (:obj:`str`): The body from API response as a string
449            header (:obj:`str`): The header from API response as a string
450
451        Returns:
452            :obj:`bool`: True if the API response comes from Terra
453        """
454
455        t, sig = (pair.split("=")[-1] for pair in header.split(","))
456
457        computed_signature = hmac.new(
458            bytes(self.secret, "utf-8"),
459            msg=bytes(f"{t}.{body}", "utf-8"),
460            digestmod=hashlib.sha256,
461        ).hexdigest()
462
463        if computed_signature != sig:
464            return False
465        # Signature was validated
466        return True
467
468    def handle_flask_webhook(self, request: flask.Request) -> typing.Optional[api_responses.TerraWebhookResponse]:
469        """
470        Parses Terra webhooks from a flask request
471
472        Args:
473            request (:obj:`flask.request`): the flask request object
474
475        Returns:
476            :obj:`models.api_responses.TerraApiResponse`: API response object containing ProvidersResponse parsed
477                response object if no error has occurred
478        """
479
480        if not self.check_terra_signature(request.get_data().decode("utf-8"), request.headers["terra-signature"]):
481            return None
482        ff = api_responses.TerraWebhookResponse(request.get_json(), dtype="hook")
483
484        return ff
485
486    def handle_webhook(
487        self, payload: str, terra_signature_header: str
488    ) -> typing.Optional[api_responses.TerraWebhookResponse]:
489        """
490        Function to Parse web hooks from Terra
491
492        Args:
493            payload (:obj:`str`): The body from API response as a string
494            terra_signature_header (:obj:`str`): The terra_signature header from API response as a string
495
496        Returns:
497            :obj:`models.api_responses.TerraApiResponse`: API response object containing ProvidersResponse parsed
498                response object if no error has occurred
499        """
500
501        if not self.check_terra_signature(payload, terra_signature_header):
502            return None
503        return api_responses.TerraWebhookResponse(json.loads(payload), dtype="hook")

18class TerraException(Exception):
19    """Base class for all exceptions raised by this library"""

29class NoBodyException(TerraException):
30    """
31    Exception raised when a TerraApi object has no body
32    """

22class NoClientAvailable(TerraException):
23    """
24    Exception raised when a TerraUser object has no Terra client set,
25    but a method requiring it is called.
26    """

41class NoDtypeException(TerraException):
42    """
43    Exception raised when a TerraPi object has no dtype
44    """

35class NoUserInfoException(TerraException):
36    """
37    Exception raised when a UserInfoDict is None
38    """