Commits

Paul Tan committed 85f52af

Removed Tokens class.
New Credentials class which is returned from OAuth1Server.{temp_cred,token_cred} does the same job.

Comments (0)

Files changed (3)

oauth1client/__init__.py

 
 urlquote = _partial(urlquote, safe='')
 
-TemporaryCredentials = _namedtuple("TemporaryCredentials",
-        ("token", "secret"))
+""" Credentials """
 
-TokenCredentials = _namedtuple("TokenCredentials", ("token", "secret"))
+class Credentials(UserDict):
+    #OKAY!
+    def __init__(self, dict = None, provider = None, **kwargs):
+        self.provider = provider
+        super().__init__(dict = dict, **kwargs)
 
+    oauth_token = property(lambda x: x["oauth_token"])
+    oauth_token_secret = property(lambda x: x["oauth_token_secret"])
 
-#Applying oauth parameters to request
+    @classmethod
+    def loads(self, input):
+        x = json.loads(input)
+        if "provider" in x:
+            del x["provider"]
+        return cls(dict = x)
+
+    @classmethod
+    def load(cls, file):
+        if isinstance(file, str):
+            file = open(file, "r")
+        x = json.load(file)
+        if "provider" in x:
+            del x["provider"]
+        return cls(dict = x)
+
+    def dumps(self):
+        x = {"oauth_token": self.oauth_token, 
+                "oauth_token_secret": self.oauth_token_secret}
+        if self.provider and self.provider.name:
+            x["provider"] = self.provider.name
+        return json.dumps(x, indent = 4)
+
+    def dump(self, file):
+        if isinstance(file, str):
+            makedirs(dirname(file), exist_ok = True)
+            file = open(file, "wb")
+        x = {"oauth_token": self.oauth_token,
+                "oauth_token_secret": self.oauth_token_secret}
+        if self.provider and self.provider.name:
+            x["provider"] = self.provider.name
+        return json.dump(x, file, indent = 4)
+
+    @classmethod
+    def load_profile(cls, provider, profile = "default"):
+        if isinstance(provider, str):
+            provider_name = provider
+        else:
+            provider_name = provider.name
+        x = cls.load(token_store_open(provider_name, profile, "r"))
+        if not isinstance(provider, str):
+            x.provider = provider
+        return x
+
+    def dump_profile(self, profile = "default"):
+        return self.dump(token_store_open(self.provider.name, profile, "w"))
+
+    def oauth_req(self, req):
+        return self.provider.oauth(req, self)
+
+
+""" Applying oauth parameters to request """
 def req_apply_oauth_header(req: Request, oauth_params) -> Request:
     x = ",".join(['{0}="{1}"'.format(urlquote(str(k)), urlquote(str(v)))
             for k, v in oauth_params.items()])
             "oauth_nonce": nonce(),
             "oauth_signature_method": signature_method.name,
             "oauth_version": "1.0"}
-    if credentials: p["oauth_token"] = credentials.token
+    if credentials: 
+        p["oauth_token"] = credentials.oauth_token
     p.update(params)
-    signature = signature_method(req, p, credentials.secret if credentials else "")
+    signature = signature_method(req, p, credentials.oauth_token_secret if credentials else "")
     p["oauth_signature"] = signature
     x = req_oauth(req, p)
     return x
 
 
 """ Token Storage """
+#API Rationale: These are global functions because they should not be overriden
+#in order to keep compatibility between libraries.
+
 
 def _token_store_path_posix(name, profile = "default"):
     return joinpath(expanduser("~/.local/share/oauth1/profiles"), profile, name)
     makedirs(dirname(path), exist_ok = True)
     return open(path, mode)
 
