|
2 | 2 |
|
3 | 3 | from .utils import is_mergin_config, is_qgis_file, is_versioned_file |
4 | 4 |
|
| 5 | +EDITOR_ROLE_NAME = "editor" |
5 | 6 |
|
6 | | -class EditorHandler: |
7 | | - def __init__(self, mergin_client, project_info: dict): |
8 | | - """ |
9 | | - Initializes the editor instance with the provided Mergin client and project information. |
10 | | -
|
11 | | - Args: |
12 | | - mergin_client (MerginClient): The Mergin client instance. |
13 | | - project_info (dict): The project information dictionary. |
14 | | - """ |
15 | | - self.mc = mergin_client |
16 | | - self.project_info = project_info |
17 | | - self.ROLE_NAME = "editor" |
18 | | - |
19 | | - _disallowed_added_changes = lambda self, change: is_qgis_file(change["path"]) or is_mergin_config(change["path"]) |
20 | | - _disallowed_updated_changes = ( |
21 | | - lambda self, change: is_qgis_file(change["path"]) |
22 | | - or is_mergin_config(change["path"]) |
23 | | - or (is_versioned_file(change["path"]) and change.get("diff") is None) |
24 | | - ) |
25 | | - _disallowed_removed_changes = ( |
26 | | - lambda self, change: is_qgis_file(change["path"]) |
27 | | - or is_mergin_config(change["path"]) |
28 | | - or is_versioned_file(change["path"]) |
29 | | - ) |
30 | | - |
31 | | - def is_enabled(self): |
32 | | - """ |
33 | | - Returns True if the current user has editor access to the project, False otherwise. |
34 | | -
|
35 | | - The method checks if the server supports editor access, and if the current user's project role matches the expected role name for editors. |
36 | | - """ |
37 | | - server_support_editor = self.mc.has_editor_support() |
38 | | - project_role = self.project_info.get("role") |
39 | | - |
40 | | - return server_support_editor and project_role == self.ROLE_NAME |
41 | | - |
42 | | - def _apply_editor_filters(self, changes: dict[str, list[dict]]) -> dict[str, list[dict]]: |
43 | | - added = changes.get("added", []) |
44 | | - updated = changes.get("updated", []) |
45 | | - removed = changes.get("removed", []) |
46 | | - |
47 | | - # filter out files that are not in the editor's list of allowed files |
48 | | - changes["added"] = list(filterfalse(self._disallowed_added_changes, added)) |
49 | | - changes["updated"] = list(filterfalse(self._disallowed_updated_changes, updated)) |
50 | | - changes["removed"] = list(filterfalse(self._disallowed_removed_changes, removed)) |
| 7 | +""" |
| 8 | +Determines whether a given file change should be disallowed based on the file path. |
| 9 | +
|
| 10 | +Returns: |
| 11 | + bool: True if the file change should be disallowed, False otherwise. |
| 12 | +""" |
| 13 | +_disallowed_added_changes = lambda change: is_qgis_file(change["path"]) or is_mergin_config(change["path"]) |
| 14 | +""" |
| 15 | +Determines whether a given file change should be disallowed from being updated. |
| 16 | +
|
| 17 | +The function checks the following conditions: |
| 18 | +- If the file path matches a QGIS file |
| 19 | +- If the file path matches a Mergin configuration file |
| 20 | +- If the file is versioned and the change does not have a diff |
| 21 | +
|
| 22 | +Returns: |
| 23 | + bool: True if the change should be disallowed, False otherwise. |
| 24 | +""" |
| 25 | +_disallowed_updated_changes = ( |
| 26 | + lambda change: is_qgis_file(change["path"]) |
| 27 | + or is_mergin_config(change["path"]) |
| 28 | + or (is_versioned_file(change["path"]) and change.get("diff") is None) |
| 29 | +) |
| 30 | +""" |
| 31 | +Determines whether a given file change should be disallowed from being removed. |
| 32 | +
|
| 33 | +The function checks if the file path of the change matches any of the following conditions: |
| 34 | +- The file is a QGIS file (e.g. .qgs, .qgz) |
| 35 | +- The file is a Mergin configuration file (mergin-config.json) |
| 36 | +- The file is a versioned file (.gpkg, .sqlite) |
| 37 | +
|
| 38 | +If any of these conditions are met, the change is considered disallowed from being removed. |
| 39 | +""" |
| 40 | +_disallowed_removed_changes = ( |
| 41 | + lambda change: is_qgis_file(change["path"]) or is_mergin_config(change["path"]) or is_versioned_file(change["path"]) |
| 42 | +) |
| 43 | + |
| 44 | + |
| 45 | +def is_editor_enabled(mc, project_info: dict) -> bool: |
| 46 | + """ |
| 47 | + The function checks if the server supports editor access, and if the current user's project role matches the expected role name for editors. |
| 48 | + """ |
| 49 | + server_support = mc.has_editor_support() |
| 50 | + project_role = project_info.get("role") |
| 51 | + |
| 52 | + return server_support and project_role == EDITOR_ROLE_NAME |
| 53 | + |
| 54 | + |
| 55 | +def _apply_editor_filters(changes: dict[str, list[dict]]) -> dict[str, list[dict]]: |
| 56 | + """ |
| 57 | + Applies editor-specific filters to the changes dictionary, removing any changes to files that are not in the editor's list of allowed files. |
| 58 | +
|
| 59 | + Args: |
| 60 | + changes (dict[str, list[dict]]): A dictionary containing the added, updated, and removed changes. |
| 61 | +
|
| 62 | + Returns: |
| 63 | + dict[str, list[dict]]: The filtered changes dictionary. |
| 64 | + """ |
| 65 | + added = changes.get("added", []) |
| 66 | + updated = changes.get("updated", []) |
| 67 | + removed = changes.get("removed", []) |
| 68 | + |
| 69 | + # filter out files that are not in the editor's list of allowed files |
| 70 | + changes["added"] = list(filterfalse(_disallowed_added_changes, added)) |
| 71 | + changes["updated"] = list(filterfalse(_disallowed_updated_changes, updated)) |
| 72 | + changes["removed"] = list(filterfalse(_disallowed_removed_changes, removed)) |
| 73 | + return changes |
| 74 | + |
| 75 | + |
| 76 | +def filter_changes(mc, project_info: dict, changes: dict[str, list[dict]]) -> dict[str, list[dict]]: |
| 77 | + """ |
| 78 | + Filters the given changes dictionary based on the editor's enabled state. |
| 79 | +
|
| 80 | + If the editor is not enabled, the changes dictionary is returned as-is. Otherwise, the changes are passed through the `_apply_editor_filters` method to apply any configured filters. |
| 81 | +
|
| 82 | + Args: |
| 83 | + changes (dict[str, list[dict]]): A dictionary mapping file paths to lists of change dictionaries. |
| 84 | +
|
| 85 | + Returns: |
| 86 | + dict[str, list[dict]]: The filtered changes dictionary. |
| 87 | + """ |
| 88 | + if not is_editor_enabled(mc, project_info): |
51 | 89 | return changes |
| 90 | + return _apply_editor_filters(changes) |
| 91 | + |
52 | 92 |
|
53 | | - def filter_changes(self, changes: dict[str, list[dict]]) -> dict[str, list[dict]]: |
54 | | - if not self.is_enabled(): |
55 | | - return changes |
56 | | - return self._apply_editor_filters(changes) |
| 93 | +def prevent_conflicted_copy(path: str, mc, project_info: dict) -> bool: |
| 94 | + return is_editor_enabled(mc, project_info) and any([is_qgis_file(path), is_mergin_config(path)]) |
0 commit comments