Coverage for gws-app/gws/plugin/qfieldcloud/delta.py: 100%
78 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +0200
1"""Storage of the deltas sent by QField, with their status."""
3from typing import Optional
5import gws
6import gws.lib.jsonx
7import gws.lib.datetimex as dtx
9from . import core, api, patcher
12PENDING_TIMEOUT = 300
13"""Time in seconds after which a pending delta is considered failed."""
15STORAGE_DDL = """
16 CREATE TABLE IF NOT EXISTS deltas (
17 uid TEXT PRIMARY KEY,
18 created_by TEXT NOT NULL,
19 id TEXT NOT NULL,
20 client_id TEXT,
21 created_at INTEGER NOT NULL,
22 updated_at INTEGER NOT NULL,
23 status TEXT NOT NULL,
24 output TEXT,
25 content TEXT NOT NULL
26 );
27 CREATE INDEX IF NOT EXISTS deltas_created_at ON deltas (created_at);
28 CREATE INDEX IF NOT EXISTS deltas_created_by_created_at ON deltas (created_by, created_at);
29 CREATE TABLE IF NOT EXISTS payloads (
30 created_by TEXT NOT NULL,
31 payload_id TEXT NOT NULL,
32 delta_uid TEXT NOT NULL,
33 PRIMARY KEY (created_by, payload_id, delta_uid)
34 );
35 CREATE INDEX IF NOT EXISTS payloads_delta_uid ON payloads (delta_uid);
36"""
39class Record(gws.Data):
40 """A delta as stored on the server, with its status."""
42 uid: str
43 """Record uid, a hash of the user and the delta content."""
44 id: str
45 """Delta id (the ``uuid`` of the delta)."""
46 created_by: str
47 """Login name of the user who sent the delta."""
48 client_id: str
49 """Id of the local export on the device."""
50 created_at: int
51 """Creation time as a timestamp."""
52 updated_at: int
53 """Last update time as a timestamp."""
54 status: api.DeltaStatusType
55 """Delta status."""
56 output: str
57 """Error message, if the delta could not be applied."""
58 content: api.Delta
59 """Delta as sent by QField."""
62def parse_payload(text: str) -> api.DeltasPayload:
63 """Parse the content of a delta file, raise ``gws.BadRequestError`` if it is invalid."""
64 try:
65 js = gws.lib.jsonx.from_string(text)
66 return api.DeltasPayload(
67 deltas=[_from_dict(d) for d in js['deltas']],
68 files=js.get('files', []),
69 id=js['id'],
70 project=js['project'],
71 version=js['version'],
72 )
73 except Exception as exc:
74 raise gws.BadRequestError(f'invalid delta file content: {exc}')
77def new_record(uid: str, d: api.Delta, user_name: str) -> Record:
78 """Create a pending delta record for a delta."""
79 now = dtx.to_timestamp()
80 return Record(
81 uid=uid,
82 id=d.uuid,
83 created_by=user_name,
84 client_id=d.clientId or '',
85 created_at=now,
86 updated_at=now,
87 status=api.DeltaStatusType.pending,
88 output='',
89 content=d,
90 )
93def store_all(db: gws.lib.sqlitex.Object, drs: list[Record]):
94 """Store delta records, replacing the stored ones with the same uid."""
95 for dr in drs:
96 db.insert('deltas', _row_from_record(dr), on_conflict='update')
99def link_payload(db: gws.lib.sqlitex.Object, payload_id: str, delta_uids: list[str], user_name: str):
100 """Link a payload to its delta records, replacing the links stored for the same payload and user."""
101 db.execute(
102 'DELETE FROM payloads WHERE created_by=:user_name AND payload_id=:payload_id',
103 user_name=user_name,
104 payload_id=payload_id,
105 )
106 for uid in delta_uids:
107 db.insert('payloads', {'created_by': user_name, 'payload_id': payload_id, 'delta_uid': uid}, on_conflict='ignore')
110def extract_changes(drs: list[Record]) -> list[patcher.Change]:
111 """Convert delta records to patcher changes."""
112 changes = []
113 for dr in drs:
114 d = dr.content
115 changes.append(
116 patcher.Change(
117 uid=d.uuid,
118 type=d.method,
119 layerUid=d.localLayerId,
120 newAtts=(d.new.attributes or {}) if d.new else {},
121 oldAtts=(d.old.attributes or {}) if d.old else {},
122 wkt=(d.new.geometry or '') if d.new else '',
123 )
124 )
125 return changes
128def set_status_for_all(db: gws.lib.sqlitex.Object, drs: list[Record], status: api.DeltaStatusType, output: str = ''):
129 """Set the status and output of delta records, in the objects and in the database."""
130 now = dtx.to_timestamp()
132 for dr in drs:
133 dr.status = status
134 dr.output = output
135 dr.updated_at = now
137 store_all(db, drs)
140def get_one(db: gws.lib.sqlitex.Object, uid: str, user_name: str) -> Optional[Record]:
141 """Load a delta record of a user, ``None`` if not found."""
142 rs = db.select(
143 'SELECT * FROM deltas WHERE uid=:uid AND created_by=:user_name',
144 uid=uid,
145 user_name=user_name,
146 )
147 return _record_from_row(rs[0]) if rs else None
150def get_for_payload(db: gws.lib.sqlitex.Object, payload_id: str, user_name: str) -> list[Record]:
151 """Load the delta records of a payload sent by a user, empty if not found."""
152 rs = db.select(
153 """
154 SELECT deltas.* FROM payloads JOIN deltas ON deltas.uid = payloads.delta_uid
155 WHERE payloads.created_by=:user_name AND payloads.payload_id=:payload_id
156 ORDER BY payloads.rowid
157 """,
158 user_name=user_name,
159 payload_id=payload_id,
160 )
161 return [_record_from_row(r) for r in rs]
164def get_history(db: gws.lib.sqlitex.Object, user_name: str, limit: int, offset: int) -> list[tuple[Record, str]]:
165 """Load delta records of a user, newest first, each with the id of the last payload it was sent with."""
166 rs = db.select(
167 """
168 SELECT
169 deltas.*,
170 COALESCE((
171 SELECT payload_id FROM payloads
172 WHERE payloads.delta_uid = deltas.uid
173 ORDER BY payloads.rowid DESC LIMIT 1
174 ), '') AS payload_id
175 FROM deltas
176 WHERE created_by=:user_name
177 ORDER BY created_at DESC, rowid DESC
178 LIMIT :limit OFFSET :offset
179 """,
180 user_name=user_name,
181 limit=limit,
182 offset=offset,
183 )
184 return [(_record_from_row(r), r['payload_id']) for r in rs]
187def status(dr: Record, payload_id: str) -> core.DeltaStatus:
188 """Convert a delta record to the status returned for a payload, reporting a stale pending delta as failed."""
189 now = dtx.now().timestamp()
191 status = dr.status
192 output = dr.output
194 # delta record has been pending for longer than ``PENDING_TIMEOUT``
195 if dr.status == api.DeltaStatusType.pending and now - dr.updated_at > PENDING_TIMEOUT:
196 status = api.DeltaStatusType.error
197 output = 'timed out'
199 return core.DeltaStatus(
200 id=dr.id,
201 deltafile_id=payload_id,
202 created_by=dr.created_by,
203 created_at=dtx.to_iso_string(dtx.from_timestamp(dr.created_at)),
204 updated_at=dtx.to_iso_string(dtx.from_timestamp(dr.updated_at)),
205 status=status,
206 output=output,
207 content=dr.content,
208 )
211def cleanup(db: gws.lib.sqlitex.Object, life_time: int):
212 """Remove delta records older than ``life_time`` seconds, and the payload links to them."""
213 db.execute('DELETE FROM deltas WHERE created_at < :t', t=dtx.to_timestamp() - life_time)
214 db.execute('DELETE FROM payloads WHERE delta_uid NOT IN (SELECT uid FROM deltas)')
217def _from_dict(d: dict) -> api.Delta:
218 """Convert a delta from the payload to an API delta."""
219 for key in ('uuid', 'method', 'localLayerId'):
220 if not d.get(key):
221 raise ValueError(f'delta: missing {key!r}')
222 new = d.get('new')
223 old = d.get('old')
224 return api.Delta(
225 d,
226 new=api.DeltaFeature(new) if new else None,
227 old=api.DeltaFeature(old) if old else None,
228 )
231def _row_from_record(dr: Record) -> dict:
232 """Convert a delta record to a database row."""
233 return dict(
234 uid=dr.uid,
235 created_by=dr.created_by,
236 id=dr.id,
237 client_id=dr.client_id,
238 created_at=dr.created_at,
239 updated_at=dr.updated_at,
240 status=dr.status,
241 output=dr.output,
242 content=gws.lib.jsonx.to_string(dr.content),
243 )
246def _record_from_row(rec: dict) -> Record:
247 """Convert a database row to a delta record."""
248 return Record(
249 uid=rec['uid'],
250 id=rec['id'],
251 created_by=rec['created_by'],
252 client_id=rec['client_id'] or '',
253 created_at=rec['created_at'],
254 updated_at=rec['updated_at'],
255 status=rec['status'],
256 output=rec['output'] or '',
257 content=_from_dict(gws.lib.jsonx.from_string(rec['content'])),
258 )