Coverage for gws-app/gws/plugin/qfieldcloud/patcher.py: 99%
158 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +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 project: Optional[gws.Project]
59 """GWS project context."""
60 user: gws.User
61 """User who sent the changes."""
62 baseDir: str
63 """Base directory (not used by the patcher)."""
64 changes: list[Change]
65 """Changes to apply."""
66 filePath: str
67 """Path of an uploaded file, as sent by QField."""
68 fileContent: bytes
69 """Content of an uploaded file."""
72class Object:
73 """Patcher, applies changes and file uploads from QField.
75 Changes are converted to model operations, grouped by model, and passed
76 to the model methods ``create_feature``, ``update_feature`` and
77 ``delete_feature``. Changes for unknown or non-editable layers, and
78 updates or deletions of features that do not exist, are skipped with a warning.
79 """
81 root: gws.Root
82 """Root object."""
83 qfcProject: core.QfcProject
84 """QField project."""
85 project: Optional[gws.Project]
86 """GWS project context."""
87 user: gws.User
88 """User who sent the changes."""
89 args: Args
90 """Patcher arguments."""
92 caps: caps_mod.Caps
93 """Capabilities of the QField project."""
94 ops_by_model: dict[str, list[Operation]]
95 """Operations by GeoPackage name of the model."""
97 def apply_changes(self, root: gws.Root, args: Args) -> bool:
98 """Apply the changes in ``args.changes``.
100 Args:
101 root: Root object.
102 args: Patcher arguments.
104 Returns:
105 ``True`` if any operations were committed, ``False`` if there was nothing to apply.
106 """
107 self.root = root
108 self.prepare(args)
110 self.ops_by_model = {}
112 for cc in self.args.changes:
113 self.prepare_change(cc)
115 if not self.ops_by_model:
116 return False
118 for gpName, ops in self.ops_by_model.items():
119 self.commit_operations_for_model(self.caps.modelMap[gpName], ops)
121 return True
123 def prepare(self, args: Args):
124 """Store the arguments in the patcher.
126 Args:
127 args: Patcher arguments.
128 """
129 self.args = args
130 self.qfcProject = self.args.qfcProject
131 self.project = self.args.project
132 self.user = self.args.user
133 self.caps = args.caps
135 def commit_operations_for_model(self, me: caps_mod.ModelEntry, ops: list[Operation]):
136 """Commit operations for a model.
138 Args:
139 me: Model entry.
140 ops: Operations to commit.
141 """
142 with me.model.db.connect() as conn:
143 for op in ops:
144 gws.log.debug(f'{op.type=} {op.feature.attributes=}')
145 mc = gws.ModelContext(op=op.type, user=self.user, project=self.project)
146 if op.type == gws.ModelOperation.create:
147 me.model.create_feature(op.feature, mc)
148 continue
149 if op.type == gws.ModelOperation.update:
150 me.model.update_feature(op.feature, mc)
151 continue
152 if op.type == gws.ModelOperation.delete:
153 me.model.delete_feature(op.feature, mc)
154 conn.commit()
156 def apply_upload(self, root: gws.Root, args: Args) -> bool:
157 """Apply the file upload in ``args.filePath`` and ``args.fileContent``.
159 Args:
160 root: Root object.
161 args: Patcher arguments.
163 Returns:
164 ``True`` if a feature for the file was found and updated.
165 """
166 self.root = root
167 self.prepare(args)
168 return self.commit_upload(args.filePath, args.fileContent)
170 def commit_upload(self, path: str, content: bytes) -> bool:
171 """Write uploaded file content into the first feature that refers to the file.
173 Args:
174 path: File path, as sent by QField.
175 content: File content.
177 Returns:
178 ``True`` if a feature was found and updated.
179 """
180 for me in self.caps.modelMap.values():
181 if self.commit_upload_for_model(me, path, content):
182 return True
183 gws.log.warning(f'commit_upload: feature not found: {path=}')
184 return False
186 def commit_upload_for_model(self, me: caps_mod.ModelEntry, path: str, content: bytes) -> bool:
187 """Write uploaded file content into a feature of a model.
189 Looks for a file field with a ``nameColumn`` whose value equals the path,
190 and writes the content into the field's ``contentColumn``.
192 Args:
193 me: Model entry.
194 path: File path, as sent by QField.
195 content: File content.
197 Returns:
198 ``True`` if a feature was found and updated.
199 """
200 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
202 for fld in me.model.fields:
203 if fld.extType != 'file':
204 continue
205 fld = cast(file_field.Object, fld)
206 if fld.nameColumn is None:
207 continue
208 uid = self.find_uid_for_path(me, fld, path, mc)
209 if not uid:
210 continue
211 gws.log.debug(f'commit_upload: found feature: model={me.gpName}: {fld.name=} {uid=} {path=} ')
213 with me.model.db.connect() as conn:
214 sql = me.model.table().update().where(me.model.uid_equals(uid)).values({fld.contentColumn: content})
215 conn.execute(sql)
216 conn.commit()
218 return True
220 return False
222 def find_uid_for_path(
223 self,
224 me: caps_mod.ModelEntry,
225 ff: file_field.Object,
226 path: str,
227 mc: gws.ModelContext,
228 ) -> Optional[str]:
229 """Find the feature whose file name column equals the path.
231 Args:
232 me: Model entry.
233 ff: File field.
234 path: File path, as sent by QField.
235 mc: Model context.
237 Returns:
238 Primary key of the feature, or ``None`` if not found.
239 """
240 with me.model.db.connect() as conn:
241 sel = me.model.table().select().with_only_columns(me.model.uid_column()).where(ff.nameColumn == path)
242 rec = conn.fetch_first(sel)
243 if rec:
244 return rec[me.model.uidName]
246 def prepare_change(self, cc: Change):
247 """Convert a change to an operation and add it to ``ops_by_model``.
249 The ``fid_gws`` attribute is renamed back to ``fid`` (see the packager).
250 For creations, an automatic primary key is removed from the attributes.
252 Args:
253 cc: Change.
254 """
255 le = self.caps.layerMap.get(cc.layerUid)
256 if not le:
257 gws.log.warning(f'layer not found: {cc.layerUid!r}')
258 return
259 if le.action != caps_mod.LayerAction.edit:
260 gws.log.warning(f'unsupported layer action: {cc.layerUid!r} {le.action!r}')
261 return
263 me = le.modelEntry
264 pk_name = me.model.uidName
266 atts = dict(cc.newAtts)
268 # see notes in packager.py about fid/fid_gws
269 if 'fid_gws' in atts:
270 atts['fid'] = atts.pop('fid_gws')
272 if cc.wkt:
273 geom = me.model.geometryName
274 if not geom:
275 gws.log.warning(f'geometry field not found: {me.gpName!r}')
276 else:
277 atts[geom] = gws.lib.shape.from_wkt(cc.wkt, me.model.geometryCrs)
279 ops = self.ops_by_model.setdefault(me.gpName, [])
281 if cc.type == ChangeType.create:
282 mc = gws.ModelContext(op=gws.ModelOperation.create, user=self.user, project=self.project)
283 pk_field = me.model.field(pk_name)
284 if pk_field and pk_field.isAuto:
285 atts.pop(pk_name, None)
286 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts, isNew=True), mc)
287 ops.append(Operation(type=gws.ModelOperation.create, feature=feat))
288 return
290 if cc.type == ChangeType.delete:
291 mc = gws.ModelContext(op=gws.ModelOperation.delete, user=self.user, project=self.project)
292 pk = cc.oldAtts.get(pk_name, '')
293 feat = self.get_feature(me, pk)
294 if not feat:
295 gws.log.warning(f'delete: not found: {pk=} {me.gpName=} {le.qgisId=}')
296 return
297 ops.append(Operation(type=gws.ModelOperation.delete, feature=feat))
298 return
300 if cc.type == ChangeType.patch:
301 pk = cc.oldAtts.get(pk_name, '')
302 feat = self.get_feature(me, pk)
303 if not feat:
304 gws.log.warning(f'update: not found: {pk=} {me.gpName=} {le.qgisId=}')
305 return
306 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
307 atts[pk_name] = pk
308 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts), mc)
309 ops.append(Operation(type=gws.ModelOperation.update, feature=feat))
310 return
312 def get_feature(self, me: caps_mod.ModelEntry, pk: str) -> gws.Feature | None:
313 """Read a feature by its primary key.
315 Args:
316 me: Model entry.
317 pk: Primary key.
319 Returns:
320 The feature, or ``None`` if not found.
321 """
322 mc = gws.ModelContext(op=gws.ModelOperation.read, user=self.user, project=self.project)
323 fs = me.model.get_features([pk], mc)
324 if fs:
325 return fs[0]