import logging import requests from requests_oauthlib import OAuth2Session from labelaudit.lib.vapi.oauth_code_error import OauthCodeError class OauthClient: def __init__(self, client_id, client_secret, base_url, redirect_uri, user_id, user_type): self.client_id = client_id self.client_secret = client_secret self.base_url = base_url self.redirect_uri = redirect_uri self.user_id = user_id self.user_type = user_type self.oauth_session = OAuth2Session(self.client_id) def auth_url(self): """Generates the URL to be called when fetching an auth code.""" base_url = self.base_url + '/authorize/getoauthcode' auth_url, state = self.oauth_session.authorization_url( base_url, None, user_id=self.user_id, user_type=self.user_type) # passing this in kwargs to authorization_url doesn't seem to work... auth_url += '&isRedirect=0' return auth_url def fetch_code(self): """Fetches an authorization code for the API.""" auth_response = requests.get(self.auth_url()) auth_response_body = auth_response.json() if auth_response.status_code != requests.codes.ok: raise OauthCodeError('Error fetching OAuth code', auth_response_body, auth_response.status_code) return auth_response_body['query']['code'] def fetch_token(self): """Fetches an access token for the API. :return: A dictionary object containing token data. Keys: access_token refresh_token """ token_url = self.base_url + '/authorize/getaccesstoken' oauth_code = self.fetch_code() return self.oauth_session.fetch_token(token_url, oauth_code, client_secret=self.client_secret, user_type=self.user_type) def logger(self): """Returns the logger instance used internally by requests_oauthlib. Useful for debugging purposes. Example usage: import logging import sys logger = oauth_client.logger() logger.setLevel(logging.DEBUG) handler = logging.StreamHandler(stream=sys.stdout) handler.setLevel(logging.DEBUG) logger.addHandler(handler) """ return logging.getLogger('requests_oauthlib')