-class Token:
-    def __init__(self, server, token_credentials = None):
-        self.server = server
-        self.token_credentials = token_credentials
-        self.modified = True
-
-    def apply_req(self, r):
-        return self.server.oauth(r, self.token_credentials)
-
-    def loads(self, input):
-        import pickle
-        self.token_credentials = pickle.loads(input)
-        self.modified = False
-
-    def load(self, file):
-        import pickle
-        if isinstance(file, str):
-            file = open(file, "rb")
-        self.token_credentials = pickle.load(file)
-        self.modified = False
-
-    def load_profile(self, profile = "default", *, name = None):
-        if name is None:
-            name = self.server.name
-        return self.load(token_store_open(name, profile, "rb"))
-
-    def dumps(self):
-        import pickle
-        return pickle.dumps(self.token_credentials)
-    
-    def dump(self, file):
-        import pickle
-        if self.modified:
-            if isinstance(file, str):
-                makedirs(dirname(file), exist_ok = True)
-                file = open(file, "wb")
-            pickle.dump(self.token_credentials, file)
-            return True
-        else:
-            return False
-
-    def dump_profile(self, profile = "default", *, name = None):
-        """Dump the token into the global store"""
-        if name is None:
-            name = self.server.name
-        return self.dump(token_store_open(name, profile, "wb"))
 
 """ User Handlers """
 
         If not, if oauth version is 1.0a or RFC, 
         return oob. If oauth version is 1.0, return None
         (which would mean no callback)"""
-        return "oob"
         from multiprocessing import Process, Pipe
         from urllib.parse import urlsplit, urlunsplit
         if redirect_uri is None:
             except:
                 pass
 
+
 """ Server """
 
-class AuthorizationEndpointResponse(UserDict):
+class TempCredentials(Credentials):
+    pass
+
+class AuthResponse(UserDict):
     oauth_token = property(lambda x: x["oauth_token"], doc = ( "The temporary "
         "credential token which was authorized."))
     @property
         "The verification code"
         return self["oauth_verifier"] if "oauth_verifier" in self else None
 
+class TokenCredentials(Credentials):
+    pass
+
+ServerCred = _namedtuple("ServerCred", ("server", "cred"))
 
 class OAuth1Server:
     callback = "http://localhost"
             token_cred_endpoint = None, 
             token_cred_endpoint_method = "POST",
             auth_endpoint = None,
-            callback = "http://localhost"):
+            callback = "http://localhost",
+            name = None):
         for x in ("client_id", "signature_method", "temp_cred_endpoint",
                 "temp_cred_endpoint_method", "token_cred_endpoint",
                 "token_cred_endpoint_method", "auth_endpoint",
-                "callback"):
+                "callback", "name"):
             setattr(self, x, locals()[x])
 
     def oauth(self, req, credentials = None, params = {}):
 
     def temp_cred_parse_resp(self, resp):
         x = dict((k, v[0]) for k, v in parse_qs(resp.text).items())
-        return TemporaryCredentials(x["oauth_token"], x["oauth_token_secret"])
+        return TempCredentials(dict = x, provider = self)
 
     def temp_cred(self, oauth_callback = None, **kwargs):
         if oauth_callback is None: 
     """ Authorization Endpoint """
 
     def auth_userreq(self, temp_cred, **kwargs):
-        kwargs.update({"oauth_token": temp_cred.token})
+        kwargs.update({"oauth_token": temp_cred.oauth_token})
         return apply_query_to_url(self.auth_endpoint, kwargs)
     
     def auth_parse_userresp(self, redirect_url):
         p = dict((k, v[0]) for k, v in parse_qs(urlsplit(redirect_url).query).items())
-        return AuthorizationEndpointResponse(dict = p)
+        return AuthResponse(dict = p)
 
     """ Token Credentials Endpoint """
 
 
     def token_cred_parse_resp(self, resp):
         x = dict((k, v[0]) for k, v in parse_qs(resp.text).items())
-        return TokenCredentials(x["oauth_token"], x["oauth_token_secret"])
+        return TokenCredentials(dict = x, provider = self)
 
     def token_cred(self, temp_cred, oauth_verifier):
         req = self.token_cred_req(temp_cred, oauth_verifier)
         r.raise_for_status()
         return self.token_cred_parse_resp(r)
 
-    def basic_flow(self, oauth_callback = None, *, handler = DefaultUserHandler,
-            token_class = Token):
+    def basic_flow(self, oauth_callback = None, *, handler = DefaultUserHandler):
         if oauth_callback is None:
             oauth_callback = self.callback
         with handler() as x:
             else:
                 redirected_url = x.authorization(url)
                 code = self.auth_parse_userresp(redirected_url).oauth_verifier
-            return token_class(self, self.token_cred(temp_cred, code))
+            return self.token_cred(temp_cred, code)
 
-    def load_profile(self, profile = "default", *, token_class = Token):
+    """ Credentials Storage """
+
+    def load_profile(self, profile = "default"):
         """Load a Token from a profile."""
-        x = token_class(self)
-        x.load_profile(profile)
+        x = Credentials.load_profile(self, profile)
         return x
 
 class LegacyOAuth1Server(OAuth1Server):
         r = Request(self.token_cred_endpoint_method, self.token_cred_endpoint)
         return self.oauth(r, temp_cred)
 
-    def basic_flow(self, oauth_callback = None, *, handler = DefaultUserHandler,
-            token_class = Token):
+    def basic_flow(self, oauth_callback = None, *, handler = DefaultUserHandler):
         if oauth_callback is None:
             oauth_callback = self.callback
         with handler() as x:
                 result = x.authorization(url)
             if not result:
                 raise Exception("User cancel")
-            return token_class(self, self.token_cred(temp_cred, None))
+            return self.token_cred(temp_cred, None)
 
 
 def get_all_file_paths_in_path(path: str):
             raise ValueError("Unknown oAuth version {!r}".format(version))
         x = dict((k, v) for k, v in x.items() if not (k.startswith("_") or 
             k in ["client_secret", "resources", "version"]))
+        x["name"] = name
         x.update(kwargs)
         y = cls(**x)
-        y.name = name
         return y
 
     def url(self, url, **kwargs):
         else:
             raise KeyError(url)
 
+
 def load_config(paths = config_search_paths):
     z = get_all_file_paths_in_paths(paths)
     def load_file(x):

oauth1client/curl.py

         
 
 class CurlWrapper(SubprocessWrapper):
-    def __init__(self, token):
-        self.token = token
+    def __init__(self, cred):
+        self.cred = cred
 
     @staticmethod
     def req_to_curl_args(req):
         else:
             method = "GET"
         req = Request(method = method, url = url, data = data, headers = {}, cookies = {})
-        if self.token:
-            req = self.token.apply_req(req)
+        if self.cred:
+            req = self.cred.oauth_req(req)
         return before_rest + rest + self.req_to_curl_args(req)
 
     def main(self, args = None):

oauth1client/requests.py

 import requests.auth
 from oauth1client import Request
 
-class OAuth1Auth(requests.auth.AuthBase):
-    def __init__(self, token_store):
-        self.token_store = token_store
+class OAuth1(requests.auth.AuthBase):
+    def __init__(self, cred):
+        self.cred = cred
     def __call__(self, x):
         if x.files:
             data = {}
             replace_data = True
             data = x.data
         y = Request(x.method, x.full_url, data, x.headers, x.cookies)
-        y = self.token_store.apply_req(y)
+        y = self.cred.oauth_req(y)
         x.method = y.method
         x.url = y.url
         if replace_data: