|
| 1 | +import requests |
| 2 | +from urllib.parse import urlencode |
| 3 | +from youtube.exceptions import YouTubeException |
| 4 | + |
| 5 | + |
| 6 | +class API: |
| 7 | + _client_id = None |
| 8 | + _client_secret = None |
| 9 | + _api_key = None |
| 10 | + _access_token = None |
| 11 | + _api_base_url = 'https://www.googleapis.com/youtube/v3/' |
| 12 | + _auth_url = 'https://accounts.google.com/o/oauth2/auth' |
| 13 | + _exchange_code_url = 'https://accounts.google.com/o/oauth2/token' |
| 14 | + _scope = [ |
| 15 | + 'https://www.googleapis.com/auth/youtube', |
| 16 | + 'https://www.googleapis.com/auth/userinfo.profile' |
| 17 | + ] |
| 18 | + _part = 'id,snippet' |
| 19 | + |
| 20 | + def __init__(self, client_id, client_secret, api_key, access_token=None, api_url=None): |
| 21 | + self._client_id = client_id |
| 22 | + self._client_secret = client_secret |
| 23 | + self._api_key = api_key |
| 24 | + self._access_token = access_token |
| 25 | + |
| 26 | + if api_url: |
| 27 | + self.api_url = api_url |
| 28 | + |
| 29 | + def get(self, endpoint, **kwargs): |
| 30 | + if self._access_token: |
| 31 | + kwargs['access_token'] = self._access_token |
| 32 | + else: |
| 33 | + kwargs['api_key'] = self._api_key |
| 34 | + |
| 35 | + if 'part' not in kwargs: |
| 36 | + kwargs['part'] = self._part |
| 37 | + |
| 38 | + return self.response(self._get(self._api_base_url+endpoint, params=kwargs)) |
| 39 | + |
| 40 | + def post(self, endpoint, **kwargs): |
| 41 | + if self._access_token: |
| 42 | + kwargs['access_token'] = self._access_token |
| 43 | + else: |
| 44 | + kwargs['api_key'] = self._api_key |
| 45 | + |
| 46 | + return self.response(self._post(self._api_base_url + endpoint, params=kwargs)) |
| 47 | + |
| 48 | + def get_auth_url(self, redirect_uri, **kwargs): |
| 49 | + kwargs = {**{ |
| 50 | + 'response_type': 'code', |
| 51 | + 'redirect_uri': redirect_uri, |
| 52 | + 'client_id': self._client_id, |
| 53 | + 'access_type': 'offline', |
| 54 | + 'approval_prompt': 'force' |
| 55 | + }, **kwargs} |
| 56 | + |
| 57 | + if 'scope' not in kwargs: |
| 58 | + kwargs['scope'] = self._scope |
| 59 | + |
| 60 | + kwargs['scope'] = ' '.join(kwargs['scope']) |
| 61 | + |
| 62 | + return self._auth_url+'?'+urlencode(kwargs) |
| 63 | + |
| 64 | + def exchange_code(self, code, redirect_uri): |
| 65 | + params = { |
| 66 | + 'code': code, 'client_id': self._client_id, 'client_secret': self._client_secret, |
| 67 | + 'redirect_uri': redirect_uri, 'grant_type': 'authorization_code' |
| 68 | + } |
| 69 | + headers = {'Content-Type': 'application/x-www-form-urlencoded'} |
| 70 | + response = self.response(self._post(self._exchange_code_url, params=params, headers=headers)) |
| 71 | + |
| 72 | + if response and 'access_token' in response: |
| 73 | + self._access_token = response['access_token'] |
| 74 | + |
| 75 | + return response |
| 76 | + |
| 77 | + def refresh_token(self, refresh_token): |
| 78 | + params = { |
| 79 | + 'client_id': self._client_id, 'client_secret': self._client_secret, 'refresh_token': refresh_token, |
| 80 | + 'grant_type': 'refresh_token' |
| 81 | + } |
| 82 | + headers = {'Content-Type': 'application/x-www-form-urlencoded'} |
| 83 | + response = self.response(self._post(self._exchange_code_url, params=params, headers=headers)) |
| 84 | + |
| 85 | + if response and 'access_token' in response: |
| 86 | + self._access_token = response['access_token'] |
| 87 | + |
| 88 | + return response |
| 89 | + |
| 90 | + def get_profile(self): |
| 91 | + return self.response(self._get( |
| 92 | + 'https://www.googleapis.com/oauth2/v1/userinfo', |
| 93 | + {'access_token': self._access_token} |
| 94 | + )) |
| 95 | + |
| 96 | + @staticmethod |
| 97 | + def _get(url, params=None, **kwargs): |
| 98 | + result = requests.get(url, params=params, **kwargs) |
| 99 | + _response = result.json() |
| 100 | + |
| 101 | + if _response and 'error' in _response: |
| 102 | + raise YouTubeException(_response['error']['code'], _response['error']['message'], result) |
| 103 | + |
| 104 | + if result.status_code != 200: |
| 105 | + raise YouTubeException(result.status_code, result.reason, result) |
| 106 | + |
| 107 | + return result |
| 108 | + |
| 109 | + @staticmethod |
| 110 | + def _post(url, params=None, **kwargs): |
| 111 | + result = requests.post(url, data=params, **kwargs) |
| 112 | + _response = result.json() |
| 113 | + |
| 114 | + if _response and 'error' in _response: |
| 115 | + if isinstance(_response['error'], str): |
| 116 | + raise YouTubeException(result.status_code, _response['error'], result) |
| 117 | + else: |
| 118 | + raise YouTubeException(_response['error']['code'], _response['error']['message'], result) |
| 119 | + |
| 120 | + if result.status_code != 200: |
| 121 | + raise YouTubeException(result.status_code, result.reason, result) |
| 122 | + |
| 123 | + return result |
| 124 | + |
| 125 | + @staticmethod |
| 126 | + def response(response): |
| 127 | + return response.json() |
0 commit comments