Coverage for gws-app/gws/plugin/qfieldcloud/patcher.py: 99%
155 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 12:46 +0200
1from typing import cast, Optional
3import gws
4import gws.base.shape
5import gws.plugin.model_field.file as file_field
6import gws.base.feature
7import gws.lib.osx
9from . import core, caps as caps_mod
12class ChangeType(gws.Enum):
13 patch = 'patch'
14 create = 'create'
15 delete = 'delete'
18class Change(gws.Data):
19 uid: str
20 type: ChangeType
21 layerUid: str
22 newAtts: dict
23 oldAtts: dict
24 wkt: str
27class Operation(gws.Data):
28 type: gws.ModelOperation
29 feature: gws.Feature
32class Args(gws.Data):
33 qfcProject: core.QfcProject
34 caps: caps_mod.Caps
35 project: gws.Project
36 user: gws.User
37 baseDir: str
38 changes: list[Change]
39 filePath: str
40 fileContent: bytes
43class Object:
44 root: gws.Root
45 qfcProject: core.QfcProject
46 project: gws.Project
47 user: gws.User
48 args: Args
50 caps: caps_mod.Caps
51 ops_by_model: dict[str, list[Operation]]
53 def apply_changes(self, root: gws.Root, args: Args) -> bool:
54 self.root = root
55 self.prepare(args)
57 self.ops_by_model = {}
59 for cc in self.args.changes:
60 self.prepare_change(cc)
62 if not self.ops_by_model:
63 return False
65 for gpName, ops in self.ops_by_model.items():
66 self.commit_operations_for_model(self.caps.modelMap[gpName], ops)
68 return True
70 def prepare(self, args: Args):
71 self.args = args
72 self.qfcProject = self.args.qfcProject
73 self.project = self.args.project
74 self.user = self.args.user
75 self.caps = args.caps
77 def commit_operations_for_model(self, me: caps_mod.ModelEntry, ops: list[Operation]):
78 with me.model.db.connect() as conn:
79 for op in ops:
80 gws.log.debug(f'{op.type=} {op.feature.attributes=}')
81 mc = gws.ModelContext(op=op.type, user=self.user, project=self.project)
82 if op.type == gws.ModelOperation.create:
83 me.model.create_feature(op.feature, mc)
84 continue
85 if op.type == gws.ModelOperation.update:
86 me.model.update_feature(op.feature, mc)
87 continue
88 if op.type == gws.ModelOperation.delete:
89 me.model.delete_feature(op.feature, mc)
90 conn.commit()
92 def apply_upload(self, root: gws.Root, args: Args) -> bool:
93 self.root = root
94 self.prepare(args)
95 return self.commit_upload(args.filePath, args.fileContent)
97 def commit_upload(self, path: str, content: bytes) -> bool:
98 for me in self.caps.modelMap.values():
99 if self.commit_upload_for_model(me, path, content):
100 return True
101 gws.log.warning(f'commit_upload: feature not found: {path=}')
102 return False
104 def commit_upload_for_model(self, me: caps_mod.ModelEntry, path: str, content: bytes) -> bool:
105 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
107 for fld in me.model.fields:
108 if fld.extType != 'file':
109 continue
110 fld = cast(file_field.Object, fld)
111 if fld.nameColumn is None:
112 continue
113 uid = self.find_uid_for_path(me, fld, path, mc)
114 if not uid:
115 continue
116 gws.log.debug(f'commit_upload: found feature: model={me.gpName}: {fld.name=} {uid=} {path=} ')
118 with me.model.db.connect() as conn:
119 sql = me.model.table().update().where(me.model.uid_column().__eq__(uid)).values({fld.contentColumn: content})
120 conn.execute(sql)
121 conn.commit()
123 return True
125 return False
127 def find_uid_for_path(
128 self,
129 me: caps_mod.ModelEntry,
130 ff: file_field.Object,
131 path: str,
132 mc: gws.ModelContext,
133 ) -> Optional[str]:
134 with me.model.db.connect() as conn:
135 sel = me.model.table().select().with_only_columns(me.model.uid_column()).where(ff.nameColumn == path)
136 rec = conn.fetch_first(sel)
137 if rec:
138 return rec[me.model.uidName]
140 def prepare_change(self, cc: Change):
141 le = self.caps.layerMap.get(cc.layerUid)
142 if not le:
143 gws.log.warning(f'layer not found: {cc.layerUid!r}')
144 return
145 if le.action != caps_mod.LayerAction.edit:
146 gws.log.warning(f'unsupported layer action: {cc.layerUid!r} {le.action!r}')
147 return
149 me = le.modelEntry
150 pk_name = me.model.uidName
152 atts = dict(cc.newAtts)
154 # see notes in packager.py about fid/fid_gws
155 if 'fid_gws' in atts:
156 atts['fid'] = atts.pop('fid_gws')
158 if cc.wkt:
159 geom = me.model.geometryName
160 if not geom:
161 gws.log.warning(f'geometry field not found: {me.gpName!r}')
162 else:
163 atts[geom] = gws.base.shape.from_wkt(cc.wkt, me.model.geometryCrs)
165 ops = self.ops_by_model.setdefault(me.gpName, [])
167 if cc.type == ChangeType.create:
168 mc = gws.ModelContext(op=gws.ModelOperation.create, user=self.user, project=self.project)
169 pk_field = me.model.field(pk_name)
170 if pk_field and pk_field.isAuto:
171 atts.pop(pk_name, None)
172 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts, isNew=True), mc)
173 ops.append(Operation(type=gws.ModelOperation.create, feature=feat))
174 return
176 if cc.type == ChangeType.delete:
177 mc = gws.ModelContext(op=gws.ModelOperation.delete, user=self.user, project=self.project)
178 pk = cc.oldAtts.get(pk_name, '')
179 feat = self.get_feature(me, pk)
180 if not feat:
181 gws.log.warning(f'delete: not found: {pk=} {me.gpName=} {le.qgisId=}')
182 return
183 ops.append(Operation(type=gws.ModelOperation.delete, feature=feat))
184 return
186 if cc.type == ChangeType.patch:
187 pk = cc.oldAtts.get(pk_name, '')
188 feat = self.get_feature(me, pk)
189 if not feat:
190 gws.log.warning(f'update: not found: {pk=} {me.gpName=} {le.qgisId=}')
191 return
192 mc = gws.ModelContext(op=gws.ModelOperation.update, user=self.user, project=self.project)
193 atts[pk_name] = pk
194 feat = me.model.feature_from_props(gws.FeatureProps(attributes=atts), mc)
195 ops.append(Operation(type=gws.ModelOperation.update, feature=feat))
196 return
198 def get_feature(self, me: caps_mod.ModelEntry, pk: str) -> gws.Feature | None:
199 mc = gws.ModelContext(op=gws.ModelOperation.read, user=self.user, project=self.project)
200 fs = me.model.get_features([pk], mc)
201 if fs:
202 return fs[0]