V2 endpoints for user and access management

.github/workflows/autotests.yml

@@ -28,10 +28,9 @@ jobs:
 
  - name: Run tests
  run: |
- pytest --cov=mergin mergin/test/
+ pytest --cov=mergin --cov-report=lcov mergin/test/
 
  - name: Submit coverage to Coveralls
- env:
- GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- run: |
- coveralls --service=github
+ uses: coverallsapp/github-action@v2
+ with:
+ format: lcov

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.