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