Create invitation method

mergin/client.py

@@ -1278,6 +1278,14 @@ def has_editor_support(self):
         """
         return is_version_acceptable(self.server_version(), "2024.4.0")
 
+    def check_collaborators_members_support(self):
+        """
+        Check if the server is compatible with v2 endpoints for project collaborators and workspace members
+        """
+        min_version = "2025.1.0"
+        if not is_version_acceptable(self.server_version(), f"{min_version}"):
+            raise NotImplementedError(f"This needs server at version {min_version} or later")
+
     def create_user(
         self,
         email: str,
@@ -1297,6 +1305,7 @@ def create_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
         """
+        self.check_collaborators_members_support()
         params = {
             "email": email,
             "password": password,
@@ -1313,13 +1322,15 @@ def get_workspace_member(self, workspace_id: int, user_id: int) -> dict:
         """
         Get a workspace member detail
         """
+        self.check_collaborators_members_support()
         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
         """
+        self.check_collaborators_members_support()
         resp = self.get(f"v2/workspaces/{workspace_id}/members")
         return json.load(resp)
 
@@ -1331,6 +1342,7 @@ def update_workspace_member(
 
         param reset_projects_roles: all project specific roles will be removed
         """
+        self.check_collaborators_members_support()
         params = {
             "reset_projects_roles": reset_projects_roles,
             "workspace_role": workspace_role.value,
@@ -1342,12 +1354,14 @@ def remove_workspace_member(self, workspace_id: int, user_id: int):
         """
         Remove a user from workspace members
         """
+        self.check_collaborators_members_support()
         self.delete(f"v2/workspaces/{workspace_id}/members/{user_id}")
 
     def list_project_collaborators(self, project_id: str) -> List[dict]:
         """
         Get a list of project collaborators
         """
+        self.check_collaborators_members_support()
         project_collaborators = self.get(f"v2/projects/{project_id}/collaborators")
         return json.load(project_collaborators)
 
@@ -1358,6 +1372,7 @@ def add_project_collaborator(self, project_id: str, user: str, project_role: Pro
 
         param user: login (username or email) of the user
         """
+        self.check_collaborators_members_support()
         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)
@@ -1367,6 +1382,7 @@ def update_project_collaborator(self, project_id: str, user_id: int, project_rol
         Update project role of the existing project collaborator.
         Fails if user is not a member of the project yet.
         """
+        self.check_collaborators_members_support()
         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)
@@ -1375,6 +1391,7 @@ def remove_project_collaborator(self, project_id: str, user_id: int):
         """
         Remove a user from project collaborators
         """
+        self.check_collaborators_members_support()
         self.delete(f"v2/projects/{project_id}/collaborators/{user_id}")
 
     def server_config(self) -> dict:
@@ -1441,3 +1458,14 @@ def send_logs(
         else:
             request = urllib.request.Request(url, data=payload, headers=header)
             return self._do_request(request)
+
+    def create_invitation(self, workspace_id: int, email: str, workspace_role: WorkspaceRole):
+        """
+        Create invitation to workspace for specific role
+        """
+        min_version = "2025.6.1"
+        if not is_version_acceptable(self.server_version(), min_version):
+            raise NotImplementedError(f"This needs server at version {min_version} or later")
+        params = {"email": email, "role": workspace_role.value}
+        ws_inv = self.post(f"v2/workspaces/{workspace_id}/invitations", params, json_headers)
+        return json.load(ws_inv)

mergin/test/test_client.py

@@ -21,7 +21,6 @@
     decode_token_data,
     TokenError,
     ServerType,
-    WorkspaceRole,
 )
 from ..client_push import push_project_async, push_project_cancel
 from ..client_pull import (
@@ -2913,6 +2912,22 @@ def test_do_request_error_handling(mc: MerginClient):
     assert "Passwords must be at least 8 characters long." in e.value.detail
 
 
+def test_create_invitation(mc: MerginClient):
+    """Test client method to create workspace invitation"""
+    workspace_id = next((w["id"] for w in mc.workspaces_list() if w["name"] == mc.username()))
+    role = WorkspaceRole.WRITER
+    email = "invitation@client.py"
+    inv = mc.create_invitation(workspace_id, email, role)
+    assert inv["email"] == email
+    assert inv["role"] == role.value
+    mc.delete(f"v1/workspace/invitation/{inv['id']}")  # resolves invitation to allow another invitation to the email
+    role = WorkspaceRole.GUEST
+    inv = mc.create_invitation(workspace_id, email, role)
+    assert inv["email"] == email
+    assert "projects" not in inv
+    mc.delete(f"v1/workspace/invitation/{inv['id']}")
+
+
 def test_validate_auth(mc: MerginClient):
     """Test validate authentication under different scenarios."""