diff --git a/tests/conftest.py b/tests/conftest.py index 80c988d..b8fe5fa 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,35 +26,27 @@ ] -class MockCore(trakt.core.Core): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) +class MockCore: + def __init__(self): self.mock_data = {} for mock_file in MOCK_DATA_FILES: with open(mock_file, encoding='utf-8') as f: self.mock_data.update(json.load(f)) - def _handle_request(self, method, url, data=None): - uri = url[len(trakt.core.BASE_URL):] + def request(self, method, uri, data=None): if uri.startswith('/'): uri = uri[1:] # use a deepcopy of the mocked data to ensure clean responses on every # request. this prevents rewrites to JSON responses from persisting - method_responses = deepcopy(self.mock_data).get(uri, {}) - result = method_responses.get(method.upper()) - if result is None: - print(f"Missing mock for {method.upper()} {trakt.core.BASE_URL}{uri}") - - return result - - -"""Override utility functions from trakt.core to use an underlying MockCore -instance -""" -trakt.core.CORE = MockCore() -trakt.core.get = trakt.core.CORE.get -trakt.core.post = trakt.core.CORE.post -trakt.core.delete = trakt.core.CORE.delete -trakt.core.put = trakt.core.CORE.put + method_responses = self.mock_data.get(uri, {}) + response = method_responses.get(method.upper()) + if response is None: + print(f"No mock for {uri}") + return deepcopy(response) + + trakt.core.CLIENT_ID = 'FOO' trakt.core.CLIENT_SECRET = 'BAR' + +# Override request function with MockCore instance +trakt.core.api().request = MockCore().request diff --git a/trakt/__version__.py b/trakt/__version__.py index bef3069..2005fa3 100644 --- a/trakt/__version__.py +++ b/trakt/__version__.py @@ -1 +1 @@ -__version__ = "3.4.0.dev0" +__version__ = "4.0.0.dev0" diff --git a/trakt/api.py b/trakt/api.py new file mode 100644 index 0000000..2d34305 --- /dev/null +++ b/trakt/api.py @@ -0,0 +1,181 @@ +import json +import logging +from datetime import datetime, timedelta, timezone +from functools import lru_cache +from json import JSONDecodeError + +from requests import Session +from requests.auth import AuthBase + +from trakt import errors +from trakt.config import AuthConfig +from trakt.core import TIMEOUT +from trakt.errors import BadResponseException, OAuthException + +__author__ = 'Elan Ruusamäe' + + +class HttpClient: + """Class for abstracting HTTP requests + """ + + #: Default request HEADERS + headers = {'Content-Type': 'application/json', 'trakt-api-version': '2'} + + def __init__(self, base_url: str, session: Session, timeout=None): + self.base_url = base_url + self.session = session + self.auth = None + self.timeout = timeout or TIMEOUT + self.logger = logging.getLogger('trakt.http_client') + + def get(self, url: str): + return self.request('get', url) + + def delete(self, url: str): + self.request('delete', url) + + def post(self, url: str, data): + return self.request('post', url, data=data) + + def put(self, url: str, data): + return self.request('put', url, data=data) + + def set_auth(self, auth): + self.auth = auth + + def request(self, method, url, data=None): + """Handle actually talking out to the trakt API, logging out debug + information, raising any relevant `TraktException` Exception types, + and extracting and returning JSON data + + :param method: The HTTP method we're executing on. Will be one of + post, put, delete, get + :param url: The fully qualified url to send our request to + :param data: Optional data payload to send to the API + :return: The decoded JSON response from the Trakt API + :raises TraktException: If any non-200 return code is encountered + """ + + url = self.base_url + url + self.logger.debug('REQUEST [%s] (%s)', method, url) + if method == 'get': # GETs need to pass data as params, not body + response = self.session.request(method, url, headers=self.headers, auth=self.auth, timeout=self.timeout, params=data) + else: + response = self.session.request(method, url, headers=self.headers, auth=self.auth, timeout=self.timeout, data=json.dumps(data)) + self.logger.debug('RESPONSE [%s] (%s): %s', method, url, str(response)) + if response.status_code == 204: # HTTP no content + return None + self.raise_if_needed(response) + + return self.decode_response(response) + + @staticmethod + def decode_response(response): + try: + return json.loads(response.content.decode('UTF-8', 'ignore')) + except JSONDecodeError as e: + raise BadResponseException(f"Unable to parse JSON: {e}") + + def raise_if_needed(self, response): + if response.status_code in self.error_map: + raise self.error_map[response.status_code](response) + + @property + @lru_cache(maxsize=None) + def error_map(self): + """Map HTTP response codes to exception types + """ + + # Get all of our exceptions except the base exception + errs = [getattr(errors, att) for att in errors.__all__ + if att != 'TraktException'] + + return {err.http_code: err for err in errs} + + +class TokenAuth(AuthBase): + """Attaches Trakt.tv token Authentication to the given Request object.""" + + #: The OAuth2 Redirect URI for your OAuth Application + REDIRECT_URI: str = 'urn:ietf:wg:oauth:2.0:oob' + + def __init__(self, client: HttpClient, config: AuthConfig): + super().__init__() + self.config = config + self.client = client + # OAuth token validity checked + self.OAUTH_TOKEN_VALID = None + self.logger = logging.getLogger('trakt.api.token_auth') + + def __call__(self, r): + # Skip oauth requests + if r.path_url.startswith('/oauth/'): + return r + + [client_id, client_token] = self.get_token() + + r.headers.update({ + 'trakt-api-key': client_id, + 'Authorization': f'Bearer {client_token}', + }) + return r + + def get_token(self): + """Return client_id, client_token pair needed for Trakt.tv authentication + """ + + self.config.load() + # Check token validity and refresh token if needed + if not self.OAUTH_TOKEN_VALID and self.config.have_refresh_token(): + self.validate_token() + + return [ + self.config.CLIENT_ID, + self.config.OAUTH_TOKEN, + ] + + def validate_token(self): + """Check if current OAuth token has not expired""" + + current = datetime.now(tz=timezone.utc) + expires_at = datetime.fromtimestamp(self.config.OAUTH_EXPIRES_AT, tz=timezone.utc) + if expires_at - current > timedelta(days=2): + self.OAUTH_TOKEN_VALID = True + else: + self.refresh_token() + + def refresh_token(self): + """Request Trakt API for a new valid OAuth token using refresh_token""" + + self.logger.info("OAuth token has expired, refreshing now...") + data = { + 'client_id': self.config.CLIENT_ID, + 'client_secret': self.config.CLIENT_SECRET, + 'refresh_token': self.config.OAUTH_REFRESH, + 'redirect_uri': self.REDIRECT_URI, + 'grant_type': 'refresh_token' + } + + try: + response = self.client.post('/oauth/token', data) + except OAuthException: + self.logger.debug( + "Rejected - Unable to refresh expired OAuth token, " + "refresh_token is invalid" + ) + return + + self.config.update( + OAUTH_TOKEN=response.get("access_token"), + OAUTH_REFRESH=response.get("refresh_token"), + OAUTH_EXPIRES_AT=response.get("created_at") + response.get("expires_in"), + ) + self.OAUTH_TOKEN_VALID = True + + self.logger.info( + "OAuth token successfully refreshed, valid until {}".format( + datetime.fromtimestamp(self.config.OAUTH_EXPIRES_AT, tz=timezone.utc) + ) + ) + self.config.store() diff --git a/trakt/auth/__init__.py b/trakt/auth/__init__.py new file mode 100644 index 0000000..4e45749 --- /dev/null +++ b/trakt/auth/__init__.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +"""Authentication methods""" + +__author__ = 'Jon Nappi, Elan Ruusamäe' + +from trakt import DEVICE_AUTH, OAUTH_AUTH, PIN_AUTH, api +from trakt import config as config_factory +from trakt.config import AuthConfig + + +def pin_auth(*args, config, **kwargs): + from trakt.auth.pin import PinAuthAdapter + + return PinAuthAdapter(*args, client=api(), config=config, **kwargs).authenticate() + + +def oauth_auth(*args, config, **kwargs): + from trakt.auth.oauth import OAuthAdapter + + return OAuthAdapter(*args, client=api(), config=config, **kwargs).authenticate() + + +def device_auth(config): + from trakt.auth.device import DeviceAuthAdapter + + return DeviceAuthAdapter(client=api(), config=config).authenticate() + + +def get_client_info(app_id: bool, config: AuthConfig): + """Helper function to poll the user for Client ID and Client Secret + strings + + :return: A 2-tuple of client_id, client_secret + """ + print('If you do not have a client ID and secret. Please visit the ' + 'following url to create them.') + print('https://trakt.tv/oauth/applications') + client_id = input('Please enter your client id: ') + client_secret = input('Please enter your client secret: ') + if app_id: + msg = f'Please enter your application ID ({config.APPLICATION_ID}): ' + user_input = input(msg) + if user_input: + config.APPLICATION_ID = user_input + return client_id, client_secret + + +def init_auth(method: str, *args, client_id=None, client_secret=None, store=False, **kwargs): + """Run the auth function specified by *AUTH_METHOD* + + :param store: Boolean flag used to determine if your trakt api auth data + should be stored locally on the system. Default is :const:`False` for + the security conscious + """ + + methods = { + PIN_AUTH: pin_auth, + OAUTH_AUTH: oauth_auth, + DEVICE_AUTH: device_auth, + } + + config = config_factory() + adapter = methods.get(method, PIN_AUTH) + + """ + Update client_id, client_secret from input or ask them interactively + """ + if client_id is None and client_secret is None: + client_id, client_secret = get_client_info(adapter.NEEDS_APPLICATION_ID, config) + config.CLIENT_ID, config.CLIENT_SECRET = client_id, client_secret + + adapter(*args, config=config, **kwargs) + + if store: + config.store() diff --git a/trakt/auth/base.py b/trakt/auth/base.py new file mode 100644 index 0000000..2ecca9d --- /dev/null +++ b/trakt/auth/base.py @@ -0,0 +1,6 @@ +class BaseAdapter: + #: The OAuth2 Redirect URI for your OAuth Application + REDIRECT_URI: str = 'urn:ietf:wg:oauth:2.0:oob' + + #: True if the Adapter needs APPLICATION_ID + NEEDS_APPLICATION_ID = False diff --git a/trakt/auth/device.py b/trakt/auth/device.py new file mode 100644 index 0000000..5bd7234 --- /dev/null +++ b/trakt/auth/device.py @@ -0,0 +1,115 @@ +from time import sleep, time + +from trakt.api import HttpClient +from trakt.auth.base import BaseAdapter +from trakt.config import AuthConfig +from trakt.errors import (BadRequestException, RateLimitException, + TraktException) + + +class DeviceAuthAdapter(BaseAdapter): + error_messages = { + 404: 'Invalid device_code', + 409: 'You already approved this code', + 410: 'The tokens have expired, restart the process', + 418: 'You explicitly denied this code', + } + + success_message = ( + "You've been successfully authenticated. " + "With access_token {access_token} and refresh_token {refresh_token}" + ) + + def __init__(self, client: HttpClient, config: AuthConfig): + self.client = client + self.config = config + + def authenticate(self): + """Process for authenticating using device authentication. + + The function will attempt getting the device_id, and provide + the user with a url and code. After getting the device + id, a timer is started to poll periodic for a successful authentication. + This is a blocking action, meaning you + will not be able to run any other code, while waiting for an access token. + + If you want more control over the authentication flow, use the functions + get_device_code and get_device_token. + Where poll_for_device_token will check if the "offline" + authentication was successful. + """ + + response = self.get_device_code() + device_code = response['device_code'] + interval = response['interval'] + + # No need to check for expiration, the API will notify us. + while True: + try: + response = self.get_device_token(device_code) + print(self.success_message.format_map(response)) + break + except RateLimitException: + # slow down + interval *= 2 + except BadRequestException: + # not pending + pass + except TraktException as e: + print(self.error_messages.get(e.http_code, response.response)) + + sleep(interval) + + def get_device_code(self): + """Generate a device code, used for device oauth authentication. + + Trakt docs: https://trakt.docs.apiary.io/#reference/ + authentication-devices/device-code + :return: Your OAuth device code. + """ + + data = {"client_id": self.config.CLIENT_ID} + response = self.client.post('/oauth/device/code', data=data) + + print('Your user code is: {user_code}, please navigate to {verification_url} to authenticate'.format( + user_code=response.get('user_code'), + verification_url=response.get('verification_url') + )) + + response['requested'] = time() + + return response + + def get_device_token(self, device_code): + """ + Trakt docs: https://trakt.docs.apiary.io/#reference/ + authentication-devices/get-token + Response: + { + "access_token": "", + "token_type": "bearer", + "expires_in": 7776000, + "refresh_token": "", + "scope": "public", + "created_at": 1519329051 + } + :return: Information regarding the authentication polling. + :return type: dict + """ + + data = { + "code": device_code, + "client_id": self.config.CLIENT_ID, + "client_secret": self.config.CLIENT_SECRET + } + + # We only get json on success. Code throws on errors + response = self.client.post('/oauth/device/token', data=data) + + self.config.update( + OAUTH_TOKEN=response.get('access_token'), + OAUTH_REFRESH=response.get('refresh_token'), + OAUTH_EXPIRES_AT=response.get("created_at") + response.get("expires_in"), + ) + + return response diff --git a/trakt/auth/oauth.py b/trakt/auth/oauth.py new file mode 100644 index 0000000..16c963a --- /dev/null +++ b/trakt/auth/oauth.py @@ -0,0 +1,60 @@ +from urllib.parse import urljoin + +from requests_oauthlib import OAuth2Session + +from trakt.api import HttpClient +from trakt.auth.base import BaseAdapter +from trakt.config import AuthConfig + + +class OAuthAdapter(BaseAdapter): + def __init__(self, username, client: HttpClient, config: AuthConfig, oauth_cb=None): + """ + :param username: Your trakt.tv username + :param oauth_cb: Callback function to handle the retrieving of the OAuth + PIN. Default function `_terminal_oauth_pin` for terminal auth + """ + self.username = username + self.client = client + self.config = config + self.oauth_cb = self.terminal_oauth_pin if oauth_cb is None else oauth_cb + + def authenticate(self): + """Generate an access_token to allow your application to authenticate via + OAuth + """ + + base_url = self.client.base_url + authorization_base_url = urljoin(base_url, '/oauth/authorize') + token_url = urljoin(base_url, '/oauth/token') + + # OAuth endpoints given in the API documentation + oauth = OAuth2Session(self.config.CLIENT_ID, redirect_uri=self.REDIRECT_URI, state=None) + + # Authorization URL to redirect user to Trakt for authorization + authorization_url, _ = oauth.authorization_url(authorization_base_url, username=self.username) + + # Calling callback function to get the OAuth PIN + oauth_pin = self.oauth_cb(authorization_url) + + # Fetch, assign, and return the access token + oauth.fetch_token(token_url, client_secret=self.config.CLIENT_SECRET, code=oauth_pin) + self.config.update( + OAUTH_TOKEN=oauth.token['access_token'], + OAUTH_REFRESH=oauth.token['refresh_token'], + OAUTH_EXPIRES_AT=oauth.token["created_at"] + oauth.token["expires_in"], + ) + + @staticmethod + def terminal_oauth_pin(authorization_url): + """Default OAuth callback used for terminal applications. + + :param authorization_url: Predefined url by function `oauth_auth`. URL will + be prompted to you in the terminal + :return: OAuth PIN + """ + print('Please go here and authorize,', authorization_url) + + # Get the authorization verifier code from the callback url + response = input('Paste the Code returned here: ') + return response diff --git a/trakt/auth/pin.py b/trakt/auth/pin.py new file mode 100644 index 0000000..fdd5c43 --- /dev/null +++ b/trakt/auth/pin.py @@ -0,0 +1,45 @@ +import sys + +from trakt.api import HttpClient +from trakt.auth.base import BaseAdapter +from trakt.config import AuthConfig + + +class PinAuthAdapter(BaseAdapter): + NEEDS_APPLICATION_ID = True + + def __init__(self, client: HttpClient, config: AuthConfig, pin=None): + """ + :param pin: Optional Trakt API PIN code. If one is not specified, you will + be prompted to go generate one + """ + self.pin = pin + self.client = client + self.config = config + + def authenticate(self): + """Generate an access_token from a Trakt API PIN code. + """ + + if self.pin is None and self.config.APPLICATION_ID is None: + print('You must set the APPLICATION_ID of the Trakt application you ' + 'wish to use. You can find this ID by visiting the following ' + 'URL.') + print('https://trakt.tv/oauth/applications') + sys.exit(1) + if self.pin is None: + print('If you do not have a Trakt.tv PIN, please visit the following ' + 'url and log in to generate one.') + pin_url = 'https://trakt.tv/pin/{id}'.format(id=self.config.APPLICATION_ID) + print(pin_url) + self.pin = input('Please enter your PIN: ') + data = { + 'code': self.pin, + 'redirect_uri': self.REDIRECT_URI, + 'grant_type': 'authorization_code', + 'client_id': self.config.CLIENT_ID, + 'client_secret': self.config.CLIENT_SECRET, + } + + response = self.client.post('/oauth/token', data) + self.config.OAUTH_TOKEN = response.get('access_token', None) diff --git a/trakt/config.py b/trakt/config.py new file mode 100644 index 0000000..62bf868 --- /dev/null +++ b/trakt/config.py @@ -0,0 +1,72 @@ +"""Class for trakt.tv auth config""" + +__author__ = 'Elan Ruusamäe' + +import json +from dataclasses import dataclass +from os.path import exists +from typing import Optional + + +@dataclass +class AuthConfig: + APPLICATION_ID: Optional[str] + CLIENT_ID: Optional[str] + CLIENT_SECRET: Optional[str] + OAUTH_EXPIRES_AT: Optional[int] + OAUTH_REFRESH: Optional[int] + OAUTH_TOKEN: Optional[str] + + def __init__(self, config_path): + self.config_path = config_path + + def have_refresh_token(self): + return self.OAUTH_EXPIRES_AT and self.OAUTH_REFRESH + + def get(self, name, default=None): + try: + return self.__getattribute__(name) + except AttributeError: + return default + + def set(self, name, value): + self.__setattr__(name, value) + + def update(self, **kwargs): + for name, value in kwargs.items(): + self.__setattr__(name, value) + + return self + + def all(self): + result = {} + for key in self.__annotations__.keys(): + result[key] = self.get(key) + + return result + + def load(self): + """ + Load in trakt API auth data from CONFIG_PATH + """ + if self.CLIENT_ID and self.CLIENT_SECRET or not exists(self.config_path): + return + + with open(self.config_path) as config_file: + config_data = json.load(config_file) + + for key in self.__annotations__.keys(): + # Don't overwrite + if self.get(key) is not None: + continue + + value = config_data.get(key, None) + self.set(key, value) + + def store(self): + """Store Trakt configurations at ``CONFIG_PATH`` + """ + + config = self.all() + with open(self.config_path, 'w') as config_file: + json.dump(config, config_file) diff --git a/trakt/core.py b/trakt/core.py index 2d62fcd..1a1a32f 100644 --- a/trakt/core.py +++ b/trakt/core.py @@ -2,30 +2,20 @@ """Objects, properties, and methods to be shared across other modules in the trakt package """ -import json -import logging import os -import sys -import time -from datetime import datetime, timedelta, timezone -from functools import wraps -from json import JSONDecodeError +from functools import lru_cache from typing import NamedTuple -from urllib.parse import urljoin import requests -from requests_oauthlib import OAuth2Session - -from trakt import errors -from trakt.errors import BadResponseException __author__ = 'Jon Nappi' __all__ = ['Airs', 'Alias', 'Comment', 'Genre', 'get', 'delete', 'post', 'put', 'init', 'BASE_URL', 'CLIENT_ID', 'CLIENT_SECRET', 'DEVICE_AUTH', - 'REDIRECT_URI', 'HEADERS', 'CONFIG_PATH', 'OAUTH_TOKEN', + 'CONFIG_PATH', 'OAUTH_TOKEN', 'OAUTH_REFRESH', 'PIN_AUTH', 'OAUTH_AUTH', 'AUTH_METHOD', + 'config', 'api', 'TIMEOUT', - 'APPLICATION_ID', 'get_device_code', 'get_device_token'] + 'APPLICATION_ID'] #: The base url for the Trakt API. Can be modified to run against different #: Trakt.tv environments @@ -37,21 +27,12 @@ #: The Trakt.tv OAuth Client Secret for your OAuth Application CLIENT_SECRET = None -#: The OAuth2 Redirect URI for your OAuth Application -REDIRECT_URI = 'urn:ietf:wg:oauth:2.0:oob' - -#: Default request HEADERS -HEADERS = {'Content-Type': 'application/json', 'trakt-api-version': '2'} - #: Default path for where to store your trakt.tv API authentication information CONFIG_PATH = os.path.join(os.path.expanduser('~'), '.pytrakt.json') #: Your personal Trakt.tv OAUTH Bearer Token OAUTH_TOKEN = None -# OAuth token validity checked -OAUTH_TOKEN_VALID = None - # Your OAUTH token expiration date OAUTH_EXPIRES_AT = None @@ -80,290 +61,36 @@ session = requests.Session() -def _store(**kwargs): - """Helper function used to store Trakt configurations at ``CONFIG_PATH`` - - :param kwargs: Keyword args to store at ``CONFIG_PATH`` - """ - with open(CONFIG_PATH, 'w') as config_file: - json.dump(kwargs, config_file) - - -def _get_client_info(app_id=False): - """Helper function to poll the user for Client ID and Client Secret - strings - - :return: A 2-tuple of client_id, client_secret - """ - global APPLICATION_ID - print('If you do not have a client ID and secret. Please visit the ' - 'following url to create them.') - print('http://trakt.tv/oauth/applications') - client_id = input('Please enter your client id: ') - client_secret = input('Please enter your client secret: ') - if app_id: - msg = 'Please enter your application ID ({default}): '.format( - default=APPLICATION_ID) - user_input = input(msg) - if user_input: - APPLICATION_ID = user_input - return client_id, client_secret - - -def pin_auth(pin=None, client_id=None, client_secret=None, store=False): - """Generate an access_token from a Trakt API PIN code. - - :param pin: Optional Trakt API PIN code. If one is not specified, you will - be prompted to go generate one - :param client_id: The oauth client_id for authenticating to the trakt API. - :param client_secret: The oauth client_secret for authenticating to the - trakt API. - :param store: Boolean flag used to determine if your trakt api auth data - should be stored locally on the system. Default is :const:`False` for - the security conscious - :return: Your OAuth access token - """ - global OAUTH_TOKEN, CLIENT_ID, CLIENT_SECRET - CLIENT_ID, CLIENT_SECRET = client_id, client_secret - if client_id is None and client_secret is None: - CLIENT_ID, CLIENT_SECRET = _get_client_info(app_id=True) - if pin is None and APPLICATION_ID is None: - print('You must set the APPLICATION_ID of the Trakt application you ' - 'wish to use. You can find this ID by visiting the following ' - 'URL.') - print('https://trakt.tv/oauth/applications') - sys.exit(1) - if pin is None: - print('If you do not have a Trakt.tv PIN, please visit the following ' - 'url and log in to generate one.') - pin_url = 'https://trakt.tv/pin/{id}'.format(id=APPLICATION_ID) - print(pin_url) - pin = input('Please enter your PIN: ') - args = {'code': pin, - 'redirect_uri': REDIRECT_URI, - 'grant_type': 'authorization_code', - 'client_id': CLIENT_ID, - 'client_secret': CLIENT_SECRET} - - response = session.post(''.join([BASE_URL, '/oauth/token']), data=args, timeout=TIMEOUT) - OAUTH_TOKEN = response.json().get('access_token', None) - - if store: - _store(CLIENT_ID=CLIENT_ID, CLIENT_SECRET=CLIENT_SECRET, - OAUTH_TOKEN=OAUTH_TOKEN, APPLICATION_ID=APPLICATION_ID) - return OAUTH_TOKEN - - -def _terminal_oauth_pin(authorization_url): - """Default OAuth callback used for terminal applications. - - :param authorization_url: Predefined url by function `oauth_auth`. URL will - be prompted to you in the terminal - :return: OAuth PIN - """ - print('Please go here and authorize,', authorization_url) - - # Get the authorization verifier code from the callback url - response = input('Paste the Code returned here: ') - return response - - -def oauth_auth(username, client_id=None, client_secret=None, store=False, - oauth_cb=_terminal_oauth_pin): - """Generate an access_token to allow your application to authenticate via - OAuth - - :param username: Your trakt.tv username - :param client_id: Your Trakt OAuth Application's Client ID - :param client_secret: Your Trakt OAuth Application's Client Secret - :param store: Boolean flag used to determine if your trakt api auth data - should be stored locally on the system. Default is :const:`False` for - the security conscious - :param oauth_cb: Callback function to handle the retrieving of the OAuth - PIN. Default function `_terminal_oauth_pin` for terminal auth - :return: Your OAuth access token - """ - global CLIENT_ID, CLIENT_SECRET, OAUTH_TOKEN, OAUTH_REFRESH, OAUTH_EXPIRES_AT - if client_id is None and client_secret is None: - client_id, client_secret = _get_client_info() - CLIENT_ID, CLIENT_SECRET = client_id, client_secret - HEADERS['trakt-api-key'] = CLIENT_ID - - authorization_base_url = urljoin(BASE_URL, '/oauth/authorize') - token_url = urljoin(BASE_URL, '/oauth/token') - - # OAuth endpoints given in the API documentation - oauth = OAuth2Session(CLIENT_ID, redirect_uri=REDIRECT_URI, state=None) - - # Authorization URL to redirect user to Trakt for authorization - authorization_url, _ = oauth.authorization_url(authorization_base_url, - username=username) - - # Calling callback function to get the OAuth PIN - oauth_pin = oauth_cb(authorization_url) - - # Fetch, assign, and return the access token - oauth.fetch_token(token_url, client_secret=CLIENT_SECRET, code=oauth_pin) - OAUTH_TOKEN = oauth.token['access_token'] - OAUTH_REFRESH = oauth.token['refresh_token'] - OAUTH_EXPIRES_AT = oauth.token["created_at"] + oauth.token["expires_in"] - - if store: - _store(CLIENT_ID=CLIENT_ID, CLIENT_SECRET=CLIENT_SECRET, - OAUTH_TOKEN=OAUTH_TOKEN, OAUTH_REFRESH=OAUTH_REFRESH, - OAUTH_EXPIRES_AT=OAUTH_EXPIRES_AT) - return OAUTH_TOKEN - - -def get_device_code(client_id=None, client_secret=None): - """Generate a device code, used for device oauth authentication. - - Trakt docs: https://trakt.docs.apiary.io/#reference/ - authentication-devices/device-code - :param client_id: Your Trakt OAuth Application's Client ID - :param client_secret: Your Trakt OAuth Application's Client Secret - :return: Your OAuth device code. - """ - global CLIENT_ID, CLIENT_SECRET, OAUTH_TOKEN - if client_id is None and client_secret is None: - client_id, client_secret = _get_client_info() - CLIENT_ID, CLIENT_SECRET = client_id, client_secret - HEADERS['trakt-api-key'] = CLIENT_ID - - device_code_url = urljoin(BASE_URL, '/oauth/device/code') - headers = {'Content-Type': 'application/json'} - data = {"client_id": CLIENT_ID} - - device_response = session.post(device_code_url, - json=data, headers=headers, timeout=TIMEOUT).json() - print('Your user code is: {user_code}, please navigate to ' - '{verification_url} to authenticate'.format( - user_code=device_response.get('user_code'), - verification_url=device_response.get('verification_url') - )) - - device_response['requested'] = time.time() - return device_response - - -def get_device_token(device_code, client_id=None, client_secret=None, - store=False): - """ - Trakt docs: https://trakt.docs.apiary.io/#reference/ - authentication-devices/get-token - Response: - { - "access_token": "", - "token_type": "bearer", - "expires_in": 7776000, - "refresh_token": "", - "scope": "public", - "created_at": 1519329051 - } - :return: Information regarding the authentication polling. - :return type: dict - """ - global CLIENT_ID, CLIENT_SECRET, OAUTH_TOKEN, OAUTH_REFRESH, OAUTH_EXPIRES_AT - if client_id is None and client_secret is None: - client_id, client_secret = _get_client_info() - CLIENT_ID, CLIENT_SECRET = client_id, client_secret - HEADERS['trakt-api-key'] = CLIENT_ID - - data = { - "code": device_code, - "client_id": CLIENT_ID, - "client_secret": CLIENT_SECRET - } - - response = session.post( - urljoin(BASE_URL, '/oauth/device/token'), json=data, timeout=TIMEOUT - ) - - # We only get json on success. - if response.status_code == 200: - data = response.json() - OAUTH_TOKEN = data.get('access_token') - OAUTH_REFRESH = data.get('refresh_token') - OAUTH_EXPIRES_AT = data.get("created_at") + data.get("expires_in") - - if store: - _store( - CLIENT_ID=CLIENT_ID, CLIENT_SECRET=CLIENT_SECRET, - OAUTH_TOKEN=OAUTH_TOKEN, OAUTH_REFRESH=OAUTH_REFRESH, - OAUTH_EXPIRES_AT=OAUTH_EXPIRES_AT - ) - - return response - - -def device_auth(client_id=None, client_secret=None, store=False): - """Process for authenticating using device authentication. - - The function will attempt getting the device_id, and provide - the user with a url and code. After getting the device - id, a timer is started to poll periodic for a successful authentication. - This is a blocking action, meaning you - will not be able to run any other code, while waiting for an access token. - - If you want more control over the authentication flow, use the functions - get_device_code and get_device_token. - Where poll_for_device_token will check if the "offline" - authentication was successful. - - :param client_id: Your Trakt OAuth Application's Client ID - :param client_secret: Your Trakt OAuth Application's Client Secret - :param store: Boolean flag used to determine if your trakt api auth data - should be stored locally on the system. Default is :const:`False` for - the security conscious - :return: A dict with the authentication result. - Or False of authentication failed. - """ - error_messages = { - 404: 'Invalid device_code', - 409: 'You already approved this code', - 410: 'The tokens have expired, restart the process', - 418: 'You explicitly denied this code', - } - - success_message = ( - "You've been successfully authenticated. " - "With access_token {access_token} and refresh_token {refresh_token}" - ) - - response = get_device_code(client_id=client_id, - client_secret=client_secret) - device_code = response['device_code'] - interval = response['interval'] - - # No need to check for expiration, the API will notify us. - while True: - response = get_device_token(device_code, client_id, client_secret, - store) - - if response.status_code == 200: - print(success_message.format_map(response.json())) - break +def init(*args, **kwargs): + """Run the auth function specified by *AUTH_METHOD*""" + from trakt.auth import init_auth - elif response.status_code == 429: # slow down - interval *= 2 + return init_auth(AUTH_METHOD, *args, **kwargs) - elif response.status_code != 400: # not pending - print(error_messages.get(response.status_code, response.reason)) - break - time.sleep(interval) +@lru_cache(maxsize=None) +def config(): + from trakt.config import AuthConfig - return response + return AuthConfig(CONFIG_PATH).update( + APPLICATION_ID=APPLICATION_ID, + CLIENT_ID=CLIENT_ID, + CLIENT_SECRET=CLIENT_SECRET, + OAUTH_EXPIRES_AT=OAUTH_EXPIRES_AT, + OAUTH_REFRESH=OAUTH_REFRESH, + OAUTH_TOKEN=OAUTH_TOKEN, + ) -auth_method = { - PIN_AUTH: pin_auth, OAUTH_AUTH: oauth_auth, DEVICE_AUTH: device_auth -} +@lru_cache(maxsize=None) +def api(): + from trakt.api import HttpClient, TokenAuth + client = HttpClient(BASE_URL, session) + auth = TokenAuth(client=client, config=config()) + client.set_auth(auth) -def init(*args, **kwargs): - """Run the auth function specified by *AUTH_METHOD*""" - return auth_method.get(AUTH_METHOD, PIN_AUTH)(*args, **kwargs) + return client class Airs(NamedTuple): @@ -396,258 +123,22 @@ class Comment(NamedTuple): user_rating: str -def _validate_token(s): - """Check if current OAuth token has not expired""" - global OAUTH_TOKEN_VALID - current = datetime.now(tz=timezone.utc) - expires_at = datetime.fromtimestamp(OAUTH_EXPIRES_AT, tz=timezone.utc) - if expires_at - current > timedelta(days=2): - OAUTH_TOKEN_VALID = True - else: - _refresh_token(s) - - -def _refresh_token(s): - """Request Trakt API for a new valid OAuth token using refresh_token""" - global OAUTH_TOKEN, OAUTH_EXPIRES_AT, OAUTH_REFRESH, OAUTH_TOKEN_VALID - s.logger.info("OAuth token has expired, refreshing now...") - url = urljoin(BASE_URL, '/oauth/token') - data = { - 'client_id': CLIENT_ID, - 'client_secret': CLIENT_SECRET, - 'refresh_token': OAUTH_REFRESH, - 'redirect_uri': REDIRECT_URI, - 'grant_type': 'refresh_token' - } - response = session.post(url, json=data, headers=HEADERS, timeout=TIMEOUT) - s.logger.debug('RESPONSE [post] (%s): %s - %s', url, str(response), response.content) - if response.status_code == 200: - data = response.json() - OAUTH_TOKEN = data.get("access_token") - OAUTH_REFRESH = data.get("refresh_token") - OAUTH_EXPIRES_AT = data.get("created_at") + data.get("expires_in") - OAUTH_TOKEN_VALID = True - s.logger.info( - "OAuth token successfully refreshed, valid until {}".format( - datetime.fromtimestamp(OAUTH_EXPIRES_AT, tz=timezone.utc) - ) - ) - _store( - CLIENT_ID=CLIENT_ID, CLIENT_SECRET=CLIENT_SECRET, - OAUTH_TOKEN=OAUTH_TOKEN, OAUTH_REFRESH=OAUTH_REFRESH, - OAUTH_EXPIRES_AT=OAUTH_EXPIRES_AT - ) - elif response.status_code in (401, 400): - from .errors import OAuthRefreshException - raise OAuthRefreshException(response) - elif response.status_code in s.error_map: - raise s.error_map[response.status_code](response) - else: - from .errors import BadRequestException - raise BadRequestException(response) - - -def load_config(): - """Manually load config from json config file.""" - global CLIENT_ID, CLIENT_SECRET, OAUTH_TOKEN, OAUTH_EXPIRES_AT - global OAUTH_REFRESH, APPLICATION_ID, CONFIG_PATH - if (CLIENT_ID is None or CLIENT_SECRET is None) and \ - os.path.exists(CONFIG_PATH): - # Load in trakt API auth data from CONFIG_PATH - with open(CONFIG_PATH) as config_file: - config_data = json.load(config_file) - - if CLIENT_ID is None: - CLIENT_ID = config_data.get('CLIENT_ID', None) - if CLIENT_SECRET is None: - CLIENT_SECRET = config_data.get('CLIENT_SECRET', None) - if OAUTH_TOKEN is None: - OAUTH_TOKEN = config_data.get('OAUTH_TOKEN', None) - if OAUTH_EXPIRES_AT is None: - OAUTH_EXPIRES_AT = config_data.get('OAUTH_EXPIRES_AT', None) - if OAUTH_REFRESH is None: - OAUTH_REFRESH = config_data.get('OAUTH_REFRESH', None) - if APPLICATION_ID is None: - APPLICATION_ID = config_data.get('APPLICATION_ID', None) - - -class Core: - """This class contains all of the functionality required for interfacing - with the Trakt.tv API - """ - - def __init__(self): - """Create a :class:`Core` instance and give it a logger attribute""" - self.logger = logging.getLogger('trakt.core') - - # Get all of our exceptions except the base exception - errs = [getattr(errors, att) for att in errors.__all__ - if att != 'TraktException'] - - # Map HTTP response codes to exception types - self.error_map = {err.http_code: err for err in errs} - self._bootstrapped = False - - def _bootstrap(self): - """Bootstrap your authentication environment when authentication is - needed and if a file at `CONFIG_PATH` exists. - The process is completed by setting the client id header. - """ - - if self._bootstrapped: - return - self._bootstrapped = True - - global OAUTH_TOKEN_VALID, OAUTH_EXPIRES_AT - global OAUTH_REFRESH, OAUTH_TOKEN - - load_config() - # Check token validity and refresh token if needed - if (not OAUTH_TOKEN_VALID and OAUTH_EXPIRES_AT is not None - and OAUTH_REFRESH is not None): - _validate_token(self) - - @staticmethod - def _get_first(f, *args, **kwargs): - """Extract the first value from the provided generator function *f* - - :param f: A generator function to extract data from - :param args: Non keyword args for the generator function - :param kwargs: Keyword args for the generator function - :return: The full url for the resource, a generator, and either a data - payload or `None` - """ - generator = f(*args, **kwargs) - uri = next(generator) - if not isinstance(uri, (str, tuple)): - # Allow properties to safely yield arbitrary data - return uri - if isinstance(uri, tuple): - uri, data = uri - return BASE_URL + uri, generator, data - else: - return BASE_URL + uri, generator, None - - def _handle_request(self, method, url, data=None): - """Handle actually talking out to the trakt API, logging out debug - information, raising any relevant `TraktException` Exception types, - and extracting and returning JSON data - - :param method: The HTTP method we're executing on. Will be one of - post, put, delete, get - :param url: The fully qualified url to send our request to - :param data: Optional data payload to send to the API - :return: The decoded JSON response from the Trakt API - :raises TraktException: If any non-200 return code is encountered - """ - self.logger.debug('%s: %s', method, url) - HEADERS['trakt-api-key'] = CLIENT_ID - HEADERS['Authorization'] = 'Bearer {0}'.format(OAUTH_TOKEN) - self.logger.debug('method, url :: %s, %s', method, url) - if method == 'get': # GETs need to pass data as params, not body - response = session.request(method, url, headers=HEADERS, - params=data, timeout=TIMEOUT) - else: - response = session.request(method, url, headers=HEADERS, - data=json.dumps(data), timeout=TIMEOUT) - self.logger.debug('RESPONSE [%s] (%s): %s', method, url, str(response)) - if response.status_code in self.error_map: - raise self.error_map[response.status_code](response) - elif response.status_code == 204: # HTTP no content - return None - - try: - json_data = json.loads(response.content.decode('UTF-8', 'ignore')) - except JSONDecodeError as e: - raise BadResponseException(response, f"Unable to parse JSON: {e}") - - return json_data - - def get(self, f): - """Perform a HTTP GET request using the provided uri yielded from the - *f* co-routine. The processed JSON results are then sent back to the - co-routine for post-processing, the results of which are then returned - - :param f: Generator co-routine that yields uri, args, and processed - results - :return: The results of the generator co-routine - """ - @wraps(f) - def inner(*args, **kwargs): - self._bootstrap() - resp = self._get_first(f, *args, **kwargs) - if not isinstance(resp, tuple): - # Handle cached property responses - return resp - url, generator, _ = resp - json_data = self._handle_request('get', url) - try: - return generator.send(json_data) - except StopIteration: - return None - return inner - - def delete(self, f): - """Perform an HTTP DELETE request using the provided uri - - :param f: Function that returns a uri to delete to - """ - @wraps(f) - def inner(*args, **kwargs): - self._bootstrap() - generator = f(*args, **kwargs) - uri = next(generator) - url = BASE_URL + uri - self._handle_request('delete', url) - return inner - - def post(self, f): - """Perform an HTTP POST request using the provided uri and optional - args yielded from the *f* co-routine. The processed JSON results are - then sent back to the co-routine for post-processing, the results of - which are then returned - - :param f: Generator co-routine that yields uri, args, and processed - results - :return: The results of the generator co-routine - """ - @wraps(f) - def inner(*args, **kwargs): - self._bootstrap() - url, generator, args = self._get_first(f, *args, **kwargs) - json_data = self._handle_request('post', url, data=args) - try: - return generator.send(json_data) - except StopIteration: - return None - return inner - - def put(self, f): - """Perform an HTTP PUT request using the provided uri and optional args - yielded from the *f* co-routine. The processed JSON results are then - sent back to the co-routine for post-processing, the results of which - are then returned - - :param f: Generator co-routine that yields uri, args, and processed - results - :return: The results of the generator co-routine - """ - @wraps(f) - def inner(*args, **kwargs): - self._bootstrap() - url, generator, args = self._get_first(f, *args, **kwargs) - json_data = self._handle_request('put', url, data=args) - try: - return generator.send(json_data) - except StopIteration: - return None - return inner - - -# Here we can simplify the code in each module by exporting these instance -# method decorators as if they were simple functions. -CORE = Core() -get = CORE.get -post = CORE.post -delete = CORE.delete -put = CORE.put +# Backward compat with 3.x +def delete(f): + from trakt.decorators import delete + return delete(f) + + +def get(f): + from trakt.decorators import get + return get(f) + + +def post(f): + from trakt.decorators import post + return post(f) + + +def put(f): + from trakt.decorators import put + return put(f) diff --git a/trakt/decorators.py b/trakt/decorators.py new file mode 100644 index 0000000..06b4a54 --- /dev/null +++ b/trakt/decorators.py @@ -0,0 +1,115 @@ +"""Decorators to handle HTTP methods magically""" + +__author__ = 'Elan Ruusamäe' + +from functools import wraps + +from trakt.core import api + + +def _get_first(f, *args, **kwargs): + """Extract the first value from the provided generator function *f* + + :param f: A generator function to extract data from + :param args: Non keyword args for the generator function + :param kwargs: Keyword args for the generator function + :return: The full url for the resource, a generator, and either a data + payload or `None` + """ + generator = f(*args, **kwargs) + uri = next(generator) + if not isinstance(uri, (str, tuple)): + # Allow properties to safely yield arbitrary data + return uri + if isinstance(uri, tuple): + uri, data = uri + return uri, generator, data + else: + return uri, generator, None + + +def get(f): + """Perform a HTTP GET request using the provided uri yielded from the + *f* co-routine. The processed JSON results are then sent back to the + co-routine for post-processing, the results of which are then returned + + :param f: Generator co-routine that yields uri, args, and processed + results + :return: The results of the generator co-routine + """ + + @wraps(f) + def inner(*args, **kwargs): + resp = _get_first(f, *args, **kwargs) + if not isinstance(resp, tuple): + # Handle cached property responses + return resp + url, generator, _ = resp + json_data = api().get(url) + try: + return generator.send(json_data) + except StopIteration: + return None + + return inner + + +def delete(f): + """Perform an HTTP DELETE request using the provided uri + + :param f: Function that returns a uri to delete to + """ + + @wraps(f) + def inner(*args, **kwargs): + generator = f(*args, **kwargs) + url = next(generator) + api().delete(url) + + return inner + + +def post(f): + """Perform an HTTP POST request using the provided uri and optional + args yielded from the *f* co-routine. The processed JSON results are + then sent back to the co-routine for post-processing, the results of + which are then returned + + :param f: Generator co-routine that yields uri, args, and processed + results + :return: The results of the generator co-routine + """ + + @wraps(f) + def inner(*args, **kwargs): + url, generator, data = _get_first(f, *args, **kwargs) + json_data = api().post(url, data) + try: + return generator.send(json_data) + except StopIteration: + return None + + return inner + + +def put(f): + """Perform an HTTP PUT request using the provided uri and optional args + yielded from the *f* co-routine. The processed JSON results are then + sent back to the co-routine for post-processing, the results of which + are then returned + + :param f: Generator co-routine that yields uri, args, and processed + results + :return: The results of the generator co-routine + """ + + @wraps(f) + def inner(*args, **kwargs): + url, generator, data = _get_first(f, *args, **kwargs) + json_data = api().put(url, data) + try: + return generator.send(json_data) + except StopIteration: + return None + + return inner