added helper methods for path_root header

Author: rob

dropbox/dropbox.py

@@ -22,13 +22,19 @@
     AuthError_validator,
     RateLimitError_validator,
 )
+from .common import (
+    PathRoot,
+    PathRoot_validator,
+    PathRootError_validator
+)
 from .base import DropboxBase
 from .base_team import DropboxTeamBase
 from .exceptions import (
     ApiError,
     AuthError,
     BadInputError,
     HttpError,
+    PathRootError,
     InternalServerError,
     RateLimitError,
 )
@@ -42,6 +48,9 @@
     pinned_session,
 )
 
+PATH_ROOT_HEADER = 'Dropbox-API-Path-Root'
+HTTP_STATUS_INVALID_PATH_ROOT = 422
+
 class RouteResult(object):
     """The successful result of a call to a route."""
 
@@ -182,6 +191,26 @@ def __init__(self,
 
         self._timeout = timeout
 
+    def clone(
+            self,
+            oauth2_access_token=None,
+            max_retries_on_error=None,
+            max_retries_on_rate_limit=None,
+            user_agent=None,
+            session=None,
+            headers=None,
+            timeout=None):
+
+        return self.__class__(
+            oauth2_access_token or self._oauth2_access_token,
+            max_retries_on_error or self._max_retries_on_error,
+            max_retries_on_rate_limit or self._max_retries_on_rate_limit,
+            user_agent or self._user_agent,
+            session or self._session,
+            headers or self._headers,
+            timeout or self._timeout
+        )
+
     def request(self,
                 route,
                 namespace,
@@ -421,6 +450,10 @@ def request_json_string(self,
             err = stone_serializers.json_compat_obj_decode(
                 AuthError_validator, r.json()['error'])
             raise AuthError(request_id, err)
+        elif r.status_code == HTTP_STATUS_INVALID_PATH_ROOT:
+            err = stone_serializers.json_compat_obj_decode(
+                PathRootError_validator, r.json()['error'])
+            raise PathRootError(request_id, err)
         elif r.status_code == 429:
             err = None
             if r.headers.get('content-type') == 'application/json':
@@ -479,13 +512,44 @@ def _save_body_to_file(self, download_path, http_resp, chunksize=2**16):
                 for c in http_resp.iter_content(chunksize):
                     f.write(c)
 
+    def with_path_root(self, mode, namespace_id=None):
+        """
+        Creates a clone of the Dropbox instance with the Dropbox-API-Path-Root header
+        as the appropriate serialized instance of PathRoot.
+
+        For more information, see
+        https://www.dropbox.com/developers/reference/namespace-guide#pathrootmodes
+
+        :param str mode: path root mode to use. Must be one of "home", "root", or "namespace_id"
+        :param str namespace_id: A namespace ID used in conjuntion with "root" or "namespace_id"
+        modes
+        :return: A :class: `Dropbox`
+        :rtype: Dropbox
+        """
+
+        if mode == 'root':
+            path_root = PathRoot.root(namespace_id)
+        elif mode == 'namespace_id':
+            path_root = PathRoot.namespace_id(namespace_id)
+        elif mode == 'home':
+            path_root = PathRoot.home
+        else:
+            raise ValueError("Invalid value for 'mode'.  Must be one of 'home'" +
+                             ", 'root' or 'namespace_id'")
+
+        return self.clone(
+            headers={
+                PATH_ROOT_HEADER: stone_serializers.json_encode(PathRoot_validator, path_root)
+            }
+        )
+
 class Dropbox(_DropboxTransport, DropboxBase):
     """
     Use this class to make requests to the Dropbox API using a user's access
     token. Methods of this class are meant to act on the corresponding user's
     Dropbox.
     """
-    pass
+
 
 class DropboxTeam(_DropboxTransport, DropboxTeamBase):
     """
@@ -532,12 +596,4 @@ def _get_dropbox_client_with_select_header(self, select_header_name, team_member
 
         new_headers = self._headers.copy() if self._headers else {}
         new_headers[select_header_name] = team_member_id
-        return Dropbox(
-            self._oauth2_access_token,
-            max_retries_on_error=self._max_retries_on_error,
-            max_retries_on_rate_limit=self._max_retries_on_rate_limit,
-            timeout=self._timeout,
-            user_agent=self._raw_user_agent,
-            session=self._session,
-            headers=new_headers,
-        )
+        return self.clone(headers=new_headers)

dropbox/exceptions.py

@@ -46,6 +46,17 @@ def __repr__(self):
             self.status_code, self.body)
 
 
+class PathRootError(HttpError):
+    """Error caused by an invalid path root."""
+
+    def __init__(self, request_id, error=None):
+        super(PathRootError, self).__init__(request_id, 422, None)
+        self.error = error
+
+    def __repr__(self):
+        return 'PathRootError({!r}, {!r})'.format(self.request_id, self.error)
+
+
 class BadInputError(HttpError):
     """Errors due to bad input parameters to an API Operation."""
 

test/test_dropbox.py

@@ -22,15 +22,22 @@
     DropboxOAuth2Flow,
     DropboxTeam,
     session,
+    stone_serializers,
 )
+from dropbox.dropbox import PATH_ROOT_HEADER
 from dropbox.exceptions import (
     ApiError,
     AuthError,
     BadInputError,
+    PathRootError,
 )
 from dropbox.files import (
     ListFolderError,
 )
+from dropbox.common import (
+    PathRoot,
+    PathRoot_validator,
+)
 
 def _token_from_env_or_die(env_name='DROPBOX_TOKEN'):
     oauth2_token = os.environ.get(env_name)
@@ -137,5 +144,57 @@ def test_team(self, dbxt):
             team_member_id = r.members[0].profile.team_member_id
             dbxt.as_user(team_member_id).files_list_folder('')
 
+    @dbx_from_env
+    def test_with_path_root_constructor(self, dbx):
+        # Verify valid mode types
+        for mode, ns_id in (
+           ("home", None),
+           ("root", "123"),
+           ("namespace_id", "123")
+        ):
+            dbx_new = dbx.with_path_root(mode, ns_id)
+            self.assertIsNot(dbx_new, dbx)
+
+            path_root_mode = getattr(PathRoot, mode)
+            path_root = path_root_mode(ns_id) if ns_id else path_root_mode
+
+            expected = stone_serializers.json_encode(PathRoot_validator, path_root)
+            self.assertEqual(dbx_new._headers.get(PATH_ROOT_HEADER), expected)
+
+        # verify invalid mode raises ValueError
+        with self.assertRaises(ValueError):
+            dbx.with_path_root("invalid_mode")
+
+    @dbx_from_env
+    def test_path_root(self, dbx):
+        root_info = dbx.users_get_current_account().root_info
+        root_ns = root_info.root_namespace_id
+        home_ns = root_info.home_namespace_id
+
+        # verify home mode
+        dbxpr = dbx.with_path_root('home')
+        dbxpr.files_list_folder('')
+
+        # verify root mode
+        dbxpr = dbx.with_path_root("root", root_ns)
+        dbxpr.files_list_folder('')
+
+        # verify namespace_id mode
+        dbxpr = dbx.with_path_root("namespace_id", home_ns)
+        dbxpr.files_list_folder('')
+
+    @dbx_from_env
+    def test_path_root_err(self, dbx):
+        # verify invalid namespace return is_no_permission error
+        dbxpr = dbx.with_path_root("namespace_id", "1234567890")
+        with self.assertRaises(PathRootError) as cm:
+            dbxpr.files_list_folder('')
+        self.assertTrue(cm.exception.error.is_no_permission())
+
+        dbxpr = dbx.with_path_root("root", "1234567890")
+        with self.assertRaises(PathRootError) as cm:
+            dbxpr.files_list_folder('')
+        self.assertTrue(cm.exception.error.is_invalid_root())
+
 if __name__ == '__main__':
     unittest.main()