Add access_scopes attribute to Session

Author: rezaansyed

shopify/api_access.py

@@ -1,4 +1,12 @@
 import re
+import sys
+
+
+def basestring_type():
+    if sys.version_info[0] < 3:  # Backwards compatibility for python < v3.0.0
+        return basestring
+    else:
+        return str
 
 
 class ApiAccessError(Exception):
@@ -12,7 +20,7 @@ class ApiAccess:
     IMPLIED_SCOPE_RE = re.compile(r"\A(?P<unauthenticated>unauthenticated_)?write_(?P<resource>.*)\Z")
 
     def __init__(self, scopes):
-        if type(scopes) == str:
+        if isinstance(scopes, basestring_type()):
             scopes = scopes.split(self.SCOPE_DELIMITER)
 
         self.__store_scopes(scopes)
@@ -32,7 +40,6 @@ def __eq__(self, other):
     def __store_scopes(self, scopes):
         sanitized_scopes = frozenset(filter(None, [scope.strip() for scope in scopes]))
         self.__validate_scopes(sanitized_scopes)
-
         implied_scopes = frozenset(self.__implied_scope(scope) for scope in sanitized_scopes)
         self._compressed_scopes = sanitized_scopes - implied_scopes
         self._expanded_scopes = sanitized_scopes.union(implied_scopes)

shopify/session.py

@@ -10,6 +10,7 @@
 import re
 from contextlib import contextmanager
 from six.moves import urllib
+from shopify.api_access import ApiAccess
 from shopify.api_version import ApiVersion, Release, Unstable
 import six
 
@@ -45,10 +46,11 @@ def temp(cls, domain, version, token):
         yield
         shopify.ShopifyResource.activate_session(original_session)
 
-    def __init__(self, shop_url, version=None, token=None):
+    def __init__(self, shop_url, version=None, token=None, access_scopes=None):
         self.url = self.__prepare_url(shop_url)
         self.token = token
         self.version = ApiVersion.coerce_to_version(version)
+        self.access_scopes = access_scopes
         return
 
     def create_permission_url(self, scope, redirect_uri, state=None):
@@ -72,7 +74,10 @@ def request_token(self, params):
         response = urllib.request.urlopen(request)
 
         if response.code == 200:
-            self.token = json.loads(response.read().decode("utf-8"))["access_token"]
+            json_payload = json.loads(response.read().decode("utf-8"))
+            self.token = json_payload["access_token"]
+            self.access_scopes = json_payload["scope"]
+
             return self.token
         else:
             raise Exception(response.msg)
@@ -89,6 +94,17 @@ def site(self):
     def valid(self):
         return self.url is not None and self.token is not None
 
+    @property
+    def access_scopes(self):
+        return self._access_scopes
+
+    @access_scopes.setter
+    def access_scopes(self, scopes):
+        if scopes is None or type(scopes) == ApiAccess:
+            self._access_scopes = scopes
+        else:
+            self._access_scopes = ApiAccess(scopes)
+
     @classmethod
     def __prepare_url(cls, url):
         if not url or (url.strip() == ""):

test/session_test.py

@@ -208,7 +208,7 @@ def test_param_validation_of_param_values_with_lists(self):
         }
         self.assertEqual(True, shopify.Session.validate_hmac(params))
 
-    def test_return_token_if_hmac_is_valid(self):
+    def test_return_token_and_scope_if_hmac_is_valid(self):
         shopify.Session.secret = "secret"
         params = {"code": "any-code", "timestamp": time.time()}
         hmac = shopify.Session.calculate_hmac(params)
@@ -218,12 +218,13 @@ def test_return_token_if_hmac_is_valid(self):
             None,
             url="https://localhost.myshopify.com/admin/oauth/access_token",
             method="POST",
-            body='{"access_token" : "token"}',
+            body='{"access_token" : "token", "scope": "read_products,write_orders"}',
             has_user_agent=False,
         )
         session = shopify.Session("http://localhost.myshopify.com", "unstable")
         token = session.request_token(params)
         self.assertEqual("token", token)
+        self.assertEqual(shopify.ApiAccess("read_products,write_orders"), session.access_scopes)
 
     def test_raise_error_if_hmac_is_invalid(self):
         shopify.Session.secret = "secret"
@@ -257,6 +258,32 @@ def test_raise_error_if_timestamp_is_too_old(self):
             session = shopify.Session("http://localhost.myshopify.com", "unstable")
             session = session.request_token(params)
 
+    def test_access_scopes_are_nil_by_default(self):
+        session = shopify.Session("testshop.myshopify.com", "unstable", "any-token")
+        self.assertIsNone(session.access_scopes)
+
+    def test_access_scopes_when_valid_scopes_passed_in(self):
+        session = shopify.Session(
+            shop_url="testshop.myshopify.com",
+            version="unstable",
+            token="any-token",
+            access_scopes="read_products, write_orders",
+        )
+
+        expected_access_scopes = shopify.ApiAccess("read_products, write_orders")
+        self.assertEqual(expected_access_scopes, session.access_scopes)
+
+    def test_access_scopes_set_with_api_access_object_passed_in(self):
+        session = shopify.Session(
+            shop_url="testshop.myshopify.com",
+            version="unstable",
+            token="any-token",
+            access_scopes=shopify.ApiAccess("read_products, write_orders"),
+        )
+
+        expected_access_scopes = shopify.ApiAccess("read_products, write_orders")
+        self.assertEqual(expected_access_scopes, session.access_scopes)
+
     def normalize_url(self, url):
         scheme, netloc, path, query, fragment = urllib.parse.urlsplit(url)
         query = "&".join(sorted(query.split("&")))