Improve code quality

apns2/credentials.py

@@ -1,29 +1,22 @@
+import time
+import jwt
 
 from hyper import HTTP20Connection
 from hyper.tls import init_context
 
-import jwt
-
-# For creating and comparing the time for the JWT token
-import time
-
-
 DEFAULT_TOKEN_LIFETIME = 3600
 DEFAULT_TOKEN_ENCRYPTION_ALGORITHM = 'ES256'
 
 
 # Abstract Base class. This should not be instantiated directly.
 class Credentials(object):
-
     def __init__(self, ssl_context=None):
         self.__ssl_context = ssl_context
 
     # Creates a connection with the credentials, if available or necessary.
     def create_connection(self, server, port, proto):
         # self.__ssl_context may be none, and that's fine.
-        return HTTP20Connection(server, port,
-                                ssl_context=self.__ssl_context,
-                                force_proto=proto or 'h2')
+        return HTTP20Connection(server, port, ssl_context=self.__ssl_context, force_proto=proto or 'h2')
 
     def get_authorization_header(self, topic):
         return None
@@ -41,15 +34,13 @@ class CertificateCredentials(Credentials):
 # Credentials subclass for JWT token based authentication
 class TokenCredentials(Credentials):
     def __init__(self, auth_key_path, auth_key_id, team_id,
-                 encryption_algorithm=None, token_lifetime=None):
+                 encryption_algorithm=DEFAULT_TOKEN_ENCRYPTION_ALGORITHM,
+                 token_lifetime=DEFAULT_TOKEN_LIFETIME):
         self.__auth_key = self._get_signing_key(auth_key_path)
         self.__auth_key_id = auth_key_id
         self.__team_id = team_id
-        self.__encryption_algorithm = DEFAULT_TOKEN_ENCRYPTION_ALGORITHM if \
-            encryption_algorithm is None else \
-            encryption_algorithm
-        self.__token_lifetime = DEFAULT_TOKEN_LIFETIME if \
-            token_lifetime is None else token_lifetime
+        self.__encryption_algorithm = encryption_algorithm
+        self.__token_lifetime = token_lifetime
 
         # Dictionary of {topic: (issue time, ascii decoded token)}
         self.__topicTokens = {}
@@ -64,34 +55,37 @@ class TokenCredentials(Credentials):
         token = self._get_or_create_topic_token(topic)
         return 'bearer %s' % token
 
-    def _isExpiredToken(self, issueDate):
-        now = time.time()
-        return now < issueDate + DEFAULT_TOKEN_LIFETIME
+    @staticmethod
+    def _is_expired_token(issue_date):
+        return time.time() < issue_date + DEFAULT_TOKEN_LIFETIME
 
-    def _get_or_create_topic_token(self, topic):
-        # dict of topic to issue date and JWT token
-        tokenPair = self.__topicTokens.get(topic)
-        if tokenPair is None or self._isExpiredToken(tokenPair[0]):
-            # Create a new token
-            issuedAt = time.time()
-            tokenDict = {'iss': self.__team_id,
-                         'iat': issuedAt
-                         }
-            headers = {'alg': self.__encryption_algorithm,
-                       'kid': self.__auth_key_id,
-                       }
-            jwtToken = jwt.encode(tokenDict, self.__auth_key,
-                                  algorithm=self.__encryption_algorithm,
-                                  headers=headers).decode('ascii')
-
-            self.__topicTokens[topic] = (issuedAt, jwtToken)
-            return jwtToken
-        else:
-            return tokenPair[1]
-
-    def _get_signing_key(self, key_path):
+    @staticmethod
+    def _get_signing_key(key_path):
         secret = ''
         if key_path:
             with open(key_path) as f:
                 secret = f.read()
         return secret
+
+    def _get_or_create_topic_token(self, topic):
+        # dict of topic to issue date and JWT token
+        token_pair = self.__topicTokens.get(topic)
+        if token_pair is None or self._is_expired_token(token_pair[0]):
+            # Create a new token
+            issued_at = time.time()
+            token_dict = {
+                'iss': self.__team_id,
+                'iat': issued_at,
+            }
+            headers = {
+                'alg': self.__encryption_algorithm,
+                'kid': self.__auth_key_id,
+            }
+            jwt_token = jwt.encode(token_dict, self.__auth_key,
+                                   algorithm=self.__encryption_algorithm,
+                                   headers=headers).decode('ascii')
+
+            self.__topicTokens[topic] = (issued_at, jwt_token)
+            return jwt_token
+        else:
+            return token_pair[1]