feat: bearer token client

twilio/http/bearer_token_http_client.py

@@ -2,16 +2,18 @@
 import jwt
 
 from twilio.base.version import Version
-from twilio.http.orgs_token_manager import OrgTokenManager
+from twilio.http.token_manager import TokenManager
 from twilio.twilio_bearer_token_auth import TwilioBearerTokenAuth
 
 
 class BearerTokenHTTPClient:
-    def __init__(self, orgs_token_manager: OrgTokenManager):
+    def __init__(self, orgs_token_manager: TokenManager):
         self.orgs_token_manager = orgs_token_manager
 
     def get_headers(self, version: Version):
-        if TwilioBearerTokenAuth.get_access_token() is None or self.is_token_expired(TwilioBearerTokenAuth.get_access_token()):
+        if TwilioBearerTokenAuth.get_access_token() is None or self.is_token_expired(
+            TwilioBearerTokenAuth.get_access_token()
+        ):
             access_token = self.orgs_token_manager.fetch_access_token(version)
             TwilioBearerTokenAuth.init(access_token)
         else:
@@ -21,7 +23,7 @@ def get_headers(self, version: Version):
 
     def is_token_expired(self, token):
         decoded_jwt = jwt.decode(token, options={"verify_signature": True})
-        expires_at = decoded_jwt.get('exp')
+        expires_at = decoded_jwt.get("exp")
         # Add a buffer of 30 seconds
         buffer_seconds = 30
         buffer_expires_at = expires_at - buffer_seconds

twilio/http/no_auth_http_client.py

@@ -1,5 +1,4 @@
 class NoAuthHTTPClient:
-
     def get_headers(self):
         headers = {}
         return headers

twilio/http/orgs_token_manager.py

@@ -1,13 +1,24 @@
-
 from twilio.base.version import Version
+from twilio.http.token_manager import TokenManager
 from twilio.rest.preview_iam.organizations.token import TokenList
 
-class OrgTokenManager():
+
+class OrgTokenManager(TokenManager):
     """
     Orgs Token Manager
     """
 
-    def __init__(self, grant_type: str, client_id: str, client_secret: str, code: str = None, redirect_uri: str = None, audience: str = None, refreshToken: str = None, scope: str = None):
+    def __init__(
+        self,
+        grant_type: str,
+        client_id: str,
+        client_secret: str,
+        code: str = None,
+        redirect_uri: str = None,
+        audience: str = None,
+        refreshToken: str = None,
+        scope: str = None,
+    ):
         self.grant_type = grant_type
         self.client_id = client_id
         self.client_secret = client_secret
@@ -26,6 +37,6 @@ def fetch_access_token(self, version: Version):
             code=self.code,
             redirect_uri=self.redirect_uri,
             audience=self.audience,
-            scope=self.scope
+            scope=self.scope,
         )
         return token_instance.access_token

twilio/http/token_manager.py

@@ -1,18 +1,7 @@
 from twilio.base.version import Version
-from twilio.http.orgs_token_manager import OrgTokenManager
-class TokenManager:
-    org_token_manager = None
 
-    @classmethod
-    def init(cls, grant_type: str, client_id: str, client_secret: str, code: str = None, redirect_uri: str = None, audience: str = None, refreshToken: str = None, scope: str = None, tokenManager: OrgTokenManager = None):
-        if tokenManager is None:
-            cls.org_token_manager = OrgTokenManager(grant_type, client_id, client_secret, code, redirect_uri, audience, refreshToken, scope)
-        else:
-            cls.org_token_manager = tokenManager
 
-    @classmethod
-    def get_token_manager(cls, version: Version):
-        if cls.org_token_manager is None:
-            raise Exception("Token Manager not initialized")
-        cls.org_token_manager.version = version
-        return cls.org_token_manager
+class TokenManager:
+
+    def fetch_access_token(self, version: Version):
+        pass

twilio/http/token_manager_initializer.py

@@ -0,0 +1,16 @@
+from twilio.http.token_manager import TokenManager
+
+
+class TokenManagerInitializer:
+
+    org_token_manager = None
+
+    @classmethod
+    def set_token_manager(cls, token_manager: TokenManager):
+        cls.org_token_manager = token_manager
+
+    @classmethod
+    def get_token_manager(cls):
+        if cls.org_token_manager is None:
+            raise Exception('Token Manager not initialized')
+        return cls.org_token_manager