diff --git a/plexauth.py b/plexauth.py deleted file mode 100755 index f09e8e3..0000000 --- a/plexauth.py +++ /dev/null @@ -1,84 +0,0 @@ -"""Handle a Plex.tv authorization flow to obtain an access token.""" -import aiohttp -from asyncio import sleep -from datetime import datetime, timedelta -import urllib.parse -import uuid - -__version__ = '0.0.5' - -CODES_URL = 'https://plex.tv/api/v2/pins.json?strong=true' -AUTH_URL = 'https://app.plex.tv/auth#!?{}' -TOKEN_URL = 'https://plex.tv/api/v2/pins/{}' - -class PlexAuth(): - - def __init__(self, payload, session=None, headers=None): - '''Create PlexAuth instance.''' - self.client_identifier = str(uuid.uuid4()) - self._code = None - self._headers = headers - self._identifier = None - self._payload = payload - self._payload['X-Plex-Client-Identifier'] = self.client_identifier - - self._local_session = False - self._session = session - if session is None: - self._session = aiohttp.ClientSession() - self._local_session = True - - async def initiate_auth(self): - '''Request codes needed to create an auth URL. Starts external timeout.''' - async with self._session.post(CODES_URL, data=self._payload, headers=self._headers) as resp: - response = await resp.json() - self._code = response['code'] - self._identifier = response['id'] - - def auth_url(self, forward_url=None): - '''Return an auth URL for the user to follow.''' - parameters = { - 'clientID': self.client_identifier, - 'code': self._code, - } - if forward_url: - parameters['forwardUrl'] = forward_url - - url = AUTH_URL.format(urllib.parse.urlencode(parameters)) - return url - - async def request_auth_token(self): - '''Request an auth token from Plex.''' - token_url = TOKEN_URL.format(self._code) - payload = dict(self._payload) - payload['Accept'] = 'application/json' - async with self._session.get(TOKEN_URL.format(self._identifier), headers=payload) as resp: - response = await resp.json() - token = response['authToken'] - return token - - async def token(self, timeout=60): - '''Poll Plex endpoint until a token is retrieved or times out.''' - token = None - wait_until = datetime.now() + timedelta(seconds=timeout) - break_loop = False - while not break_loop: - await sleep(3) - token = await self.request_auth_token() - if token or wait_until < datetime.now(): - break_loop = True - - return token - - async def close(self): - """Close open client session.""" - if self._local_session: - await self._session.close() - - async def __aenter__(self): - """Async enter.""" - return self - - async def __aexit__(self, *exc_info): - """Async exit.""" - await self.close() diff --git a/plexauth/__init__.py b/plexauth/__init__.py new file mode 100644 index 0000000..290b052 --- /dev/null +++ b/plexauth/__init__.py @@ -0,0 +1,3 @@ +from .plexauth import PlexAuth + +__all__ = ["PlexAuth"] diff --git a/plexauth/plexauth.py b/plexauth/plexauth.py new file mode 100644 index 0000000..7c01e1a --- /dev/null +++ b/plexauth/plexauth.py @@ -0,0 +1,97 @@ +"""Handle a Plex.tv authorization flow to obtain an access token.""" + +import urllib.parse +import uuid +from asyncio import sleep +from datetime import datetime, timedelta +from typing import Any, Dict, Optional + +import aiohttp + +__version__ = "0.0.5" + +CODES_URL = "https://plex.tv/api/v2/pins.json?strong=true" +AUTH_URL = "https://app.plex.tv/auth#!?{}" +TOKEN_URL = "https://plex.tv/api/v2/pins/{}" + + +class PlexAuth: + def __init__( + self, + payload: Dict[str, str], + session: Optional[aiohttp.ClientSession] = None, + headers: Any = None, + ): + """Create PlexAuth instance.""" + self.client_identifier = str(uuid.uuid4()) + self._code = None + self._headers = headers + self._identifier = None + self._payload = payload + self._payload["X-Plex-Client-Identifier"] = self.client_identifier + + self._local_session = False + if session is None: + session = aiohttp.ClientSession() + self._local_session = True + self._session = session + + async def initiate_auth(self): + """Request codes needed to create an auth URL. + Starts external timeout. + """ + async with self._session.post( + CODES_URL, data=self._payload, headers=self._headers + ) as resp: + response = await resp.json() + self._code = response["code"] + self._identifier = response["id"] + + def auth_url(self, forward_url: Optional[str] = None): + """Return an auth URL for the user to follow.""" + parameters = { + "clientID": self.client_identifier, + "code": self._code, + } + if forward_url: + parameters["forwardUrl"] = forward_url + + url = AUTH_URL.format(urllib.parse.urlencode(parameters)) + return url + + async def request_auth_token(self): + """Request an auth token from Plex.""" + payload = dict(self._payload) + payload["Accept"] = "application/json" + async with self._session.get( + TOKEN_URL.format(self._identifier), headers=payload + ) as resp: + response = await resp.json() + token: str = response["authToken"] + return token + + async def token(self, timeout: int = 60): + """Poll Plex endpoint until a token is retrieved or times out.""" + token = None + wait_until = datetime.now() + timedelta(seconds=timeout) + break_loop = False + while not break_loop: + await sleep(3) + token = await self.request_auth_token() + if token or wait_until < datetime.now(): + break_loop = True + + return token + + async def close(self): + """Close open client session.""" + if self._local_session: + await self._session.close() + + async def __aenter__(self): + """Async enter.""" + return self + + async def __aexit__(self, *_): + """Async exit.""" + await self.close() diff --git a/plexauth/py.typed b/plexauth/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/setup.py b/setup.py index c37ddde..fed9552 100644 --- a/setup.py +++ b/setup.py @@ -1,28 +1,33 @@ from setuptools import setup -with open('README.md') as f: +with open("README.md") as f: long_description = f.read() -VERSION="0.0.6" +VERSION = "0.0.6" setup( - name='plexauth', + name="plexauth", version=VERSION, - description='Handles the authorization flow to obtain tokens from Plex.tv via external redirection.', + description=( + "Handles the authorization flow to obtain tokens from Plex.tv via" + " external redirection." + ), long_description=long_description, - long_description_content_type='text/markdown', - url='https://github.com/jjlawren/python-plexauth/', - license='MIT', - author='Jason Lawrence', - author_email='jjlawren@users.noreply.github.com', - platforms='any', - py_modules=['plexauth'], - install_requires=['aiohttp'], + long_description_content_type="text/markdown", + url="https://github.com/jjlawren/python-plexauth/", + license="MIT", + author="Jason Lawrence", + author_email="jjlawren@users.noreply.github.com", + platforms="any", + py_modules=["plexauth"], + install_requires=["aiohttp"], classifiers=[ - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', - 'Programming Language :: Python :: 3.5', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', - ] + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3.5", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + ], + package_data={"plexauth": ["py.typed"]}, + packages=["plexauth"], )