Coverage for gws-app/gws/plugin/qfieldcloud/patcher.py: 98%
164 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +0200
1"""Applies changes and file uploads from QField to the database."""
3from typing import cast, Optional
5import gws
6import gws.lib.shape
7import gws.plugin.model_field.file as file_field
8import gws.base.feature
9import gws.lib.osx
11from . import core, caps as caps_mod
14class ChangeType(gws.Enum):
15 """Type of a change, as in the delta ``method``."""
17 patch = 'patch'
18 """Update an existing feature."""
19 create = 'create'
20 """Create a new feature."""
21 delete = 'delete'
22 """Delete a feature."""
25class Change(gws.Data):
26 """A single change, extracted from a QField delta."""
28 uid: str
29 """Delta uuid."""
30 type: ChangeType
31 """Change type."""
32 layerUid: str
33 """Id of the QGIS layer."""
34 newAtts: dict
35 """New attribute values."""
36 oldAtts: dict
37 """Old attribute values, used to find the primary key of updated and deleted features."""
38 wkt: str
39 """New geometry as WKT, or an empty string."""
42class Operation(gws.Data):
43 """A model operation to be committed."""
45 type: gws.ModelOperation
46 """Operation type."""
47 feature: gws.Feature
48 """Feature to create, update or delete."""
51class Args(gws.Data):
52 """Arguments for the patcher."""
54 qfcProject: core.QfcProject
55 """QField project."""
56 caps: caps_mod.Caps
57 """Capabilities of the QField project."""
58 db: gws.DatabaseProvider
59 """Database provider of the models, used for the transaction."""
60 project: Optional[gws.Project]
61 """GWS project context."""
62 user: gws.User
63 """User who sent the changes."""
64 baseDir: str
65 """Base directory (not used by the patcher)."""
66 changes: list[Change]
67 """Changes to apply."""
68 filePath: str
69 """Path of an uploaded file, as sent by QField."""
70 fileContent: bytes
71 """Content of an uploaded file."""
74class Object:
75 """Patcher, applies changes and file uploads from QField.
77 Changes are converted to model operations, grouped by model, and passed
78 to the model methods ``create_feature``, ``update_feature`` and
79 ``delete_feature``. All changes are applied in one transaction:
80 if any change fails, including changes for unknown or non-editable layers and
81 updates or deletions of features that do not exist, nothing is written.
82 """
84 root: gws.Root
85 """Root object."""
86 qfcProject: core.QfcProject
87 """QField project."""
88 project: Optional[gws.Project]
89 """GWS project context."""
90 user: gws.User
91 """User who sent the changes."""
92 args: Args
93 """Patcher arguments."""
95 caps: caps_mod.Caps
96 """Capabilities of the QField project."""
97 ops_by_model: dict[str, list[Operation]]
98 """Operations by GeoPackage name of the model."""
100 def apply_changes(self, root: gws.Root, args: Args) -> bool:
101 """Apply the changes in ``args.changes``.
103 The changes are applied in one transaction, which is rolled back if any change fails.
104 All models must use the database provider ``args.db``.
106 Args:
107 root: Root object.
108 args: Patcher arguments.
110 Returns:
111 ``True`` if any operations were committed, ``False`` if there was nothing to apply.
113 Raises:
114 ``gws.Error``: If a change refers to an unknown layer or a missing feature. Database errors are raised as is.
115 """
116 self.root = root
117 self.prepare(args)
119 self.ops_by_model = {}
120 if not self.args.changes:
121 return False
123 with self.args.db.begin():
124 for cc in self.args.changes:
125 self.prepare_change(cc)
127 for gpName, ops in self.ops_by_model.items():
128 self.perform_operations_for_model(self.caps.modelMap[gpName], ops)
130 return any(self.ops_by_model.values())
132 def prepare(self, args: Args):
133 """Store the arguments in the patcher.
135 Args:
136 args: Patcher arguments.
137 """
138 self.args = args
139 self.qfcProject = self.args.qfcProject
140 self.project = self.args.project
141 self.user = self.args.user
142 self.caps = args.caps
144 def perform_operations_for_model(self, me: caps_mod.ModelEntry, ops: list[Operation]):
145 """Perform operations for a model, in the transaction opened by ``apply_changes``.
147 Args:
148 me: Model entry.
149 ops: Operations to perform.
150 """
151 for op in ops:
152 gws.log.debug(f'{op.type=} {op.feature.attributes=}')
153 mc = gws.ModelContext(op=op.type, user=self.user, project=self.project)
154 if op.type == gws.ModelOperation.create:
155 me.model.create_feature(op.feature, mc)
156 elif op.type == gws.ModelOperation.update:
157 me.model.update_feature(op.feature, mc)
158 elif op.type == gws.ModelOperation.delete:
159 me.model.delete_feature(op.feature, mc)
161 def apply_upload(self, root: gws.Root, args: Args) -> bool:
162 """Apply the file upload in ``args.filePath`` and ``args.fileContent``.
164 Args:
165 root: Root object.
166 args: Patcher arguments.
168 Returns:
169 ``True`` if a feature for the file was found and updated.
170 """
171 self.root = root
172 self.prepare(args)
173 return self.commit_upload(args.filePath, args.fileContent)
175 def commit_upload(self, path: str, content: bytes) -> bool:
176 """Write uploaded file content into the first feature that refers to the file.
178 Args:
179 path: File path, as sent by QField.
180 content: File content.
182 Returns:
183 ``True`` if a feature was found and updated.
184 """
185 for me in self.caps.modelMap.values():
186 if self.commit_upload_for_model(me, path, content):
187 return True
188 gws.log.warning(f'commit_upload: feature not found: {path=}')
189 return False
191 def commit_upload_for_model(self, me: caps_mod.ModelEntry, path: str, content: bytes) -> bool:
192 """Write uploaded file content into a feature of a model.
194 Looks for a file field with a ``nameColumn`` whose value equals the path,
195 and writes the content into the field's ``contentColumn``.
197 Args:
198 me: Model entry.
199 path: File path, as sent by QField.
200 content: File content.
202 Returns:
203 ``True`` if a feature was found and updated.
204 """
205 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
207 for fld in me.model.fields:
208 if fld.extType != 'file':
209 continue
210 fld = cast(file_field.Object, fld)
211 if not fld.nameColumnName:
212 continue
213 uid = self.find_uid_for_path(me, fld, path, mc)
214 if not uid:
215 continue
216 gws.log.debug(f'commit_upload: found feature: model={me.gpName}: {fld.name=} {uid=} {path=} ')
218 with me.model.db.begin() as conn:
219 sql = me.model.table().update().where(me.model.uid_equals(uid)).values({fld.contentColumnName: content})
220 conn.execute(sql)
222 return True
224 return False
226 def find_uid_for_path(
227 self,
228 me: caps_mod.ModelEntry,
229 ff: file_field.Object,
230 path: str,
231 mc: gws.ModelContext,
232 ) -> Optional[str]:
233 """Find the feature whose file name column equals the path.
235 Args:
236 me: Model entry.
237 ff: File field.
238 path: File path, as sent by QField.
239 mc: Model context.
241 Returns:
242 Primary key of the feature, or ``None`` if not found.
243 """
244 with me.model.db.begin() as conn:
245 sel = me.model.table().select().with_only_columns(me.model.uid_column()).where(me.model.column(ff.nameColumnName) == path)
246 rec = conn.fetch_first(sel)
247 if rec:
248 return rec[me.model.uidName]
250 def prepare_change(self, cc: Change):
251 """Convert a change to an operation and add it to ``ops_by_model``.
253 The ``fid_gws`` attribute is renamed back to ``fid`` (see the packager).
254 For creations, an automatic primary key is removed from the attributes.
255 Changes locked in the QGIS project are rejected.
257 Args:
258 cc: Change.
260 Raises:
261 ``gws.Error``: If the layer is not found or not editable, the geometry field is not found,
262 or the feature to update or delete does not exist.
263 ``gws.ForbiddenError``: If the change is locked in the QGIS project.
264 """
265 le = self.caps.layerMap.get(cc.layerUid)
266 if not le:
267 raise gws.Error(f'layer not found: {cc.layerUid!r}')
268 if le.action != caps_mod.LayerAction.edit:
269 raise gws.Error(f'unsupported layer action: {cc.layerUid!r} {le.action!r}')
271 if self.is_locked_in_qgis(le, cc):
272 raise gws.ForbiddenError(f'{cc.type}: locked in QGIS: {le.qgisId=}')
274 me = le.modelEntry
275 pk_name = me.model.uidName
277 atts = {core.db_field_name(k): v for k, v in cc.newAtts.items()}
279 if cc.wkt:
280 geom = me.model.geometryName
281 if not geom:
282 raise gws.Error(f'geometry field not found: {me.gpName!r}')
283 atts[geom] = gws.lib.shape.from_wkt(cc.wkt, me.model.geometryCrs)
285 ops = self.ops_by_model.setdefault(me.gpName, [])
287 if cc.type == ChangeType.create:
288 mc = gws.ModelContext(op=gws.ModelOperation.create, user=self.user, project=self.project)
289 pk_field = me.model.field(pk_name)
290 if pk_field and pk_field.isAuto:
291 atts.pop(pk_name, None)
292 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts, isNew=True), mc)
293 ops.append(Operation(type=gws.ModelOperation.create, feature=feat))
294 return
296 if cc.type == ChangeType.delete:
297 mc = gws.ModelContext(op=gws.ModelOperation.delete, user=self.user, project=self.project)
298 pk = cc.oldAtts.get(core.gp_field_name(pk_name), '')
299 feat = self.get_feature(me, pk)
300 if not feat:
301 raise gws.Error(f'delete: not found: {pk=} {me.gpName=} {le.qgisId=}')
302 ops.append(Operation(type=gws.ModelOperation.delete, feature=feat))
303 return
305 if cc.type == ChangeType.patch:
306 pk = cc.oldAtts.get(core.gp_field_name(pk_name), '')
307 feat = self.get_feature(me, pk)
308 if not feat:
309 raise gws.Error(f'update: not found: {pk=} {me.gpName=} {le.qgisId=}')
310 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
311 atts[pk_name] = pk
312 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts), mc)
313 ops.append(Operation(type=gws.ModelOperation.update, feature=feat))
314 return
316 def is_locked_in_qgis(self, le: caps_mod.LayerEntry, cc: Change) -> bool:
317 """Check if a change is locked in the QGIS project.
319 Args:
320 le: Layer entry.
321 cc: Change.
323 Returns:
324 ``True`` if the layer is read-only or the operation of the change is locked without an expression.
325 """
326 if le.readOnly:
327 return True
328 if cc.type == ChangeType.create:
329 return caps_mod.is_locked(le.props, 'feature_addition')
330 if cc.type == ChangeType.delete:
331 return caps_mod.is_locked(le.props, 'feature_deletion')
332 if cc.type == ChangeType.patch:
333 if cc.newAtts and caps_mod.is_locked(le.props, 'attribute_editing'):
334 return True
335 if cc.wkt and caps_mod.is_locked(le.props, 'geometry_editing'):
336 return True
337 return False
339 def get_feature(self, me: caps_mod.ModelEntry, pk: str) -> gws.Feature | None:
340 """Read a feature by its primary key.
342 Args:
343 me: Model entry.
344 pk: Primary key.
346 Returns:
347 The feature, or ``None`` if not found.
348 """
349 mc = gws.ModelContext(op=gws.ModelOperation.read, user=self.user, project=self.project)
350 fs = me.model.get_features([pk], mc)
351 if fs:
352 return fs[0]