Merge pull request #231 from MerginMaps/v2_endpoints_for_user_and_access_management

V2 endpoints for user and access management

Author: MarcelGeo

mergin/client.py

@@ -17,7 +17,9 @@
 import typing
 import warnings
 
-from .common import ClientError, LoginError, InvalidProject, ErrorCode
+from typing import List
+
+from .common import ClientError, LoginError, WorkspaceRole, ProjectRole
 from .merginproject import MerginProject
 from .client_pull import (
     download_file_finalize,
@@ -36,6 +38,7 @@
 from .version import __version__
 
 this_dir = os.path.dirname(os.path.realpath(__file__))
+json_headers = {"Content-Type": "application/json"}
 
 
 class TokenError(Exception):
@@ -207,9 +210,23 @@ def _do_request(self, request):
         except urllib.error.HTTPError as e:
             server_response = json.load(e)
 
-            # We first to try to get the value from the response otherwise we set a default value
-            err_detail = server_response.get("detail", e.read().decode("utf-8"))
-            server_code = server_response.get("code", None)
+            err_detail = None
+            server_code = None
+            # Try to get error detail
+            if isinstance(server_response, dict):
+                server_code = server_response.get("code")
+                err_detail = server_response.get("detail")
+                if not err_detail:
+                    # Extract all field-specific errors and format them
+                    err_detail = "\n".join(
+                        f"{key}: {', '.join(map(str, value))}"
+                        for key, value in server_response.items()
+                        if isinstance(value, list)
+                    ) or str(
+                        server_response
+                    )  # Fallback to raw response if structure is unexpected
+            else:
+                err_detail = str(server_response)
 
             raise ClientError(
                 detail=err_detail,
@@ -244,6 +261,11 @@ def patch(self, path, data=None, headers={}):
         request = urllib.request.Request(url, data, headers, method="PATCH")
         return self._do_request(request)
 
+    def delete(self, path):
+        url = urllib.parse.urljoin(self.url, urllib.parse.quote(path))
+        request = urllib.request.Request(url, method="DELETE")
+        return self._do_request(request)
+
     def login(self, login, password):
         """
         Authenticate login credentials and store session token
@@ -796,6 +818,12 @@ def add_user_permissions_to_project(self, project_path, usernames, permission_le
             if permission_level in ("writer", "owner", "editor", "reader"):
                 access.get("readersnames").append(name)
         self.set_project_access(project_path, access)
+        warnings.warn(
+            "This method will be deprecated in the next major release (1.0.0)"
+            "Use `add_project_collaborator` to create a project permission and "
+            "`update_project_collaborator` to change it instead.",
+            category=DeprecationWarning,
+        )
 
     def remove_user_permissions_from_project(self, project_path, usernames):
         """
@@ -815,6 +843,11 @@ def remove_user_permissions_from_project(self, project_path, usernames):
             if name in access.get("readersnames", []):
                 access.get("readersnames").remove(name)
         self.set_project_access(project_path, access)
+        warnings.warn(
+            "This method will be deprecated in the next major release (1.0.0)"
+            "Use `remove_project_collaborator` instead.",
+            category=DeprecationWarning,
+        )
 
     def project_user_permissions(self, project_path):
         """
@@ -1228,3 +1261,102 @@ def has_editor_support(self):
         Returns whether the server version is acceptable for editor support.
         """
         return is_version_acceptable(self.server_version(), "2024.4.0")
+
+    def create_user(
+        self,
+        email: str,
+        password: str,
+        workspace_id: int,
+        workspace_role: WorkspaceRole,
+        username: str = None,
+        notify_user: bool = False,
+    ) -> dict:
+        """
+        Create a new user in a workspace. The username is generated from the email address.
+
+        param email: email of the new user - must be unique
+        param password: password - must meet the requirements
+        param workspace_id: id of the workspace user is created in
+        param workspace_role: workspace role of the user
+        param username: username - will be autogenerated from the email if not provided
+        param notify_user: flag for email notifications - confirmation email will be sent
+        """
+        params = {
+            "email": email,
+            "password": password,
+            "workspace_id": workspace_id,
+            "role": workspace_role.value,
+            "notify_user": notify_user,
+        }
+        if username:
+            params["username"] = username
+        user_info = self.post("v2/users", params, json_headers)
+        return json.load(user_info)
+
+    def get_workspace_member(self, workspace_id: int, user_id: int) -> dict:
+        """
+        Get a workspace member detail
+        """
+        resp = self.get(f"v2/workspaces/{workspace_id}/members/{user_id}")
+        return json.load(resp)
+
+    def list_workspace_members(self, workspace_id: int) -> List[dict]:
+        """
+        Get a list of workspace members
+        """
+        resp = self.get(f"v2/workspaces/{workspace_id}/members")
+        return json.load(resp)
+
+    def update_workspace_member(
+        self, workspace_id: int, user_id: int, workspace_role: WorkspaceRole, reset_projects_roles: bool = False
+    ) -> dict:
+        """
+        Update workspace role of a workspace member, optionally resets the projects role
+
+        param reset_projects_roles: all project specific roles will be removed
+        """
+        params = {
+            "reset_projects_roles": reset_projects_roles,
+            "workspace_role": workspace_role.value,
+        }
+        workspace_member = self.patch(f"v2/workspaces/{workspace_id}/members/{user_id}", params, json_headers)
+        return json.load(workspace_member)
+
+    def remove_workspace_member(self, workspace_id: int, user_id: int):
+        """
+        Remove a user from workspace members
+        """
+        self.delete(f"v2/workspaces/{workspace_id}/members/{user_id}")
+
+    def list_project_collaborators(self, project_id: int) -> List[dict]:
+        """
+        Get a list of project collaborators
+        """
+        project_collaborators = self.get(f"v2/projects/{project_id}/collaborators")
+        return json.load(project_collaborators)
+
+    def add_project_collaborator(self, project_id: int, user: str, project_role: ProjectRole) -> dict:
+        """
+        Add a user to project collaborators and grant them a project role.
+        Fails if user is already a member of the project.
+
+        param user: login (username or email) of the user
+        """
+        params = {"role": project_role.value, "user": user}
+        project_collaborator = self.post(f"v2/projects/{project_id}/collaborators", params, json_headers)
+        return json.load(project_collaborator)
+
+    def update_project_collaborator(self, project_id: int, user_id: int, project_role: ProjectRole) -> dict:
+        """
+        Update project role of the existing project collaborator.
+        Fails if user is not a member of the project yet.
+        """
+        params = {"role": project_role.value}
+        project_collaborator = self.patch(f"v2/projects/{project_id}/collaborators/{user_id}", params, json_headers)
+        return json.load(project_collaborator)
+
+    def remove_project_collaborator(self, project_id: int, user_id: int):
+        """
+        Remove a user from project collaborators
+        """
+        self.delete(f"v2/projects/{project_id}/collaborators/{user_id}")

mergin/common.py

@@ -65,3 +65,27 @@ class InvalidProject(Exception):
 
         import dateutil.parser
         from dateutil.tz import tzlocal
+
+
+class WorkspaceRole(Enum):
+    """
+    Workspace roles
+    """
+
+    GUEST = "guest"
+    READER = "reader"
+    EDITOR = "editor"
+    WRITER = "writer"
+    ADMIN = "admin"
+    OWNER = "owner"
+
+
+class ProjectRole(Enum):
+    """
+    Project roles
+    """
+
+    READER = "reader"
+    EDITOR = "editor"
+    WRITER = "writer"
+    OWNER = "owner"

mergin/editor.py

@@ -1,5 +1,5 @@
 from itertools import filterfalse
-from typing import Callable
+from typing import Callable, Dict, List
 
 from .utils import is_mergin_config, is_qgis_file, is_versioned_file
 
@@ -24,7 +24,7 @@ def is_editor_enabled(mc, project_info: dict) -> bool:
     return server_support and project_role == EDITOR_ROLE_NAME
 
 
-def _apply_editor_filters(changes: dict[str, list[dict]]) -> dict[str, list[dict]]:
+def _apply_editor_filters(changes: Dict[str, List[dict]]) -> Dict[str, List[dict]]:
     """
     Applies editor-specific filters to the changes dictionary, removing any changes to files that are not in the editor's list of allowed files.
 
@@ -40,7 +40,7 @@ def _apply_editor_filters(changes: dict[str, list[dict]]) -> dict[str, list[dict
     return changes
 
 
-def filter_changes(mc, project_info: dict, changes: dict[str, list[dict]]) -> dict[str, list[dict]]:
+def filter_changes(mc, project_info: dict, changes: Dict[str, List[dict]]) -> Dict[str, List[dict]]:
     """
     Filters the given changes dictionary based on the editor's enabled state.