Create usosapi.py

First version of the module.

Author: m-strzelczyk

usosapi.py

@@ -0,0 +1,315 @@
+# coding=utf-8
+"""
+This package is part of the USOS API project.
+https://apps.usos.edu.pl/developers/
+"""
+import hashlib
+import os
+import os.path
+import tempfile
+import urllib.request
+import shutil
+import rauth
+import warnings
+import re
+import requests.exceptions
+import logging
+import time
+
+_REQUEST_TOKEN_SUFFIX = 'services/oauth/request_token'
+_AUTHORIZE_SUFFIX = 'services/oauth/authorize'
+_ACCESS_TOKEN_SUFFIX = 'services/oauth/access_token'
+
+SCOPES = 'offline_access'
+
+_LOGGER = logging.getLogger('USOSAPI')
+_DOWNLOAD_LOGGER = logging.getLogger('USOSAPI.download')
+
+
+class USOSAPIException(Exception):
+    pass
+
+
+def download_file(url: str) -> str:
+    """
+    This function is here for convenience. It's useful for downloading
+    for eg. user photos. It blocks until the file is saved on the disk
+    and then returns path of the file.
+    If given url was already downloaded before, it won't be downloaded
+    again (useful when you download user profile photos, most of them
+    are blanks).
+    """
+    md5 = hashlib.md5()
+    file_name, extension = os.path.splitext(url)
+    md5.update(url.encode())
+    file_name = md5.hexdigest() + extension
+    file_dir = os.path.join(tempfile.gettempdir(), 'USOSAPI')
+
+    if not os.path.exists(file_dir):
+        os.mkdir(file_dir)
+    else:
+        if not os.path.isdir(file_dir):
+            shutil.rmtree(file_dir)
+            os.mkdir(file_dir)
+
+    file_name = os.path.join(file_dir, file_name)
+
+    if os.path.exists(file_name):
+        if os.path.isfile(file_name):
+            return file_name
+        else:
+            shutil.rmtree(file_name)
+
+    with urllib.request.urlopen(url) as resp, open(file_name, 'wb') as out:
+        shutil.copyfileobj(resp, out)
+
+    _DOWNLOAD_LOGGER.info('File from {} saved as {}.'.format(url, file_name))
+
+    return file_name
+
+
+class USOSAPIConnection():
+    """
+    This class provides basic functionality required to work with
+    USOS API server. To start communication you need to provide server
+    address and your consumer key/secret pair to the constructor.
+
+    After you create an USOSAPIConnection object with working parameters
+    (check them with test_connection function), you may already use a subset
+    of USOS API services that don't require a valid access key.
+
+    To log in as a specific user you need to get an URL address with
+    get_authorization_url and somehow display it to the user (this module
+    doesn't provide any UI). On the web page, after accepting
+    scopes required by the module, user will receive a PIN code.
+    This code should be passed to authorize_with_pin function to
+    complete the authorization process. After successfully calling the
+    authorize_with_pin function, you will have an authorized_session.
+    """
+    def __init__(self, api_base_address: str, consumer_key: str,
+                 consumer_secret: str):
+        self.base_address = str(api_base_address)
+        if not self.base_address:
+            raise ValueError('Empty USOS API address.')
+        if not self.base_address.startswith('https'):
+            warnings.warn('Insecure protocol in USOS API address. '
+                          'The address should start with https.')
+        if not self.base_address.endswith('/'):
+            self.base_address += '/'
+
+        self.consumer_key = str(consumer_key)
+        self.consumer_secret = str(consumer_secret)
+
+        req_token_url = self.base_address + _REQUEST_TOKEN_SUFFIX
+        authorize_url = self.base_address + _AUTHORIZE_SUFFIX
+        access_token_url = self.base_address + _ACCESS_TOKEN_SUFFIX
+
+        self._service = rauth.OAuth1Service(consumer_key=consumer_key,
+                                            consumer_secret=consumer_secret,
+                                            name='USOSAPI',
+                                            request_token_url=req_token_url,
+                                            authorize_url=authorize_url,
+                                            access_token_url=access_token_url,
+                                            base_url=self.base_address)
+
+        self._request_token_secret = ''
+        self._request_token = ''
+
+        self._authorized_session = None
+        _LOGGER.info('New connection to {} created with key: {} '
+                     'and secret: {}.'.format(api_base_address,
+                                              consumer_key, consumer_secret))
+
+    def _generate_request_token(self):
+        params = {'oauth_callback': 'oob', 'scopes': SCOPES}
+        token_tuple = self._service.get_request_token(params=params)
+        self._request_token, self._request_token_secret = token_tuple
+        _LOGGER.info("New request token generated: {}".format(token_tuple[0]))
+        return
+
+    def is_anonymous(self) -> bool:
+        """
+        Checks if current USOS API session is anonymous.
+        This function assumes that USOS API server connection data
+        (server address, consumer key and consumer secret) are correct.
+        """
+        return self._authorized_session is None
+
+    def is_authorized(self) -> bool:
+        """
+        Checks if current USOS API session is authorized (if you are logged in
+        as specific user). This function assumes that USOS API server
+        connection data (server address, consumer key and consumer secret)
+        are correct.
+        """
+        if self._authorized_session is None:
+            return False
+        try:
+            identity = self.get('services/users/user')
+            return bool(identity['id'])
+        except USOSAPIException:
+            return False
+
+    def test_connection(self) -> bool:
+        """
+        Checks if parameters passed for this object's constructor are correct
+        and if it's possible to connect to the USOS API server.
+        """
+        time_re = '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{6}$'
+        try:
+            anonymous_session = self._service.get_session()
+            now = anonymous_session.get('services/apisrv/now')
+            now = now.json()
+            return bool(re.fullmatch(time_re, now))
+        except Exception as e:
+            _LOGGER.debug('Connection test failed: {}'.format(e))
+            return False
+
+    def get_authorization_url(self) -> str:
+        """
+        Returns an URL address that user has to visit using some
+        internet browser to obtain a PIN code required for authorization.
+        Every time you call this function, a new request token is generated,
+        so only PIN code acquired with the last generated address will allow
+        successful authorization.
+        """
+        self._generate_request_token()
+        url = self._service.get_authorize_url(self._request_token)
+        _LOGGER.info('New authorization URL generated: {}'.format(url))
+        return url
+
+    def authorize_with_pin(self, pin: str):
+        """
+        Call this function after user has obtained PIN code from website
+        which address was generated by the set_authorization_url function.
+        Remember that only PIN code from the last generated address will work.
+
+        Will raise USOSAPIException if the PIN is incorrect.
+        """
+        if not(self._request_token and self._request_token_secret):
+            raise USOSAPIException('Request token not initialized. '
+                                   'Use get_authorization_url to generate '
+                                   'the token.')
+
+        rt = self._request_token
+        rts = self._request_token_secret
+        params = {'oauth_verifier': pin}
+
+        _LOGGER.debug('Trying to authorize request token {} '
+                      'with PIN code: {}'.format(self._request_token, pin))
+
+        try:
+            self._authorized_session = \
+                self._service.get_auth_session(rt, rts, params=params)
+        except KeyError:
+            response = self._service.get_raw_access_token(rt, rts,
+                                                          params=params)
+            text = response.json()
+            if isinstance(text, dict) and 'message' in text:
+                text = text['message']
+            _LOGGER.info('Authorization failed, response message: ' + text)
+            raise USOSAPIException(text)
+        at = self.get_access_data()[0]
+        _LOGGER.info('Authorization successful, received access token: ' + at)
+
+    def get_access_data(self) -> tuple:
+        """
+        Returns a tuple of access token and access token secret.
+        You can save them somewhere and later use them to resume
+        an authorized session.
+        """
+        if self.is_anonymous():
+            raise USOSAPIException('Connection not yet authorized.')
+        at = self._authorized_session.access_token
+        ats = self._authorized_session.access_token_secret
+        return at, ats
+
+    def set_access_data(self, access_token: str,
+                        access_token_secret: str) -> bool:
+        """
+        Using this function you can resume an authorized session.
+        Although this module requires offline_access scope from users
+        it is still possible, that the session won't be valid when it's
+        resumed. Check return value to make sure if provided access
+        pair was valid.
+        """
+        self._authorized_session = self._service.get_session()
+        self._authorized_session.access_token = access_token
+        self._authorized_session.access_token_secret = access_token_secret
+
+        if not self.is_authorized():
+            self._authorized_session = None
+            _LOGGER.info("Access token {} is invalid.".format(access_token))
+            return False
+
+        _LOGGER.info('New access token ({}) and secret ({}) '
+                     'set.'.format(access_token, access_token_secret))
+        return True
+
+    def get(self, service: str, **kwargs):
+        """
+        General use function to retrieve data from USOS API server.
+        Although it is called 'get' it will issue a POST request.
+        It's arguments are service name and an optional set of keyword
+        arguments, that will be passed as parameters of the request.
+
+        Return type depends on the called service. It will usually be
+        a dictionary or a string.
+        """
+        session = self._service.get_session()
+        if self._authorized_session is not None:
+            session = self._authorized_session
+
+        start = time.time()
+        response = session.post(service, params=kwargs, data={})
+        ex_time = time.time() - start
+
+        if not response.ok:
+            try:
+                _LOGGER.info('{} ({}) FAILED: [{}] {}'
+                             ''.format(service, repr(kwargs),
+                                       response.status_code, response.text))
+                response.raise_for_status()
+            except requests.exceptions.HTTPError as e:
+                if response.status_code == 401:
+                    raise USOSAPIException('HTTP 401: Unauthorized. Your '
+                                           'access key probably expired.')
+                if response.status_code == 400:
+                    msg = response.text
+                    raise USOSAPIException('HTTP 400: Bad request: ' + msg)
+                raise e
+
+        _LOGGER.info("{} ({}) {:f}s".format(service, repr(kwargs),
+                                            ex_time))
+        _LOGGER.debug("{} ({}) -> {}".format(response.url, repr(kwargs),
+                                             response.text))
+
+        return response.json()
+
+    def logout(self):
+        """
+        This function results in revoking currently used access key
+        and closing the authenticated session.
+        You can safely call this method multiple times.
+        """
+        if self._authorized_session is None:
+            return
+
+        at = self.get_access_data()[0]
+        self.get('services/oauth/revoke_token')
+        _LOGGER.debug('Access token {} revoked.'.format(at))
+        self._authorized_session = None
+
+    def current_identity(self):
+        """
+        Returns a dictionary containing following keys: first_name,
+        last_name and id.
+
+        If current session is anonymous it will raise USOSAPIException.
+        """
+        try:
+            data = self.get('services/users/user')
+            return data
+        except USOSAPIException:
+            raise USOSAPIException('Trying to get identity of an unauthorized'
+                                   ' session.')