"""TransferWise utility class.""" import json from threading import Lock from typing import List from flask import g from oto import status import requests from collaborator import config from collaborator.connectors.datadog import call_datadog_with_event from collaborator.constants import logging from collaborator.constants import transferwise_api as api from collaborator.constants import transferwise_error as error from collaborator.constants import transferwise_webhook_events as events from collaborator.constants.transferwise_profile import TransferwiseProfileStatus from collaborator.models.rds.transferwise_profile_persister import ( TransferwiseProfilePersister, ) from collaborator.utils.error import OwsError from collaborator.utils.typing import User class Transferwise(object): """TransferWise utility class.""" base_url = config.TRANSFERWISE_BASE_URL client_id = config.TRANSFERWISE_CLIENT_ID client_secret = config.TRANSFERWISE_CLIENT_SECRET refresh_lock = Lock() def __init__( self, account=None, subaccount_id=None, ): """Init function for TransferWise object. Args: account (tuple): Optional Tuple with the account info. subaccount_id (int): Optional subaccount_id """ self.account = account self.subaccount_id = subaccount_id def __get( self, api_path, access_token=None, basic_authentication=False, headers={} ): """Abstraction of get requests. Args: api_path (str): The api uri. access_token(str): Optional parameter for api calls that need to be authenticated. basic_authentication (bool): Flag that indicates if we should use basic authentication Returns requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ combined_headers = {**self.__hydrate_headers(access_token), **headers} request_url = "{0}{1}".format(self.base_url, api_path) if basic_authentication: return requests.get( request_url, headers=combined_headers, auth=(self.client_id, self.client_secret), ) return requests.get(request_url, headers=combined_headers) def __get_with_profile_access_token(self, api_path, refresh_access_token=True): """Get request with access_token. Args: api_path (str): The api uri. of the get request refresh_access_token (bool): flag that activates/deactivates recursion Returns requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ if refresh_access_token: self.refresh_lock.acquire() try: profile = TransferwiseProfilePersister.get_active_profile( self.account, self.subaccount_id ) access_token = profile.get("access_token") get_response = self.__get(api_path, access_token=access_token) if get_response.status_code == status.UNAUTHORIZED and refresh_access_token: self.refresh_profile_access_token(profile) return self.__get_with_profile_access_token( api_path, refresh_access_token=False ) return get_response finally: if refresh_access_token: self.refresh_lock.release() def __post( self, api_path, payload, basic_authentication=False, access_token=None, headers={}, ): """Abstraction of post requests. Args: api_path (str): The api uri. payload (dict): dictionary with the payload of the post request has_access_token(str): Optional parameter for api calls that need to be authenticated. basic_authentication (bool): Flag that indicates if we should use basic_authentication Returns requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ request_url = "{0}{1}".format(self.base_url, api_path) if basic_authentication: combined_headers = { "Content-Type": "application/x-www-form-urlencoded", **headers, } return requests.post( request_url, data=payload, headers=combined_headers, auth=(self.client_id, self.client_secret), ) combined_headers = {**self.__hydrate_headers(access_token), **headers} post_response = requests.post( request_url, json=payload, headers=combined_headers ) return post_response def __post_with_profile_access_token( self, api_path, payload, refresh_access_token=True ): """Post request with access_token. Args: api_path (str): The api uri. payload (dict): dictionary with the payload of the post request refresh_access_token (bool): flag that activates/deactivates recursion Returns requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ profile = TransferwiseProfilePersister.get_active_profile( self.account, self.subaccount_id ) access_token = profile.get("access_token") post_response = self.__post(api_path, payload, access_token=access_token) if post_response.status_code == status.UNAUTHORIZED and refresh_access_token: self.refresh_profile_access_token(profile) return self.__post_with_profile_access_token( api_path, payload, refresh_access_token=False ) return post_response def __delete(self, api_path, has_access_token=None, basic_authentication=False): """Abstraction of delete requests. Args: api_path (str): The api uri. has_access_token(str): Optional parameter(str): Optional parameter for api calls that need to be authenticated. basic_authentication (bool): Flag that indicates if we should use basic authentication Returns requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ request_url = "{0}{1}".format(self.base_url, api_path) access_token = None profile = {} if basic_authentication: return requests.delete( request_url, headers={"Content-Type": "application/x-www-form-urlencoded"}, auth=(self.client_id, self.client_secret), ) if has_access_token: profile = TransferwiseProfilePersister.get_active_profile( self.account, self.subaccount_id ) access_token = profile.get("access_token") headers = self.__hydrate_headers(access_token) delete_response = requests.delete(request_url, headers=headers) if has_access_token and delete_response.status_code == status.UNAUTHORIZED: self.refresh_profile_access_token(profile) return self.__delete(api_path, has_access_token=has_access_token) return delete_response def __patch(self, api_path, payload, basic_authentication=False, access_token=None): """Abstraction of patch requests. Args: api_path (str): The api uri. payload (dict): dictionary with the payload of the post request has_access_token(str): Optional parameter for api calls that need to be authenticated. basic_authentication (bool): Flag that indicates if we should use basic_authentication Returns: requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ request_url = "{0}{1}".format(self.base_url, api_path) if basic_authentication: return requests.patch( request_url, data=payload, headers={"Content-Type": "application/x-www-form-urlencoded"}, auth=(self.client_id, self.client_secret), ) headers = self.__hydrate_headers(access_token) post_response = requests.patch(request_url, json=payload, headers=headers) return post_response def __patch_with_profile_access_token(self, api_path, payload): """Patch request with access_token. Args: api_path (str): The api uri. payload (dict): dictionary with the payload of the post request Returns: requests.Response object: Response object with attributes such as: - status_code - headers - encoding - json - text """ profile = TransferwiseProfilePersister.get_active_profile( self.account, self.subaccount_id ) access_token = profile.get("access_token") patch_response = self.__patch(api_path, payload, access_token=access_token) if patch_response.status_code == status.UNAUTHORIZED: self.refresh_profile_access_token(profile) return self.__patch_with_profile_access_token(api_path, payload) return patch_response def __hydrate_headers(self, access_token): """Hydrate headers based on the authentication type. Args: access_token(str): User's access token. for api calls that need to be authenticated. """ headers = {"Content-Type": "application/json"} if access_token: headers.update({"Authorization": "Bearer {}".format(access_token)}) return headers def get_account_requirements_uri(self, source: str, target: str) -> str: """Get account requirements uri.""" uri = api.ACCOUNT_REQUIREMENTS.format(source=source, target=target) if api.CURRENCY_JPY in [source, target]: uri += "&addressRequired=true" return uri def refresh_account_requirements(self, source, target, payload): """Refresh account requirements. Args: source (str): the source currency target (str): the target currency payload (dict): dictionary with the payload of the post request Returns: list: the account requirements. """ api_endpoint = self.get_account_requirements_uri(source, target) account_requirements_response = self.__post( api_endpoint, payload, headers=api.ACCOUNT_REQUIREMENTS_HEADERS ) if account_requirements_response.status_code != status.OK: raise OwsError( status=account_requirements_response.status_code, code=error.TRANSFERWISE_ACCOUNT_REQUIREMENTS_ERROR, message=account_requirements_response.text, ) return account_requirements_response.json() def get_account_requirements(self, source, target): """Fetch account requirements. Args: source (str): the source currency target (str): the target currency Returns: list: with the account requirements. """ api_endpoint = self.get_account_requirements_uri(source, target) account_requirements_response = self.__get( api_endpoint, headers=api.ACCOUNT_REQUIREMENTS_HEADERS ) if account_requirements_response.status_code != status.OK: raise OwsError( status=account_requirements_response.status_code, code=error.TRANSFERWISE_ACCOUNT_REQUIREMENTS_ERROR, message=account_requirements_response.text, ) return account_requirements_response.json() def get_field_validation(self, validator, query_string_params): """Validate field. Args: validator (str): This would be the path of the validator. e.g. if we are validating the sort code of the account path would have value sort-code. query_string_params (str): the constructed query string with the fields and values that needs validation. Returns: list: with the validation of the field as payload. """ api_endpoint = api.FIELD_VALIDATOR.format( validator=validator, query_string_params=query_string_params ) field_validation_response = self.__get(api_endpoint) if field_validation_response.status_code != status.OK: raise OwsError( status=field_validation_response.status_code, code=error.TRANFERWISE_FIELD_VALIDATION_ERROR, message=field_validation_response.text, ) return field_validation_response.json() def create_oauth_token(self, code, redirect_uri: str): """Use temp code to generate tokens. Args: code (str): temporary TransferWise code redirect_uri (str) Returns: dict: with the OAuth token info. """ api_endpoint = api.OAUTH_TOKEN payload = { "code": code, "grant_type": "authorization_code", "client_id": self.client_id, "redirect_uri": redirect_uri, } oauth_token_response = self.__post( api_endpoint, payload, basic_authentication=True ) if oauth_token_response.status_code != status.OK: raise OwsError( status=oauth_token_response.status_code, code=error.OAUTH_TOKEN_ERROR, message=oauth_token_response.text, ) return oauth_token_response.json() def create_recipient(self, payload: dict) -> dict: """Create a TransferWise recipient. Args: payload (dict): dictionary with the payload of the post request Returns: dict: with the result of recipient creation """ api_endpoint = api.ACCOUNTS result = self.__post_with_profile_access_token(api_endpoint, payload) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=error.APPLICATION_SUBSCRIPTION_ERROR, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_POST_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def delete_recipient(self, recipient_id: int): """Delete a TransferWise recipient. Args: recipient_id (int): recipient id """ api_endpoint = "{}/{}".format(api.ACCOUNTS, recipient_id) result = self.__delete(api_endpoint, has_access_token=True) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_DELETE_REQUEST_ERROR, message=result.text, status=result.status_code, ) def refresh_oauth_token(self, refresh_token=None, fetch_client_credentials=False): """Refresh access token. Args: refresh_token (str): the profile's activate refresh_token fetch_client_credentials (bool): Flag that is used to fetch client access token. Returns: dict: the new OAuth token info. """ api_endpoint = api.OAUTH_TOKEN payload = {"grant_type": "refresh_token", "refresh_token": refresh_token} if fetch_client_credentials: payload = {"grant_type": "client_credentials"} oauth_token_response = self.__post( api_endpoint, payload, basic_authentication=True ) if oauth_token_response.status_code != status.OK: # checks if we get a bad request if oauth_token_response.status_code == status.BAD_REQUEST: error_json = json.loads(oauth_token_response.text) # checks if the error is an invalid refresh token if error_json["error"] == "invalid_grant": # soft delete the profile TransferwiseProfilePersister.soft_delete_profile( self.account, self.subaccount_id, user=User("system", 0) ) raise OwsError( status=oauth_token_response.status_code, code=error.OAUTH_TOKEN_ERROR, message=oauth_token_response.text, ) return oauth_token_response.json() def refresh_profile_access_token(self, profile_credentials): """Refresh profile access token. Args: profile_credentials (dict): with the profile credentials e.g. profile_id, refresh_token Returns: dict: the updated profile data. """ profile_id = profile_credentials.get("profile_id") refresh_token = profile_credentials.get("refresh_token") account = self.account subaccount_id = self.subaccount_id transferwise_oauth_token = self.refresh_oauth_token(refresh_token) profile = TransferwiseProfilePersister.update_active_profile( account, subaccount_id, profile_id, transferwise_oauth_token["access_token"], transferwise_oauth_token["refresh_token"], None, ) return profile def subscribe_application_to_event(self, name, trigger_on): """Subscribe to application event. Args: name (str): The name of the subscription trigger_on (str): the event hook trigger Returns: dict: the created subscription """ client_access_token_response = self.refresh_oauth_token( fetch_client_credentials=True ) client_access_token = client_access_token_response.get("access_token") api_endpoint = api.WEBHOOK_APPLICATION_SUBSCRIPTION.format( client_key=self.client_id ) payload = { "name": name, "trigger_on": events.application_subscriptions[trigger_on], "delivery": { "version": events.delivery_version, "url": config.WEBHOOK_DELIVERY_URL, }, } client_subscription_response = self.__post( api_endpoint, payload, access_token=client_access_token ) if client_subscription_response.status_code != status.CREATED: call_datadog_with_event( title=(logging.TRANFERWISE_ERROR_APPLICATION_SUBSCRIPTION), text=error.APPLICATION_SUBSCRIPTION_ERROR, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( status=client_subscription_response.status_code, code=error.APPLICATION_SUBSCRIPTION_ERROR, message=client_subscription_response.text, ) return client_subscription_response.json() def create_quote(self, payload): """Create quote for payment. Args: payload (dict): dictionary with the payload of the quote Returns: dict: the created quote """ recipient_id = payload.pop("recipient_id", None) collaborator_id = payload.pop("collaborator_id", None) profile_id = payload.get("profile") api_endpoint = api.QUOTES.format(profile_id=profile_id) result = self.__post_with_profile_access_token( api_endpoint, payload, refresh_access_token=False ) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_POST_REQUEST_ERROR, message=result.text, status=result.status_code, ) response_data = result.json() response_data["recipient_id"] = recipient_id response_data["collaborator_id"] = collaborator_id return response_data def get_quote(self, quote_uuid: str, profile_id: int) -> dict: """Get a single quote by UUID. Args: quote_uuid (str): Quote UUID. profile_id (int): the profile unique identifier Returns: dict: Quote data. """ api_endpoint = api.GET_QUOTE.format( profile_id=profile_id, quote_uuid=quote_uuid ) result = self.__get_with_profile_access_token(api_endpoint) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def create_batch(self, profile_id, payload): """Create a batch group. Args: payload (dict): dictionary with the payload of the batch Returns: dict: the created batch """ api_endpoint = api.BATCH.format(profile_id=profile_id) result = self.__post_with_profile_access_token(api_endpoint, payload) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_POST_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def create_batch_transfer(self, payload, profile_id, batch_id): """Create batch transfer. Args: profile_id (int): the profile unique identifier batch_group_id (int): the unique identifier of a batch payload (dict): dictionary with the payload of the transfer Returns: dict: the created batch transfer """ api_endpoint = api.BATCH_TRANSFER.format( profile_id=profile_id, batch_group_id=batch_id ) payload["customerTransactionId"] = payload["quoteUuid"] result = self.__post_with_profile_access_token( api_endpoint, payload, refresh_access_token=False ) if not result: tw_error = result.json() error_message = tw_error["errors"][0] error_message.update({"called_with_params": payload}) call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_POST_REQUEST_ERROR, message=error_message, status=result.status_code, ) return result.json() def get_batch_state(self, profile_id, batch_id): """Retrieve current state of a batch. Args: profile_id (int): the profile unique identifier batch_group_id (int): the unique identifier of a batch Returns: dict: the current state """ api_endpoint = api.BATCH_STATE.format( profile_id=profile_id, batch_group_id=batch_id ) result = self.__get_with_profile_access_token(api_endpoint) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def modify_batch_state( self, profile_id, batch_id, version_number=1, complete_batch=True ): """Modify_ batch state. Args: profile_id (int): the profile unique identifier batch_id (int): the unique identifier of a batch of the transfer Returns: dict: the finalized batch """ payload = ( {"status": "COMPLETED", "version": version_number} if complete_batch else {"status": "CANCELLED", "version": version_number} ) api_endpoint = api.BATCH_STATE.format( profile_id=profile_id, batch_group_id=batch_id ) result = self.__patch_with_profile_access_token(api_endpoint, payload) if not result: call_datadog_with_event( title=logging.TRANFERWISE_REQUEST_ERROR_TITLE.format( vendor_id=self.account.id ), text=result.text, tags=logging.TRANSFERWISE_TAGS, ) raise OwsError( code=error.TRANSFERWISE_PATCH_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def get_batch_group(self, profile_id: int, batch_group_id: str) -> dict: """Get batch transfer group details by id. Args: profile_id (int): TransferWise profile id batch_group_id (str): UUID for the transfer batch group Returns: dict: with the transfer group details. """ api_endpoint = api.GET_BATCH_TRANSFER.format( profile_id=profile_id, batch_group_id=batch_group_id ) result = self.__get_with_profile_access_token(api_endpoint) if not result: raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def simulate_transfer_processing( self, transfer_id: int, transfer_status: str ) -> dict: """Simulate transfer processing. Args: transfer_id (int): the transfer_id transfer_status (str): the transfer status returns: dictionary with the transfer details. """ api_endpoint = api.SIMULATE_TRANSFER_PROCESSING.format( transfer_id=transfer_id, transfer_status=transfer_status ) result = self.__get_with_profile_access_token(api_endpoint) if not result: raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=result.text, status=result.status_code, ) return result.json() def get_profile_verification_status( self, profile_id: int, currency: str, access_token: str ) -> str: """Retrieve the verification status of a profile. Args: profile_id (int): ID of the profile to check currency (str): Currency access_token (str): Access token used to retrieve the status. Returns: str: verification status """ api_endpoint = api.PROFILE_VERIFICATION_STATUS.format( profile_id=profile_id, currency=currency ) result = self.__post(api_endpoint, None, access_token=access_token) if not result: g.log.error(f"Unable to get profile verification status: {result.text}") return TransferwiseProfileStatus.ERROR routes = result.json()["routes"] currency_route = next( route for route in routes if route["source_currency"] == currency ) status = currency_route["current_status"] if status == "verified": return TransferwiseProfileStatus.VERIFIED elif status == "not_verified": return TransferwiseProfileStatus.UNVERIFIED else: g.log.error(f"Unhandled profile verification status: {status}") return TransferwiseProfileStatus.ERROR def ping_api(self, profile_id: int): """Ping Api by retrieving profile. Args: profile_id (int): ID of the profile to check access_token (str): Access token used to retrieve the profile. Returns: str: verification status """ api_endpoint = api.PROFILE.format(profile_id=profile_id) result = self.__get_with_profile_access_token(api_endpoint) if not result: raise OwsError( code=error.OAUTH_TOKEN_ERROR, message=result.text, status=result.status_code, ) def get_application_subscriptions(self) -> List[str]: """Get all the currently active application subscriptions. Returns: [str]: List of active application subscriptions. """ api_endpoint = api.WEBHOOK_APPLICATION_SUBSCRIPTION.format( client_key=self.client_id ) api_response = self.__get(api_endpoint, basic_authentication=True) if not api_response: raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=api_response.text, status=api_response.status_code, ) return [subscription["trigger_on"] for subscription in api_response.json()] def get_transfer_status(self, transfer_id: int) -> str: """Retrieve the status of a transfer. Args: transfer_id (int): ID of the transfer to check access_token (str): Access token used to retrieve the status. Returns: str: transfer status """ api_endpoint = api.TRANSFERS.format(transfer_id=transfer_id) result = self.__get_with_profile_access_token(api_endpoint) if not result: return f"transfer_fetch_err:{transfer_id}" return result.json()["status"] def get_transfer_requirements(self, transfer: dict) -> dict: """Get transfer requirements for a given xfer. Args: transfer (dict): transfer for fetching requirements. Returns: dict: transfer requirements """ result = self.__post_with_profile_access_token( api.TRANSFER_REQUIREMENTS, payload=transfer["transfer"] ) if not result: raise OwsError( code=error.TRANSFERWISE_GET_REQUEST_ERROR, message=result.text, status=result.status_code, ) return { "currency": transfer["currency"], "requirements": result.json()[0]["fields"], }