From e2adee4db135d186acfb68ff02cdb4da355c995b Mon Sep 17 00:00:00 2001 From: Bob Date: Tue, 6 Oct 2026 15:15:54 +0000 Subject: [PATCH 01/11] feat(api): add GET /api/0/devices endpoint (phase 2.2) Mirror aw-server-rust's /api/0/devices endpoint in Python aw-server. Returns the local device id (UUID, stable across restarts) and hostname. Paired with aw-core#177 which adds device_id to the SQLite schema. Part of ActivityWatch/activitywatch#302 (bucket identity unification). Git-Session-Id: 02ce --- aw_server/api.py | 4 ++++ aw_server/rest.py | 16 ++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/aw_server/api.py b/aw_server/api.py index 55dfbd2..e42af27 100644 --- a/aw_server/api.py +++ b/aw_server/api.py @@ -107,6 +107,10 @@ def get_info(self) -> Dict[str, Any]: } return payload + def get_devices(self) -> List[Dict[str, Any]]: + """Get list of known devices (currently only the local device)""" + return [{"id": get_device_id(), "hostname": gethostname()}] + def get_buckets(self) -> Dict[str, Dict]: """Get dict {bucket_name: Bucket} of all buckets""" logger.debug("Received get request for buckets") diff --git a/aw_server/rest.py b/aw_server/rest.py index 1106b37..2f89cdf 100644 --- a/aw_server/rest.py +++ b/aw_server/rest.py @@ -102,6 +102,14 @@ def decorator(*args, **kwargs): }, ) +device = api.model( + "Device", + { + "id": fields.String(description="Stable device identifier (UUID)"), + "hostname": fields.String(description="Current hostname of the device"), + }, +) + create_bucket = api.model( "CreateBucket", { @@ -183,6 +191,14 @@ def get(self) -> Dict[str, Dict]: return current_app.api.get_info() +@api.route("/0/devices") +class DevicesResource(Resource): + @api.marshal_list_with(device) + @copy_doc(ServerAPI.get_devices) + def get(self): + return current_app.api.get_devices() + + # BUCKETS From 6372390442730c7452db080e8f3498cbd36cf829 Mon Sep 17 00:00:00 2001 From: Bob Date: Tue, 6 Oct 2026 15:28:40 +0000 Subject: [PATCH 02/11] test(api): add test_devices covering GET /api/0/devices response and stability Git-Session-Id: aa774620-bec5-581f-adf4-2c8cc3cf4eb6 --- tests/test_server.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/test_server.py b/tests/test_server.py index 85ae426..6995160 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -30,6 +30,23 @@ def test_info(flask_client): assert r.json["profile"] == "testing" +def test_devices(flask_client): + r = flask_client.get("/api/0/devices") + assert r.status_code == 200 + devices = r.json + assert isinstance(devices, list) + assert len(devices) == 1 + device = devices[0] + assert "id" in device + assert "hostname" in device + # id is stable across calls + r2 = flask_client.get("/api/0/devices") + assert r2.json[0]["id"] == device["id"] + # id matches the server's own device_id + info = flask_client.get("/api/0/info").json + assert device["id"] == info["device_id"] + + def test_buckets(flask_client, bucket, benchmark): @benchmark def list_buckets(): From 6d7a583fed8cbf574b524885ac7c7fcf96558ecc Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 12:24:57 +0000 Subject: [PATCH 03/11] feat(api): add device-scoped bucket and query routing Git-Session-Id: 931478ca-2d2b-48c5-bb99-0bb5ad608478 --- Makefile | 2 +- README.md | 48 +++++ aw_server/api.py | 178 ++++++++++++++-- aw_server/rest.py | 179 +++++++++++----- poetry.lock | 33 +-- pyproject.toml | 3 +- tests/test_device_addressing.py | 355 ++++++++++++++++++++++++++++++++ 7 files changed, 715 insertions(+), 83 deletions(-) create mode 100644 tests/test_device_addressing.py diff --git a/Makefile b/Makefile index e66765a..592fcf8 100644 --- a/Makefile +++ b/Makefile @@ -24,7 +24,7 @@ test: @# Note that extensive integration tests are also run in the bundle repo, @# for both aw-server and aw-server-rust, but without code coverage. python -c 'import aw_server' - python -m pytest tests/test_server.py tests/test_profile.py tests/test_profile_config.py tests/test_query_cache.py tests/test_import.py tests/test_host_header.py tests/test_database_concurrency.py + python -m pytest tests/test_server.py tests/test_profile.py tests/test_profile_config.py tests/test_query_cache.py tests/test_import.py tests/test_device_addressing.py tests/test_host_header.py tests/test_database_concurrency.py typecheck: python -m mypy aw_server tests --ignore-missing-imports diff --git a/README.md b/README.md index c66535c..05656bc 100644 --- a/README.md +++ b/README.md @@ -50,3 +50,51 @@ You can also get a very decent API browser by browsing to `localhost:5600` after There are also some API examples in the [documentation](https://docs.activitywatch.net/en/latest/api.html). + +## Draft device-addressed API + +This branch is implementation work for ActivityWatch/activitywatch#302. It depends +on draft ActivityWatch/aw-core#183 (pinned in `pyproject.toml`), and includes the +local discovery commits from ActivityWatch/aw-server#183 rebased on master. +**Do not release this as a complete Python/Rust identity contract yet.** + +With the SQLite backend, `/api/0/devices//buckets/` and its +`events`, `events/count`, `events/`, `events/delete`, `heartbeat`, +`export`, and `export/csv` operations use an explicit device scope. Device bucket +listing is `/api/0/devices//buckets/`; scoped queries use +`POST /api/0/devices//query`; scoped full export is +`GET /api/0/devices//export`. `local` and `/api/0/info.device_id` +normalize to the same stored local sentinel. Peer `hostname="!local"` is rejected. +Unsupported Memory/Peewee scoped operations return 501, never global fallback. + +Legacy bucket creation is local-only. Other legacy name-addressed operations +retain unique global-name compatibility; ambiguous names return 409 before a +mutation/export response. Bidirectional hostname aliases preserve stored names +and histories. Exact names win. Legacy POST cannot acknowledge a local alias +when subsequent global writes would select another device. Query cache keys +include scope; heartbeat keys use device and resolved canonical bucket ID. +Metadata changes clear query and heartbeat state. + +`GET /api/0/devices` includes the local UUID even when empty, plus stored peers. +Peer hostname is a deterministic representative of bucket metadata, not a +latest-hostname registry. `GET /api/0/devices/buckets` groups metadata explicitly; +`GET /api/0/devices/export` uses the same envelope with events: + +```json +{"devices": {"source-uuid": {"buckets": {"window": {"id": "window", "device_id": "source-uuid", "type": "test", "client": "watcher", "hostname": "host", "created": "2024-01-01T00:00:00Z", "events": []}}}}} +``` + +`POST /api/0/devices/import` accepts that envelope, rejects empty or `local` +source identities, and validates enclosing ownership, bucket keys and event +payloads before writes. Local exported ownership becomes the source UUID, so +history stays peer-owned at another server. Python's existing reject-on-existing +import policy remains (unlike Rust's deduplicating import). Individual groups +use legacy rollback; a runtime failure can leave earlier successful groups. +Metadata `data` is preserved. Export currently materializes events in memory; +streamed/snapshot export parity is unfinished. No watcher adoption or sync-v2 +import is enabled by this draft. + +Run `make test` for the normal fast server slice plus the device HTTP contract. +The pinned draft core is required; tests use temporary file-backed SQLite and +never a production database. Full shared HTTP parity and backend-specific +migration/same-name acceptance still gate removal of Rust's global name guards. diff --git a/aw_server/api.py b/aw_server/api.py index e42af27..feb5b33 100644 --- a/aw_server/api.py +++ b/aw_server/api.py @@ -1,3 +1,4 @@ +import copy import functools import json import logging @@ -16,6 +17,8 @@ from uuid import uuid4 import iso8601 +import jsonschema +from aw_core import schema from aw_core.dirs import get_data_dir from aw_core.log import get_log_file_path from aw_core.models import Event @@ -25,7 +28,7 @@ from .__about__ import __version__ from .csv_export import events_to_csv -from .exceptions import NotFound +from .exceptions import BadRequest, NotFound from .profile import profile_from_env from .query_cache import QueryCache, event_range from .settings import Settings @@ -57,8 +60,10 @@ def get_device_id() -> str: def check_bucket_exists(f): @functools.wraps(f) def g(self, bucket_id, *args, **kwargs): - if bucket_id not in self.db.buckets(): + if not self.db.has_bucket(bucket_id): raise NotFound("NoSuchBucket", f"There's no bucket named {bucket_id}") + # Aliases and canonical names must share heartbeat state as well as SQL rows. + bucket_id = self.db[bucket_id].metadata()["id"] return f(self, bucket_id, *args, **kwargs) return g @@ -87,12 +92,28 @@ def __init__(self, db, testing, query_cache: bool = True) -> None: # Every write below must invalidate the cache (see query_cache.py). self.query_cache: Optional[QueryCache] = QueryCache() if query_cache else None self._write_lock = threading.RLock() + self._device_scope: Optional[str] = None + + def for_device(self, device_id: str) -> "ServerAPI": + """Immutable request view; never switch the shared API's datastore.""" + if not device_id: + raise BadRequest("InvalidDevice", "Device scope must not be empty") + scope = "local" if device_id in ("local", get_device_id()) else device_id + scoped = copy.copy(self) + scoped.db = self.db.for_device(scope) + scoped._device_scope = scope + return scoped + + def _heartbeat_key(self, bucket_id: str): + metadata = self.db[bucket_id].metadata() + return (metadata.get("device_id", "local"), metadata["id"]) def _invalidate(self, ranges) -> None: if self.query_cache: self.query_cache.invalidate(ranges) def _invalidate_all(self) -> None: + self.last_event.clear() if self.query_cache: self.query_cache.clear() @@ -108,8 +129,104 @@ def get_info(self) -> Dict[str, Any]: return payload def get_devices(self) -> List[Dict[str, Any]]: - """Get list of known devices (currently only the local device)""" - return [{"id": get_device_id(), "hostname": gethostname()}] + """Local device plus peers with persisted bucket metadata.""" + groups = self.device_buckets()["devices"] + return [ + { + "id": device, + "hostname": gethostname() + if device == get_device_id() + else min( + (b["hostname"] for b in group["buckets"].values()), default="" + ), + } + for device, group in sorted(groups.items()) + ] + + def device_buckets(self, export: bool = False) -> Dict[str, Any]: + """Explicit grouping; never encode identity into a legacy bucket ID.""" + public_local = get_device_id() + groups: Dict[str, Any] = {public_local: {"buckets": {}}} + storage = self.db.storage_strategy + if hasattr(storage, "device_buckets"): + grouped = storage.device_buckets() + else: + # Non-device-aware backends can still expose their local catalogue. + grouped = {"local": self.db.buckets()} + for device, buckets in grouped.items(): + public = public_local if device == "local" else device + target = groups.setdefault(public, {"buckets": {}})["buckets"] + scoped = ( + self.for_device(device) if hasattr(storage, "device_buckets") else self + ) + for bid, metadata in buckets.items(): + if bid in target: + raise ValueError( + f"Ambiguous bucket '{bid}'; normalized device collision" + ) + bucket = scoped.export_bucket(bid) if export else dict(metadata) + bucket["device_id"] = public + target[bid] = bucket + return {"devices": groups} + + @_serialized + def import_devices(self, envelope: Any) -> None: + """Preflight all identities and event payloads before any writes. + + Existing buckets are rejected, matching Python's legacy import policy. + Each group uses its existing rollback contract; earlier successful groups + may remain on a runtime failure. This is not an atomic multi-device import. + """ + if not isinstance(envelope, dict) or not isinstance( + envelope.get("devices"), dict + ): + raise ValueError("Expected an object with a devices map") + prepared = [] + for device, group in envelope["devices"].items(): + if not isinstance(device, str) or not device or device == "local": + raise ValueError( + "Imported device must be a nonempty source identity, not local" + ) + if not isinstance(group, dict) or not isinstance( + group.get("buckets"), dict + ): + raise ValueError("Each device must contain a buckets map") + scoped = self.for_device(device) + buckets = copy.deepcopy(group["buckets"]) + for bid, bucket in buckets.items(): + if ( + not isinstance(bucket, dict) + or bucket.get("id") != bid + or bucket.get("device_id") != device + ): + raise ValueError( + "Bucket key and device metadata must match their enclosing identity" + ) + for field in ("type", "client", "hostname", "created"): + if not isinstance(bucket.get(field), str): + raise ValueError(f"Invalid bucket field {field}") + if bucket["hostname"] == "!local" and scoped._device_scope != "local": + raise ValueError("Peer hostname must not be !local") + iso8601.parse_date(bucket["created"]) + if bucket.get("data") is not None and not isinstance( + bucket["data"], dict + ): + raise ValueError("Bucket data must be an object") + if not isinstance(bucket.get("events"), list): + raise ValueError("Bucket events must be a list") + for event in bucket["events"]: + try: + jsonschema.validate(event, schema.get_json_schema("event")) + except jsonschema.ValidationError as e: + raise ValueError(f"Invalid event: {e.message}") from e + Event(**event) + if scoped.db.has_bucket(bid): + raise ValueError( + f"Bucket '{bid}' already exists in device '{device}'" + ) + prepared.append((scoped, buckets)) + for scoped, buckets in prepared: + scoped.import_all(buckets) def get_buckets(self) -> Dict[str, Dict]: """Get dict {bucket_name: Bucket} of all buckets""" @@ -176,6 +293,7 @@ def import_bucket(self, bucket_data: Any): type=bucket_data["type"], client=bucket_data["client"], hostname=bucket_data["hostname"], + data=bucket_data.get("data"), created=( bucket_data["created"] if isinstance(bucket_data["created"], datetime) @@ -273,7 +391,36 @@ def create_bucket( """ if created is None: created = datetime.now() - if bucket_id in self.db.buckets(): + # Legacy POST is local-only. Unsupported legacy backends retain their + # existing behavior; explicit scoped requests fail closed in for_device. + if self._device_scope is None: + try: + local = self.for_device("local") + except NotImplementedError: + local = None + if local is not None: + if self.db.has_bucket(bucket_id): + global_metadata = self.db[bucket_id].metadata() + local_metadata = ( + local.db[bucket_id].metadata() + if local.db.has_bucket(bucket_id) + else None + ) + if local_metadata is None or ( + global_metadata.get("device_id"), + global_metadata["id"], + ) != (local_metadata.get("device_id"), local_metadata["id"]): + raise ValueError( + f"Ambiguous bucket '{bucket_id}'; local POST and global writes disagree" + ) + return local.create_bucket( + bucket_id, event_type, client, hostname, created, data + ) + if hostname == "!local" and self._device_scope not in (None, "local"): + raise BadRequest( + "InvalidHostname", "!local is only valid for the local device" + ) + if self.db.has_bucket(bucket_id): return False if hostname == "!local": info = self.get_info() @@ -308,7 +455,7 @@ def update_bucket( try: self.db.update_bucket( bucket_id, - type=event_type, + type_id=event_type, client=client, hostname=hostname, data=data, @@ -326,7 +473,6 @@ def delete_bucket(self, bucket_id: str) -> None: self.db.delete_bucket(bucket_id) finally: self._invalidate_all() - self.last_event.pop(bucket_id, None) logger.debug(f"Deleted bucket '{bucket_id}'") return None @@ -378,7 +524,7 @@ def create_events(self, bucket_id: str, events: List[Event]) -> List[Event]: affected.append(event_range(old)) # The cached last heartbeat may no longer be the bucket's last event, # and a heartbeat merging into it would replace the wrong one. - self.last_event.pop(bucket_id, None) + self.last_event.pop(self._heartbeat_key(bucket_id), None) try: if len(events) == 1: # Pass as single Event so Bucket.insert uses insert_one (returns Event with ID) @@ -415,7 +561,7 @@ def delete_event(self, bucket_id: str, event_id) -> bool: # subsequent heartbeat would merge into the deleted event and # replace_last() would overwrite whatever is now the stored # last event. Drop the cache so it is re-read from storage. - self.last_event.pop(bucket_id, None) + self.last_event.pop(self._heartbeat_key(bucket_id), None) @check_bucket_exists @_serialized @@ -440,7 +586,7 @@ def delete_events(self, bucket_id: str, event_ids: List[int]) -> int: self._invalidate(ranges) # Same reason as delete_event: a deleted event may be the # cached heartbeat last_event. - self.last_event.pop(bucket_id, None) + self.last_event.pop(self._heartbeat_key(bucket_id), None) @check_bucket_exists @_serialized @@ -485,13 +631,14 @@ def heartbeat(self, bucket_id: str, heartbeat: Event, pulsetime: float) -> Event # That way we could double check that the event has been applied # and if it hasn't we simply replace it with the updated counterpart. + heartbeat_key = self._heartbeat_key(bucket_id) last_event = None - if bucket_id not in self.last_event: + if heartbeat_key not in self.last_event: last_events = self.db[bucket_id].get(limit=1) if len(last_events) > 0: last_event = last_events[0] else: - last_event = self.last_event[bucket_id] + last_event = self.last_event[heartbeat_key] if last_event: if last_event.data == heartbeat.data: @@ -510,7 +657,7 @@ def heartbeat(self, bucket_id: str, heartbeat: Event, pulsetime: float) -> Event # out-of-order heartbeat), so invalidate its range too. stored = self.db[bucket_id].get(limit=1) affected += [event_range(e) for e in stored] - self.last_event[bucket_id] = merged + self.last_event[heartbeat_key] = merged try: # Replace the event we merged into by id. replace_last() # targets whichever event sorts last by timestamp, which @@ -547,7 +694,7 @@ def heartbeat(self, bucket_id: str, heartbeat: Event, pulsetime: float) -> Event self.db[bucket_id].insert(heartbeat) finally: self._invalidate([event_range(heartbeat)]) - self.last_event[bucket_id] = heartbeat + self.last_event[heartbeat_key] = heartbeat return heartbeat def query2(self, name, query, timeperiods, cache: bool = True): @@ -577,7 +724,8 @@ def query2(self, name, query, timeperiods, cache: bool = True): if qc is None or not qc.cacheable(period): result.append(query2.query(name, query, starttime, endtime, self.db)) continue - key = QueryCache.key(query, period) + # JSON tuple encoding is collision-safe even for delimiter-bearing IDs. + key = QueryCache.key(json.dumps([self._device_scope, query]), period) cached = qc.get(key) if cached is not None: result.append(cached) diff --git a/aw_server/rest.py b/aw_server/rest.py index 2f89cdf..c091229 100644 --- a/aw_server/rest.py +++ b/aw_server/rest.py @@ -14,6 +14,7 @@ Blueprint, Response, current_app, + g, jsonify, make_response, request, @@ -180,6 +181,29 @@ def decorator(f): return decorator +def request_api() -> ServerAPI: + return getattr(g, "device_api", current_app.api) + + +class DeviceResource(Resource): + """Reuse legacy validation/auth/streaming with a request-local API view.""" + + def dispatch_request(self, *args, **kwargs): + device_id = kwargs.pop("device_id", None) + if device_id is not None: + try: + g.device_api = request_api().for_device(device_id) + except NotImplementedError as e: + return {"message": str(e)}, 501 + return super().dispatch_request(*args, **kwargs) + + +@api.errorhandler(ValueError) +def storage_value_error(error): + status = 409 if str(error).startswith("Ambiguous bucket") else 400 + return {"message": str(error)}, status + + # SERVER INFO @@ -188,7 +212,7 @@ class InfoResource(Resource): @api.marshal_with(info) @copy_doc(ServerAPI.get_info) def get(self) -> Dict[str, Dict]: - return current_app.api.get_info() + return request_api().get_info() @api.route("/0/devices") @@ -196,26 +220,56 @@ class DevicesResource(Resource): @api.marshal_list_with(device) @copy_doc(ServerAPI.get_devices) def get(self): - return current_app.api.get_devices() + return request_api().get_devices() + + +@api.route("/0/devices/buckets") +class DeviceBucketsResource(Resource): + def get(self): + return current_app.api.device_buckets() + + +@api.route("/0/devices/export") +class DevicesExportResource(Resource): + def get(self): + return jsonify(current_app.api.device_buckets(export=True)) + + +@api.route("/0/devices/import") +class DevicesImportResource(Resource): + def post(self): + try: + current_app.api.import_devices(request.get_json()) + except NotImplementedError as e: + return {"message": str(e)}, 501 + except ImportRollbackError as e: + logger.error(f"Device import failed with incomplete rollback: {e!r}") + return {"message": str(e)}, 500 + except (ValueError, TypeError, KeyError) as e: + return {"message": str(e)}, 400 + return {"message": "Import successful"}, 200 # BUCKETS -@api.route("/0/buckets/") -class BucketsResource(Resource): +@api.route("/0/buckets/", "/0/devices//buckets/") +class BucketsResource(DeviceResource): # TODO: Add response marshalling/validation @copy_doc(ServerAPI.get_buckets) def get(self) -> Dict[str, Dict]: - return current_app.api.get_buckets() + return request_api().get_buckets() -@api.route("/0/buckets/") -class BucketResource(Resource): +@api.route( + "/0/buckets/", + "/0/devices//buckets/", +) +class BucketResource(DeviceResource): @api.doc(model=bucket) @copy_doc(ServerAPI.get_bucket_metadata) def get(self, bucket_id): - return current_app.api.get_bucket_metadata(bucket_id) + return request_api().get_bucket_metadata(bucket_id) @api.expect(create_bucket) @copy_doc(ServerAPI.create_bucket) @@ -242,7 +296,7 @@ def post(self, bucket_id): ) except iso8601.ParseError as e: raise BadRequest("InvalidCreated", str(e)) - bucket_created = current_app.api.create_bucket( + bucket_created = request_api().create_bucket( bucket_id, event_type=data["type"], client=data["client"], @@ -259,12 +313,12 @@ def post(self, bucket_id): @copy_doc(ServerAPI.update_bucket) def put(self, bucket_id): data = request.get_json() - current_app.api.update_bucket( + request_api().update_bucket( bucket_id, - event_type=data["type"], - client=data["client"], - hostname=data["hostname"], - data=data["data"], + event_type=data.get("type"), + client=data.get("client"), + hostname=data.get("hostname"), + data=data.get("data"), ) return {}, 200 @@ -272,20 +326,23 @@ def put(self, bucket_id): @api.param("force", "Needs to be =1 to delete a bucket it non-testing mode") def delete(self, bucket_id): args = request.args - if not current_app.api.testing: + if not request_api().testing: if "force" not in args or args["force"] != "1": msg = "Deleting buckets is only permitted if aw-server is running in testing mode or if ?force=1" raise Unauthorized("DeleteBucketUnauthorized", msg) - current_app.api.delete_bucket(bucket_id) + request_api().delete_bucket(bucket_id) return {}, 200 # EVENTS -@api.route("/0/buckets//events") -class EventsResource(Resource): +@api.route( + "/0/buckets//events", + "/0/devices//buckets//events", +) +class EventsResource(DeviceResource): # For some reason this doesn't work with the JSONSchema variant # Marshalling doesn't work with JSONSchema events # @api.marshal_list_with(event) @@ -302,9 +359,7 @@ def get(self, bucket_id): start = _parse_arg(args, "start", iso8601.parse_date) end = _parse_arg(args, "end", iso8601.parse_date) - events = current_app.api.get_events( - bucket_id, limit=limit, start=start, end=end - ) + events = request_api().get_events(bucket_id, limit=limit, start=start, end=end) return events, 200 # TODO: How to tell expect that it could be a list of events? Until then we can't use validate. @@ -325,12 +380,15 @@ def post(self, bucket_id): else: raise BadRequest("Invalid POST data", "") - events = current_app.api.create_events(bucket_id, events) + events = request_api().create_events(bucket_id, events) return [e.to_json_dict() for e in events], 200 -@api.route("/0/buckets//events/count") -class EventCountResource(Resource): +@api.route( + "/0/buckets//events/count", + "/0/devices//buckets//events/count", +) +class EventCountResource(DeviceResource): @api.doc(model=fields.Integer) @api.param("start", "Start date of eventcount") @api.param("end", "End date of eventcount") @@ -340,7 +398,7 @@ def get(self, bucket_id): start = _parse_arg(args, "start", iso8601.parse_date) end = _parse_arg(args, "end", iso8601.parse_date) - events = current_app.api.get_eventcount(bucket_id, start=start, end=end) + events = request_api().get_eventcount(bucket_id, start=start, end=end) return events, 200 @@ -349,8 +407,11 @@ def get(self, bucket_id): MAX_BULK_DELETE_IDS = 10_000 -@api.route("/0/buckets//events/delete") -class EventsDeleteResource(Resource): +@api.route( + "/0/buckets//events/delete", + "/0/devices//buckets//events/delete", +) +class EventsDeleteResource(DeviceResource): @copy_doc(ServerAPI.delete_events) def post(self, bucket_id: str): data = request.get_json(silent=True) @@ -369,18 +430,21 @@ def post(self, bucket_id: str): f"Too many ids in bulk delete request: {len(ids)} " f"(max {MAX_BULK_DELETE_IDS})" ) - return current_app.api.delete_events(bucket_id, ids), 200 + return request_api().delete_events(bucket_id, ids), 200 -@api.route("/0/buckets//events/") -class EventResource(Resource): +@api.route( + "/0/buckets//events/", + "/0/devices//buckets//events/", +) +class EventResource(DeviceResource): @api.doc(model=event) @copy_doc(ServerAPI.get_event) def get(self, bucket_id: str, event_id: int): logger.debug( f"Received get request for event with id '{event_id}' in bucket '{bucket_id}'" ) - event = current_app.api.get_event(bucket_id, event_id) + event = request_api().get_event(bucket_id, event_id) if event: return event, 200 else: @@ -393,12 +457,15 @@ def delete(self, bucket_id: str, event_id: int): event_id, bucket_id ) ) - success = current_app.api.delete_event(bucket_id, event_id) + success = request_api().delete_event(bucket_id, event_id) return {"success": success}, 200 -@api.route("/0/buckets//heartbeat") -class HeartbeatResource(Resource): +@api.route( + "/0/buckets//heartbeat", + "/0/devices//buckets//heartbeat", +) +class HeartbeatResource(DeviceResource): # Class-level lock shared across all instances. # Flask-RESTX creates a new Resource instance per request, so an # instance-level lock would provide no mutual exclusion. @@ -429,7 +496,7 @@ def post(self, bucket_id): ) return {"message": "Server busy, try again later"}, 503 try: - event = current_app.api.heartbeat(bucket_id, heartbeat, pulsetime) + event = request_api().heartbeat(bucket_id, heartbeat, pulsetime) finally: self.lock.release() return event.to_json_dict(), 200 @@ -438,8 +505,8 @@ def post(self, bucket_id): # QUERY -@api.route("/0/query/") -class QueryResource(Resource): +@api.route("/0/query/", "/0/devices//query") +class QueryResource(DeviceResource): # TODO Docs @api.expect(query, validate=True) @api.param("name", "Name of the query") @@ -457,7 +524,7 @@ def post(self): if query.get("cache") is False: cache = False try: - result = current_app.api.query2( + result = request_api().query2( name, query["query"], query["timeperiods"], cache ) return jsonify(result) @@ -469,14 +536,15 @@ def post(self): # EXPORT AND IMPORT -@api.route("/0/export") -class ExportAllResource(Resource): +@api.route("/0/export", "/0/devices//export") +class ExportAllResource(DeviceResource): @api.doc(model=buckets_export) @copy_doc(ServerAPI.export_all) def get(self): - buckets_export = current_app.api.export_all() + buckets_export = request_api().export_all() payload = {"buckets": buckets_export} response = make_response(json.dumps(payload)) + response.mimetype = "application/json" filename = "aw-buckets-export.json" response.headers["Content-Disposition"] = "attachment; filename={}".format( filename @@ -485,14 +553,18 @@ def get(self): # TODO: Perhaps we don't need this, could be done with a query argument to /0/export instead -@api.route("/0/buckets//export") -class BucketExportResource(Resource): +@api.route( + "/0/buckets//export", + "/0/devices//buckets//export", +) +class BucketExportResource(DeviceResource): @api.doc(model=buckets_export) @copy_doc(ServerAPI.export_bucket) def get(self, bucket_id): - bucket_export = current_app.api.export_bucket(bucket_id) + bucket_export = request_api().export_bucket(bucket_id) payload = {"buckets": {bucket_export["id"]: bucket_export}} response = make_response(json.dumps(payload)) + response.mimetype = "application/json" filename = "aw-bucket-export_{}.json".format(bucket_export["id"]) response.headers["Content-Disposition"] = "attachment; filename={}".format( filename @@ -500,8 +572,11 @@ def get(self, bucket_id): return response -@api.route("/0/buckets//export/csv") -class BucketExportCsvResource(Resource): +@api.route( + "/0/buckets//export/csv", + "/0/devices//buckets//export/csv", +) +class BucketExportCsvResource(DeviceResource): @api.param("limit", "the maximum number of events to export") @api.param("start", "Start date of events") @api.param("end", "End date of events") @@ -514,7 +589,7 @@ def get(self, bucket_id): end = iso8601.parse_date(args["end"]) if "end" in args else None except (ValueError, iso8601.ParseError) as e: raise BadRequest("InvalidParameter", str(e)) - rows = current_app.api.export_bucket_csv( + rows = request_api().export_bucket_csv( bucket_id, limit=limit, start=start, end=end ) response = Response(stream_with_context(rows), mimetype="text/csv") @@ -548,11 +623,11 @@ def post(self): f"Bucket '{key}' appears in more than one uploaded file." ) buckets[key] = bucket - current_app.api.import_all(buckets) + request_api().import_all(buckets) # Normal import from body else: buckets = request.get_json()["buckets"] - current_app.api.import_all(buckets) + request_api().import_all(buckets) except ImportRollbackError as e: # The import failed *and* rollback could not remove some created # buckets: they are still stored, so a retry would be rejected as a @@ -576,7 +651,7 @@ def post(self): class LogResource(Resource): @copy_doc(ServerAPI.get_log) def get(self): - return current_app.api.get_log(), 200 + return request_api().get_log(), 200 # SETTINGS @@ -586,11 +661,11 @@ def get(self): @api.route("/0/settings/") class SettingsResource(Resource): def get(self, key: str): - data = current_app.api.get_setting(key) + data = request_api().get_setting(key) return jsonify(data) def post(self, key: str): if not key: raise BadRequest("MissingParameter", "Missing required parameter key") - data = current_app.api.set_setting(key, request.get_json()) + data = request_api().set_setting(key, request.get_json()) return data diff --git a/poetry.lock b/poetry.lock index 54ea262..42fcb8f 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.5.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand. [[package]] name = "aniso8601" @@ -59,25 +59,30 @@ name = "aw-core" version = "0.5.18" description = "Core library for ActivityWatch" optional = false -python-versions = "<4.0,>=3.8" +python-versions = "^3.8" groups = ["main"] -files = [ - {file = "aw_core-0.5.18-py3-none-any.whl", hash = "sha256:6e3df7b46e913d631e0dc8b1b10046c111284a9f4240bec8efd279ce4e438cd5"}, - {file = "aw_core-0.5.18.tar.gz", hash = "sha256:dc14867ed15b4529d6e69b1d07df3fb29c91ccf7ab174de3f9100b90a95cc826"}, -] +files = [] +develop = false [package.dependencies] click = "*" deprecation = "*" +idna = ">=3.0" iso8601 = "*" -jsonschema = ">=4.3,<5.0" +jsonschema = "^4.3" peewee = ">=3.0.0,<5.0.0" platformdirs = "3.10" -rfc3339-validator = ">=0.1.4,<0.2.0" -strict-rfc3339 = ">=0.7,<0.8" +rfc3339-validator = "^0.1.4" +strict-rfc3339 = "^0.7" timeslot = "*" tomlkit = "*" +[package.source] +type = "git" +url = "https://github.com/TimeToBuildBob/aw-core.git" +reference = "15fa628b7f85d57bd79f8d29abaa8718505d3b91" +resolved_reference = "15fa628b7f85d57bd79f8d29abaa8718505d3b91" + [[package]] name = "black" version = "24.8.0" @@ -429,11 +434,11 @@ files = [ [package.dependencies] aniso8601 = ">=0.82" -Flask = ">=0.8,!=2.0.0,<3.0.0" +Flask = ">=0.8,<2.0.0 || >2.0.0,<3.0.0" importlib-resources = "*" jsonschema = "<=4.17.3" pytz = "*" -werkzeug = "!=2.0.0,<3.0.0" +werkzeug = "<2.0.0 || >2.0.0,<3.0.0" [package.extras] dev = ["Faker (==2.0.0)", "black", "blinker", "invoke (==2.0.0)", "mock (==3.0.5)", "pytest (==7.0.1)", "pytest-benchmark (==3.4.1)", "pytest-cov (==4.0.0)", "pytest-flask (==1.2.0)", "pytest-mock (==3.6.1)", "pytest-profiling (==1.7.0)", "tox", "twine (==3.8.0)", "tzlocal"] @@ -569,7 +574,7 @@ files = [ attrs = ">=17.4.0" importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""} pkgutil-resolve-name = {version = ">=1.3.10", markers = "python_version < \"3.9\""} -pyrsistent = ">=0.14.0,!=0.17.0,!=0.17.1,!=0.17.2" +pyrsistent = ">=0.14.0,<0.17.0 || >0.17.0,<0.17.1 || >0.17.1,<0.17.2 || >0.17.2" [package.extras] format = ["fqdn", "idna", "isoduration", "jsonpointer (>1.13)", "rfc3339-validator", "rfc3987", "uri-template", "webcolors (>=1.11)"] @@ -1037,7 +1042,7 @@ description = "A lil' TOML parser" optional = false python-versions = ">=3.7" groups = ["dev"] -markers = "python_version < \"3.11\"" +markers = "python_full_version <= \"3.11.0a6\"" files = [ {file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"}, {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"}, @@ -1207,4 +1212,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = "^3.8" -content-hash = "48be42890bb2039993350e7078dfb3c783eff55955609c2c11e52d0368d60915" +content-hash = "f58788808b715d41df5300ce68eef24002ff52b6ea82bffeb7ef1ced7e621c6a" diff --git a/pyproject.toml b/pyproject.toml index 7b7f7ea..5becee7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,8 @@ aw-server = "aw_server:main" [tool.poetry.dependencies] python = "^3.8" -aw-core = "^0.5.18" +# Draft dependency: ActivityWatch/aw-core#183 (stacked on schema #177). +aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "15fa628b7f85d57bd79f8d29abaa8718505d3b91"} aw-client = "^0.5.8" flask = "^2.2" flask-restx = "^1.0.3" diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py new file mode 100644 index 0000000..fd83a88 --- /dev/null +++ b/tests/test_device_addressing.py @@ -0,0 +1,355 @@ +"""Device HTTP contract against the real pair-keyed SQLite datastore.""" +from functools import partial + +import pytest +from aw_datastore.storages import SqliteStorage + +from aw_server.server import AWFlask + + +@pytest.fixture() +def device_app(tmp_path, monkeypatch): + monkeypatch.setattr("aw_server.api.get_device_id", lambda: "server-uuid") + app = AWFlask( + "127.0.0.1", + testing=True, + storage_method=partial(SqliteStorage, filepath=str(tmp_path / "events.db")), + cors_origins=[], + ) + yield app + app.api.db.storage_strategy.conn.close() + + +def create(client, device, bid="window", hostname="host"): + return client.post( + f"/api/0/devices/{device}/buckets/{bid}", + json={ + "client": "test", + "type": "test", + "hostname": hostname, + }, + ) + + +def event(label, second=0): + return { + "timestamp": f"2024-01-01T12:00:{second:02d}Z", + "duration": 1, + "data": {"label": label}, + } + + +def test_scoped_crud_uuid_alias_and_legacy_conflict(device_app): + c = device_app.test_client() + assert create(c, "local").status_code == 200 + assert create(c, "server-uuid").status_code == 304 + assert create(c, "peer").status_code == 200 + for device in ("local", "peer"): + base = f"/api/0/devices/{device}/buckets/window" + r = c.post(base + "/events", json=event(device)) + assert r.status_code == 200 + eid = r.json[0]["id"] + assert c.get(base + "/events/count").json == 1 + assert c.get(base + f"/events/{eid}").json["data"] == {"label": device} + assert c.get(base).json["device_id"] == device + assert c.get("/api/0/devices/server-uuid/buckets/window/events").json[0][ + "data" + ] == {"label": "local"} + for path in ("/buckets/", "/buckets/window", "/buckets/window/events", "/export"): + assert c.get("/api/0" + path).status_code == 409 + assert c.delete("/api/0/devices/local/buckets/window").status_code == 200 + assert c.get("/api/0/devices/peer/buckets/window/events/count").json == 1 + + +def test_query_cache_is_scoped_and_invalidated(device_app): + c = device_app.test_client() + query = { + "query": ['RETURN = query_bucket("window");'], + "timeperiods": ["2024-01-01T00:00:00Z/2024-01-02T00:00:00Z"], + } + for device in ("local", "peer"): + assert create(c, device).status_code == 200 + assert ( + c.post( + f"/api/0/devices/{device}/buckets/window/events", json=event(device) + ).status_code + == 200 + ) + for _ in range(2): + r = c.post(f"/api/0/devices/{device}/query", json=query) + assert r.status_code == 200 + assert r.json[0][0]["data"] == {"label": device} + assert c.post("/api/0/query/", json=query).status_code == 409 + assert ( + c.post( + "/api/0/devices/server-uuid/buckets/window/events", json=event("new", 5) + ).status_code + == 200 + ) + assert len(c.post("/api/0/devices/local/query", json=query).json[0]) == 2 + assert len(c.post("/api/0/devices/peer/query", json=query).json[0]) == 1 + + +def test_alias_heartbeat_and_metadata_repair(device_app): + c = device_app.test_client() + assert create(c, "local", "window_old_host", "old_host").status_code == 200 + assert create(c, "local", "window", "old_host").status_code == 304 + for device, bid in (("local", "window"), ("server-uuid", "window_old_host")): + r = c.post( + f"/api/0/devices/{device}/buckets/{bid}/heartbeat?pulsetime=10", + json=event("same", 0 if device == "local" else 5), + ) + assert r.status_code == 200 + assert c.get("/api/0/devices/local/buckets/window/events/count").json == 1 + assert create(c, "peer", "window", "!local").status_code == 400 + # The storage anti-shadowing guard deliberately still rejects an exact peer + # name that would redirect the established global alias. + assert create(c, "peer", "window", "peer_host").status_code == 409 + assert create(c, "peer", "window_old_host", "peer_host").status_code == 200 + assert ( + c.post( + "/api/0/devices/peer/buckets/window_old_host/heartbeat?pulsetime=10", + json=event("peer", 6), + ).status_code + == 200 + ) + assert c.get("/api/0/devices/local/buckets/window/events").json[0]["data"] == { + "label": "same" + } + assert ( + c.put( + "/api/0/devices/local/buckets/window_old_host", + json={"hostname": "new_host"}, + ).status_code + == 200 + ) + assert c.get("/api/0/devices/local/buckets/window").status_code == 404 + assert c.get("/api/0/devices/local/buckets/window_old_host").status_code == 200 + + +def test_legacy_post_is_local_and_alias_aware(device_app): + c = device_app.test_client() + assert create(c, "peer", "peer_only").status_code == 200 + r = c.post( + "/api/0/buckets/local_new", + json={ + "client": "test", + "type": "test", + "hostname": "!local", + "device_id": "peer", + }, + ) + assert r.status_code == 200 + assert c.get("/api/0/devices/local/buckets/local_new").json["device_id"] == "local" + assert c.get("/api/0/devices/peer/buckets/local_new").status_code == 404 + assert create(c, "local", "old_host", "host").status_code == 200 + assert ( + c.post( + "/api/0/buckets/old", + json={"client": "test", "type": "test", "hostname": "host"}, + ).status_code + == 304 + ) + + +def test_grouped_catalogue_export_roundtrip_and_preflight( + device_app, tmp_path, monkeypatch +): + c = device_app.test_client() + assert c.get("/api/0/devices/buckets").json == { + "devices": {"server-uuid": {"buckets": {}}} + } + for device in ("local", "peer"): + assert create(c, device).status_code == 200 + assert ( + c.post( + f"/api/0/devices/{device}/buckets/window/events", json=event(device) + ).status_code + == 200 + ) + assert {d["id"] for d in c.get("/api/0/devices").json} == {"server-uuid", "peer"} + exported = c.get("/api/0/devices/export") + assert exported.status_code == 200 + assert ( + exported.json["devices"]["server-uuid"]["buckets"]["window"]["device_id"] + == "server-uuid" + ) + monkeypatch.setattr("aw_server.api.get_device_id", lambda: "destination-uuid") + destination = AWFlask( + "127.0.0.1", + testing=True, + cors_origins=[], + storage_method=partial( + SqliteStorage, filepath=str(tmp_path / "destination.db") + ), + ) + try: + dc = destination.test_client() + assert dc.post("/api/0/devices/import", json=exported.json).status_code == 200 + assert dc.get("/api/0/devices/local/buckets/window").status_code == 404 + for device in ("server-uuid", "peer"): + assert dc.get(f"/api/0/devices/{device}/buckets/window/events").json[0][ + "data" + ]["label"] == ("local" if device == "server-uuid" else "peer") + # Reject the entire malformed envelope before writing the first group. + invalid = { + "devices": { + "fresh": { + "buckets": { + "fresh": { + "id": "fresh", + "device_id": "fresh", + "hostname": "h", + "client": "c", + "type": "t", + "created": "2024-01-01T00:00:00Z", + "events": [], + } + } + }, + "local": {"buckets": {}}, + } + } + assert dc.post("/api/0/devices/import", json=invalid).status_code == 400 + assert dc.get("/api/0/devices/fresh/buckets/").json == {} + finally: + destination.api.db.storage_strategy.conn.close() + + +def test_scoped_export_delete_and_api_key(device_app): + from aw_server import auth + + auth.register(device_app, "secret") + c = device_app.test_client() + for path in ( + "/api/0/devices/local/buckets/", + "/api/0/devices/export", + "/api/0/devices/buckets", + ): + assert c.get(path).status_code == 401 + c.environ_base["HTTP_AUTHORIZATION"] = "Bearer secret" + assert create(c, "local").status_code == 200 + base = "/api/0/devices/local/buckets/window" + eid = c.post(base + "/events", json=event("local")).json[0]["id"] + assert c.get(base + "/export").json["buckets"]["window"]["events"][0]["data"] == { + "label": "local" + } + assert c.get(base + "/export/csv").status_code == 200 + assert c.post(base + "/events/delete", json={"ids": [eid]}).status_code == 200 + assert c.get(base + "/events/count").json == 0 + + +def test_import_preserves_metadata_and_rejects_alias_split(device_app): + c = device_app.test_client() + bucket = { + "id": "history_host", + "device_id": "source", + "client": "c", + "type": "t", + "hostname": "host", + "created": "2024-01-01T00:00:00Z", + "data": {"custom": "preserved"}, + "events": [event("source")], + } + envelope = {"devices": {"source": {"buckets": {"history_host": bucket}}}} + assert c.post("/api/0/devices/import", json=envelope).status_code == 200 + assert c.get("/api/0/devices/source/buckets/history").json["data"] == bucket["data"] + # A later import alias cannot create an empty canonical row over history. + bucket["id"] = "history" + envelope["devices"]["source"]["buckets"] = {"history": bucket} + assert c.post("/api/0/devices/import", json=envelope).status_code == 400 + assert c.get("/api/0/devices/source/buckets/history/events/count").json == 1 + + +def test_scoped_heartbeat_invalidates_after_insert_and_delete(device_app): + c = device_app.test_client() + assert create(c, "local").status_code == 200 + base = "/api/0/devices/local/buckets/window" + assert c.post(base + "/heartbeat?pulsetime=10", json=event("a")).status_code == 200 + inserted = c.post(base + "/events", json=event("b", 2)).json[0]["id"] + assert ( + c.post(base + "/heartbeat?pulsetime=10", json=event("a", 4)).status_code == 200 + ) + assert c.get(base + "/events/count").json == 3 + assert c.delete(base + f"/events/{inserted}").status_code == 200 + assert c.get(base + "/events/count").json == 2 + + +def test_unsupported_scoped_backend_fails_closed(monkeypatch): + monkeypatch.setattr("aw_server.api.get_device_id", lambda: "server-uuid") + c = AWFlask("127.0.0.1", testing=True, cors_origins=[]).test_client() + assert create(c, "peer").status_code == 501 + assert c.get("/api/0/buckets/").json == {} + + +def test_legacy_post_does_not_acknowledge_alias_for_a_different_global_target( + device_app, +): + db = device_app.api.db.storage_strategy + local = db.for_device("local") + local.create_bucket("window_host", "t", "c", "host", "2024-01-01T00:00:00Z") + # Simulate a pre-existing catalogue from before the anti-shadowing guard. + db.conn.execute( + "INSERT INTO buckets(id, device_id, type, client, hostname, created, datastr) VALUES (?, ?, ?, ?, ?, ?, ?)", + ("window", "peer", "t", "c", "peer-host", "2024-01-01T00:00:00Z", "{}"), + ) + db.commit() + c = device_app.test_client() + assert ( + c.post( + "/api/0/buckets/window", + json={"client": "c", "type": "t", "hostname": "host"}, + ).status_code + == 409 + ) + assert ( + c.post( + "/api/0/devices/local/buckets/window", + json={"client": "c", "type": "t", "hostname": "host"}, + ).status_code + == 304 + ) + + +def test_catalogue_internal_error_is_not_hidden_as_local_only(device_app, monkeypatch): + def broken(): + raise AttributeError("catalogue failure") + + monkeypatch.setattr(device_app.api.db.storage_strategy, "device_buckets", broken) + assert device_app.test_client().get("/api/0/devices/buckets").status_code == 500 + + +def test_grouped_import_unsupported_backend_returns_501(): + c = AWFlask("127.0.0.1", testing=True, cors_origins=[]).test_client() + assert ( + c.post( + "/api/0/devices/import", json={"devices": {"peer": {"buckets": {}}}} + ).status_code + == 501 + ) + + +def test_malformed_later_event_writes_no_groups(device_app): + c = device_app.test_client() + bucket = { + "id": "history", + "device_id": "first", + "client": "c", + "type": "t", + "hostname": "host", + "created": "2024-01-01T00:00:00Z", + "events": [event("ok")], + } + import copy + + other = copy.deepcopy(bucket) + other["device_id"] = "second" + other["events"][0]["data"] = ["not an object"] + envelope = { + "devices": { + "first": {"buckets": {"history": bucket}}, + "second": {"buckets": {"history": other}}, + } + } + assert c.post("/api/0/devices/import", json=envelope).status_code == 400 + assert c.get("/api/0/devices/first/buckets/").json == {} From 0a4ded7a38067c672fed9c6ed122be0422402017 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 12:34:16 +0000 Subject: [PATCH 04/11] feat(export): stage grouped device export from one SQLite snapshot Git-Session-Id: 931478ca-2d2b-48c5-bb99-0bb5ad608478 --- README.md | 10 ++-- aw_server/api.py | 81 +++++++++++++++++++++++++++++---- aw_server/rest.py | 19 +++++++- poetry.lock | 6 +-- pyproject.toml | 2 +- tests/test_device_addressing.py | 79 ++++++++++++++++++++++++++++++++ 6 files changed, 180 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 05656bc..a8fe815 100644 --- a/README.md +++ b/README.md @@ -90,9 +90,13 @@ payloads before writes. Local exported ownership becomes the source UUID, so history stays peer-owned at another server. Python's existing reject-on-existing import policy remains (unlike Rust's deduplicating import). Individual groups use legacy rollback; a runtime failure can leave earlier successful groups. -Metadata `data` is preserved. Export currently materializes events in memory; -streamed/snapshot export parity is unfinished. No watcher adoption or sync-v2 -import is enabled by this draft. +Metadata `data` is preserved. Grouped export streams SQLite rows from one read +snapshot into a private temporary file before responding; failures cannot send +partial JSON with a success status. The response owns the staged file, not a +live database cursor. Concurrent WAL writes are excluded from that snapshot. +Only metadata and one event at a time are retained in memory; legacy exports +keep their existing implementation. No watcher adoption or sync-v2 import is +enabled by this draft. Run `make test` for the normal fast server slice plus the device HTTP contract. The pinned draft core is required; tests use temporary file-backed SQLite and diff --git a/aw_server/api.py b/aw_server/api.py index feb5b33..b67a587 100644 --- a/aw_server/api.py +++ b/aw_server/api.py @@ -3,6 +3,7 @@ import json import logging import threading +import tempfile from datetime import datetime from pathlib import Path from socket import gethostname @@ -134,16 +135,18 @@ def get_devices(self) -> List[Dict[str, Any]]: return [ { "id": device, - "hostname": gethostname() - if device == get_device_id() - else min( - (b["hostname"] for b in group["buckets"].values()), default="" + "hostname": ( + gethostname() + if device == get_device_id() + else min( + (b["hostname"] for b in group["buckets"].values()), default="" + ) ), } for device, group in sorted(groups.items()) ] - def device_buckets(self, export: bool = False) -> Dict[str, Any]: + def device_buckets(self) -> Dict[str, Any]: """Explicit grouping; never encode identity into a legacy bucket ID.""" public_local = get_device_id() groups: Dict[str, Any] = {public_local: {"buckets": {}}} @@ -156,19 +159,79 @@ def device_buckets(self, export: bool = False) -> Dict[str, Any]: for device, buckets in grouped.items(): public = public_local if device == "local" else device target = groups.setdefault(public, {"buckets": {}})["buckets"] - scoped = ( - self.for_device(device) if hasattr(storage, "device_buckets") else self - ) for bid, metadata in buckets.items(): if bid in target: raise ValueError( f"Ambiguous bucket '{bid}'; normalized device collision" ) - bucket = scoped.export_bucket(bid) if export else dict(metadata) + bucket = dict(metadata) bucket["device_id"] = public target[bid] = bucket return {"devices": groups} + @_serialized + def stage_device_export(self): + """Stream one SQLite snapshot into a private file before HTTP success. + + The response owns the returned file. No database cursor or transaction + survives into response iteration, which may run on a different thread. + """ + storage = self.db.storage_strategy + if not hasattr(storage, "device_buckets"): + raise NotImplementedError("Snapshot device export requires SQLite") + conn = storage.conn + # Flush lazy writes before beginning an independent read snapshot. + storage.commit() + staged = tempfile.TemporaryFile(mode="w+b") + + def write(value): + staged.write(value.encode("utf-8")) + + try: + conn.execute("BEGIN") + groups = self.device_buckets()["devices"] + write('{"devices":{') + first_device = True + for device, group in groups.items(): + if not first_device: + write(",") + first_device = False + write(json.dumps(device) + ':{"buckets":{') + first_bucket = True + scoped = self.for_device(device) + for bid, metadata in group["buckets"].items(): + if not first_bucket: + write(",") + first_bucket = False + write( + json.dumps(bid) + + ":" + + json.dumps(metadata)[:-1] + + ',"events":[' + ) + first_event = True + events = scoped.db.storage_strategy.iter_events(bid, flush=False) + try: + for event in events: + if not first_event: + write(",") + first_event = False + payload = event.to_json_dict() + payload.pop("id", None) + write(json.dumps(payload)) + finally: + events.close() + write("]}") + write("}}") + write("}}") + conn.commit() + staged.seek(0) + return staged + except BaseException: + conn.rollback() + staged.close() + raise + @_serialized def import_devices(self, envelope: Any) -> None: """Preflight all identities and event payloads before any writes. diff --git a/aw_server/rest.py b/aw_server/rest.py index c091229..5f03d4f 100644 --- a/aw_server/rest.py +++ b/aw_server/rest.py @@ -18,6 +18,7 @@ jsonify, make_response, request, + send_file, stream_with_context, ) from flask_restx import Api, Resource, fields @@ -232,7 +233,23 @@ def get(self): @api.route("/0/devices/export") class DevicesExportResource(Resource): def get(self): - return jsonify(current_app.api.device_buckets(export=True)) + try: + staged = current_app.api.stage_device_export() + except NotImplementedError as e: + return {"message": str(e)}, 501 + try: + response = send_file( + staged, + mimetype="application/json", + as_attachment=True, + download_name="aw-devices-export.json", + conditional=False, + ) + response.call_on_close(staged.close) + return response + except BaseException: + staged.close() + raise @api.route("/0/devices/import") diff --git a/poetry.lock b/poetry.lock index 42fcb8f..1540f62 100644 --- a/poetry.lock +++ b/poetry.lock @@ -80,8 +80,8 @@ tomlkit = "*" [package.source] type = "git" url = "https://github.com/TimeToBuildBob/aw-core.git" -reference = "15fa628b7f85d57bd79f8d29abaa8718505d3b91" -resolved_reference = "15fa628b7f85d57bd79f8d29abaa8718505d3b91" +reference = "4026a50e2ad191c43348c80ff5c34b010af5b2d6" +resolved_reference = "4026a50e2ad191c43348c80ff5c34b010af5b2d6" [[package]] name = "black" @@ -1212,4 +1212,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = "^3.8" -content-hash = "f58788808b715d41df5300ce68eef24002ff52b6ea82bffeb7ef1ced7e621c6a" +content-hash = "c0fed743f561fe81a5da1d1f8499b368551a87f4e3c1fa87865d2a9b0cd6ee03" diff --git a/pyproject.toml b/pyproject.toml index 5becee7..a77f30f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ aw-server = "aw_server:main" [tool.poetry.dependencies] python = "^3.8" # Draft dependency: ActivityWatch/aw-core#183 (stacked on schema #177). -aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "15fa628b7f85d57bd79f8d29abaa8718505d3b91"} +aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4026a50e2ad191c43348c80ff5c34b010af5b2d6"} aw-client = "^0.5.8" flask = "^2.2" flask-restx = "^1.0.3" diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index fd83a88..f6127af 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -1,4 +1,5 @@ """Device HTTP contract against the real pair-keyed SQLite datastore.""" + from functools import partial import pytest @@ -353,3 +354,81 @@ def test_malformed_later_event_writes_no_groups(device_app): } assert c.post("/api/0/devices/import", json=envelope).status_code == 400 assert c.get("/api/0/devices/first/buckets/").json == {} + + +def test_grouped_export_streams_events_without_bulk_read(device_app, monkeypatch): + c = device_app.test_client() + assert create(c, "local").status_code == 200 + assert ( + c.post( + "/api/0/devices/local/buckets/window/events", json=event("streamed") + ).status_code + == 200 + ) + + def bulk_read_forbidden(*args, **kwargs): + raise AssertionError("Grouped export must not materialize all event rows") + + monkeypatch.setattr( + device_app.api.db.storage_strategy.__class__, "get_events", bulk_read_forbidden + ) + response = c.get("/api/0/devices/export") + assert response.status_code == 200 + assert response.json["devices"]["server-uuid"]["buckets"]["window"]["events"][0][ + "data" + ] == {"label": "streamed"} + + +def test_grouped_export_stages_before_success_and_closes_snapshot( + device_app, monkeypatch +): + c = device_app.test_client() + assert create(c, "local").status_code == 200 + + def broken_events(*args, **kwargs): + yield event("partial") + raise RuntimeError("read failed") + + monkeypatch.setattr( + device_app.api.db.storage_strategy.__class__, "iter_events", broken_events + ) + response = c.get("/api/0/devices/export") + assert response.status_code == 500 + assert not device_app.api.db.storage_strategy.conn.in_transaction + # A failed staging operation must release the writer lock too. + assert create(c, "local", "after_failure").status_code == 200 + + +def test_export_keeps_one_snapshot_across_bucket_iterators(device_app): + import sqlite3 + import json + + api = device_app.api + storage = api.db.storage_strategy + path = storage.conn.execute("PRAGMA database_list").fetchone()[2] + storage.conn.execute("PRAGMA journal_mode=WAL") + for bid in ("a", "b"): + api.for_device("local").create_bucket(bid, "t", "c", "h") + original = storage.__class__.iter_events + changed = [] + + def events_with_concurrent_write(scoped, bid, **kwargs): + if not changed: + writer = sqlite3.connect(path) + row = writer.execute("SELECT rowid FROM buckets WHERE id='b'").fetchone()[0] + writer.execute( + "INSERT INTO events(bucketrow,starttime,endtime,datastr) VALUES (?, ?, ?, ?)", + (row, 1704110400000000, 1704110401000000, "{}"), + ) + writer.commit() + writer.close() + changed.append(True) + yield from original(scoped, bid, **kwargs) + + from unittest.mock import patch + + with patch.object(storage.__class__, "iter_events", events_with_concurrent_write): + with api.stage_device_export() as staged: + exported = json.load(staged) + assert exported["devices"]["server-uuid"]["buckets"]["b"]["events"] == [] + assert api.for_device("local").get_eventcount("b") == 1 From 2d4a3f207b1bbcc5e45bb08ba4df8940d5d05f2b Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 12:38:12 +0000 Subject: [PATCH 05/11] fix(server): use thread-owned SQLite request connections Git-Session-Id: 931478ca-2d2b-48c5-bb99-0bb5ad608478 --- README.md | 8 ++++++ aw_server/rest.py | 8 +++--- aw_server/server.py | 38 ++++++++++++++++++++++++++++ tests/test_device_addressing.py | 44 +++++++++++++++++++++++++++++++++ 4 files changed, 94 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index a8fe815..55cb2f7 100644 --- a/README.md +++ b/README.md @@ -102,3 +102,11 @@ Run `make test` for the normal fast server slice plus the device HTTP contract. The pinned draft core is required; tests use temporary file-backed SQLite and never a production database. Full shared HTTP parity and backend-specific migration/same-name acceptance still gate removal of Rust's global name guards. + +SQLite HTTP requests on worker threads use their own datastore/connection, +closed at request teardown; they share API write serialization, query cache and +pair-keyed heartbeat state, never a mutable scope. This preserves SQLite's +thread ownership check. Current export staging holds the API write lock while +building the snapshot, so large exports delay writes. A dedicated read-only +export connection and unbounded timestamp/archive fidelity remain acceptance +work; event streaming retains the backend's existing time-range semantics. diff --git a/aw_server/rest.py b/aw_server/rest.py index 5f03d4f..6dfeee9 100644 --- a/aw_server/rest.py +++ b/aw_server/rest.py @@ -183,7 +183,7 @@ def decorator(f): def request_api() -> ServerAPI: - return getattr(g, "device_api", current_app.api) + return getattr(g, "device_api", getattr(g, "request_api", current_app.api)) class DeviceResource(Resource): @@ -227,14 +227,14 @@ def get(self): @api.route("/0/devices/buckets") class DeviceBucketsResource(Resource): def get(self): - return current_app.api.device_buckets() + return request_api().device_buckets() @api.route("/0/devices/export") class DevicesExportResource(Resource): def get(self): try: - staged = current_app.api.stage_device_export() + staged = request_api().stage_device_export() except NotImplementedError as e: return {"message": str(e)}, 501 try: @@ -256,7 +256,7 @@ def get(self): class DevicesImportResource(Resource): def post(self): try: - current_app.api.import_devices(request.get_json()) + request_api().import_devices(request.get_json()) except NotImplementedError as e: return {"message": str(e)}, 501 except ImportRollbackError as e: diff --git a/aw_server/server.py b/aw_server/server.py index f0f9e86..d62049f 100644 --- a/aw_server/server.py +++ b/aw_server/server.py @@ -1,5 +1,7 @@ +import copy import logging import os +import threading from datetime import datetime, timedelta from typing import Dict, List, Optional @@ -7,10 +9,12 @@ import flask.json.provider from aw_datastore import Datastore from aw_datastore.storages.peewee import PeeweeStorage +from aw_datastore.storages.sqlite import SqliteStorage from flask import ( Blueprint, Flask, current_app, + g, send_from_directory, ) from flask_cors import CORS @@ -80,6 +84,40 @@ def close_database(error): database.close() self.api = ServerAPI(db=db, testing=testing, query_cache=query_cache) + if isinstance(db.storage_strategy, SqliteStorage): + owner_thread = threading.get_ident() + filepath = db.storage_strategy.conn.execute( + "PRAGMA database_list" + ).fetchone()[2] + + @self.before_request + def sqlite_request_connection(): + # SQLite handles are thread-owned. Never disable that safety or + # swap the shared API's db while another request is running. + if threading.get_ident() == owner_thread: + return + request_db = Datastore( + SqliteStorage, + testing=testing, + filepath=filepath, + enable_lazy_commit=False, + ) + request_api = copy.copy(self.api) + request_api.db = request_db + g.request_api = request_api + + @self.teardown_request + def close_sqlite_request(error): + request_api = g.pop("request_api", None) + if request_api is not None: + storage = request_api.db.storage_strategy + try: + if error is None: + storage.commit() + else: + storage.conn.rollback() + finally: + storage.conn.close() self.register_blueprint(root) self.register_blueprint(rest.blueprint) diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index f6127af..b002341 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -432,3 +432,47 @@ def events_with_concurrent_write(scoped, bid, **kwargs): exported = json.load(staged) assert exported["devices"]["server-uuid"]["buckets"]["b"]["events"] == [] assert api.for_device("local").get_eventcount("b") == 1 + + +def test_sqlite_device_requests_use_request_thread_connections(device_app): + from concurrent.futures import ThreadPoolExecutor + + def request(): + c = device_app.test_client() + r = create(c, "local") + assert r.status_code == 200 + r = c.post("/api/0/devices/local/buckets/window/events", json=event("thread")) + assert r.status_code == 200 + r = c.get("/api/0/devices/export") + assert r.status_code == 200 + data = r.json + r.close() + return data + + with ThreadPoolExecutor(max_workers=1) as pool: + exported = pool.submit(request).result(timeout=10) + assert exported["devices"]["server-uuid"]["buckets"]["window"]["events"][0][ + "data" + ] == {"label": "thread"} + assert ( + device_app.test_client() + .get("/api/0/devices/local/buckets/window/events/count") + .json + == 1 + ) + + +def test_sqlite_parallel_devices_do_not_share_request_scope(device_app): + from concurrent.futures import ThreadPoolExecutor + + def write(device): + c = device_app.test_client() + assert create(c, device).status_code == 200 + base = f"/api/0/devices/{device}/buckets/window" + assert c.post(base + "/events", json=event(device)).status_code == 200 + return c.get(base + "/events").json[0]["data"] + + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(write, ("local", "peer"))) + assert results == [{"label": "local"}, {"label": "peer"}] + assert device_app.api._device_scope is None From d47d930a8a3ccdc8b4bb389b555f6d654b01da20 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 12:57:51 +0000 Subject: [PATCH 06/11] feat(api): merge device imports and export without blocking writes Device import now deduplicates existing same-device events, matching Rust, and merges aliases into the stored row instead of rejecting. Grouped export pins a read-only SQLite snapshot so staging no longer holds the API write lock. Pins aw-core#183 at 4ac2146. Git-Session-Id: aeaa4ffa-df33-4fe0-8487-f35c72996602 --- README.md | 17 +++--- aw_server/api.py | 94 ++++++++++++++++++++++++--------- poetry.lock | 6 +-- pyproject.toml | 2 +- tests/test_device_addressing.py | 60 +++++++++++++++++++-- 5 files changed, 137 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 55cb2f7..947e078 100644 --- a/README.md +++ b/README.md @@ -87,9 +87,12 @@ latest-hostname registry. `GET /api/0/devices/buckets` groups metadata explicitl `POST /api/0/devices/import` accepts that envelope, rejects empty or `local` source identities, and validates enclosing ownership, bucket keys and event payloads before writes. Local exported ownership becomes the source UUID, so -history stays peer-owned at another server. Python's existing reject-on-existing -import policy remains (unlike Rust's deduplicating import). Individual groups -use legacy rollback; a runtime failure can leave earlier successful groups. +history stays peer-owned at another server. Existing same-device buckets merge +events and skip duplicates (timestamp, duration, data), matching Rust; an +alias of stored history merges into that row instead of creating a sibling. +Legacy `/import` still rejects existing exact names. Individual groups use +legacy rollback for buckets this call created; a runtime failure can leave +earlier successful groups. Metadata `data` is preserved. Grouped export streams SQLite rows from one read snapshot into a private temporary file before responding; failures cannot send partial JSON with a success status. The response owns the staged file, not a @@ -106,7 +109,7 @@ migration/same-name acceptance still gate removal of Rust's global name guards. SQLite HTTP requests on worker threads use their own datastore/connection, closed at request teardown; they share API write serialization, query cache and pair-keyed heartbeat state, never a mutable scope. This preserves SQLite's -thread ownership check. Current export staging holds the API write lock while -building the snapshot, so large exports delay writes. A dedicated read-only -export connection and unbounded timestamp/archive fidelity remain acceptance -work; event streaming retains the backend's existing time-range semantics. +thread ownership check. Grouped export flushes under the write lock, then +stages from a dedicated read-only snapshot connection so later API writes are +not blocked. Event streaming retains the backend's existing time-range +semantics; unbounded archive/timestamp fidelity remains acceptance work. diff --git a/aw_server/api.py b/aw_server/api.py index b67a587..9f8ed67 100644 --- a/aw_server/api.py +++ b/aw_server/api.py @@ -169,27 +169,32 @@ def device_buckets(self) -> Dict[str, Any]: target[bid] = bucket return {"devices": groups} - @_serialized def stage_device_export(self): """Stream one SQLite snapshot into a private file before HTTP success. The response owns the returned file. No database cursor or transaction survives into response iteration, which may run on a different thread. + Staging uses a dedicated read-only connection so API writes are not + blocked for the duration of the snapshot. """ storage = self.db.storage_strategy - if not hasattr(storage, "device_buckets"): + if not hasattr(storage, "snapshot"): raise NotImplementedError("Snapshot device export requires SQLite") - conn = storage.conn - # Flush lazy writes before beginning an independent read snapshot. - storage.commit() staged = tempfile.TemporaryFile(mode="w+b") + snap = None def write(value): staged.write(value.encode("utf-8")) try: - conn.execute("BEGIN") - groups = self.device_buckets()["devices"] + with self._write_lock: + snap = storage.snapshot() + snap_db = copy.copy(self.db) + snap_db.storage_strategy = snap + snap_db.bucket_instances = {} + snap_api = copy.copy(self) + snap_api.db = snap_db + groups = snap_api.device_buckets()["devices"] write('{"devices":{') first_device = True for device, group in groups.items(): @@ -198,7 +203,7 @@ def write(value): first_device = False write(json.dumps(device) + ':{"buckets":{') first_bucket = True - scoped = self.for_device(device) + scoped = snap_api.for_device(device) for bid, metadata in group["buckets"].items(): if not first_bucket: write(",") @@ -224,21 +229,55 @@ def write(value): write("]}") write("}}") write("}}") - conn.commit() + snap.conn.commit() staged.seek(0) return staged except BaseException: - conn.rollback() + if snap is not None: + snap.conn.rollback() staged.close() raise + finally: + if snap is not None: + snap.conn.close() + + def _import_event_identity(self, event: Event): + # Match Rust's import identity: start, duration, canonical data. + return ( + event.timestamp, + event.duration, + json.dumps(event.data, sort_keys=True), + ) + + def _merge_imported_events(self, bucket_id: str, events: List[Event]) -> None: + incoming = [Event(**e) if isinstance(e, dict) else e for e in events] + for event in incoming: + if "id" in event: + del event["id"] + if not incoming: + return + start = min(event.timestamp for event in incoming) + end = max(event.timestamp + event.duration for event in incoming) + existing = { + self._import_event_identity(event) + for event in self.db[bucket_id].get(-1, start, end) + } + new_events = [ + event + for event in incoming + if self._import_event_identity(event) not in existing + ] + if new_events: + self.create_events(bucket_id, new_events) @_serialized def import_devices(self, envelope: Any) -> None: """Preflight all identities and event payloads before any writes. - Existing buckets are rejected, matching Python's legacy import policy. - Each group uses its existing rollback contract; earlier successful groups - may remain on a runtime failure. This is not an atomic multi-device import. + Existing same-device buckets merge events and skip duplicates, matching + Rust device import. Each group uses its existing rollback contract for + buckets created by this call; earlier successful groups may remain on a + runtime failure. This is not an atomic multi-device import. """ if not isinstance(envelope, dict) or not isinstance( envelope.get("devices"), dict @@ -283,13 +322,9 @@ def import_devices(self, envelope: Any) -> None: except jsonschema.ValidationError as e: raise ValueError(f"Invalid event: {e.message}") from e Event(**event) - if scoped.db.has_bucket(bid): - raise ValueError( - f"Bucket '{bid}' already exists in device '{device}'" - ) prepared.append((scoped, buckets)) for scoped, buckets in prepared: - scoped.import_all(buckets) + scoped.import_all(buckets, merge=True) def get_buckets(self) -> Dict[str, Dict]: """Get dict {bucket_name: Bucket} of all buckets""" @@ -341,10 +376,15 @@ def export_all(self) -> Dict[str, Any]: return exported_buckets @_serialized - def import_bucket(self, bucket_data: Any): + def import_bucket(self, bucket_data: Any, merge: bool = False): bucket_id = bucket_data["id"] logger.info(f"Importing bucket {bucket_id}") + if merge and self.db.has_bucket(bucket_id): + canonical = self.db[bucket_id].metadata()["id"] + self._merge_imported_events(canonical, bucket_data.get("events") or []) + return + if bucket_id in self.db.buckets(): raise ValueError( f"Bucket '{bucket_id}' already exists. Delete it first or rename the bucket before importing." @@ -378,7 +418,7 @@ def import_bucket(self, bucket_data: Any): ) @_serialized - def import_all(self, buckets: Dict[str, Any]): + def import_all(self, buckets: Dict[str, Any], merge: bool = False): if not isinstance(buckets, dict): raise ValueError( "'buckets' must be an object mapping bucket IDs to buckets" @@ -389,27 +429,29 @@ def import_all(self, buckets: Dict[str, Any]): # import_all holds the write lock shared by bucket and event writes. bucket_ids = [bucket["id"] for bucket in buckets.values()] seen: Set[str] = set() + existed_before = {} for bucket_id in bucket_ids: if bucket_id in seen: raise ValueError( f"Bucket '{bucket_id}' appears more than once in the import." ) seen.add(bucket_id) - if bucket_id in self.db.buckets(): + existed_before[bucket_id] = self.db.has_bucket(bucket_id) + if not merge and bucket_id in self.db.buckets(): raise ValueError( f"Bucket '{bucket_id}' already exists. Delete it first or rename the bucket before importing." ) try: for bucket in buckets.values(): - self.import_bucket(bucket) + self.import_bucket(bucket, merge=merge) except Exception as import_error: - # None of these buckets existed before (checked above, under the - # write lock), so any that exist now were created by this import, - # including one whose events failed after the bucket was created. + # Delete only buckets this call created. A merge into pre-existing + # history, including an alias of a stored name, must not roll back + # that row. A failed create can still leave an empty bucket. existing = self.db.buckets() failed_rollbacks = [] for bucket_id in seen: - if bucket_id not in existing: + if existed_before[bucket_id] or bucket_id not in existing: continue try: self.delete_bucket(bucket_id) diff --git a/poetry.lock b/poetry.lock index 1540f62..56fb556 100644 --- a/poetry.lock +++ b/poetry.lock @@ -80,8 +80,8 @@ tomlkit = "*" [package.source] type = "git" url = "https://github.com/TimeToBuildBob/aw-core.git" -reference = "4026a50e2ad191c43348c80ff5c34b010af5b2d6" -resolved_reference = "4026a50e2ad191c43348c80ff5c34b010af5b2d6" +reference = "4ac214618dd9e93cc8e054b4e797ad70df96f641" +resolved_reference = "4ac214618dd9e93cc8e054b4e797ad70df96f641" [[package]] name = "black" @@ -1212,4 +1212,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = "^3.8" -content-hash = "c0fed743f561fe81a5da1d1f8499b368551a87f4e3c1fa87865d2a9b0cd6ee03" +content-hash = "1171ee065f23094fdcf1cf0f0c7af8010b4775bf500b31679873a64103bf5298" diff --git a/pyproject.toml b/pyproject.toml index a77f30f..370b404 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ aw-server = "aw_server:main" [tool.poetry.dependencies] python = "^3.8" # Draft dependency: ActivityWatch/aw-core#183 (stacked on schema #177). -aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4026a50e2ad191c43348c80ff5c34b010af5b2d6"} +aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4ac214618dd9e93cc8e054b4e797ad70df96f641"} aw-client = "^0.5.8" flask = "^2.2" flask-restx = "^1.0.3" diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index b002341..182c7d7 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -255,11 +255,23 @@ def test_import_preserves_metadata_and_rejects_alias_split(device_app): envelope = {"devices": {"source": {"buckets": {"history_host": bucket}}}} assert c.post("/api/0/devices/import", json=envelope).status_code == 200 assert c.get("/api/0/devices/source/buckets/history").json["data"] == bucket["data"] - # A later import alias cannot create an empty canonical row over history. - bucket["id"] = "history" - envelope["devices"]["source"]["buckets"] = {"history": bucket} - assert c.post("/api/0/devices/import", json=envelope).status_code == 400 - assert c.get("/api/0/devices/source/buckets/history/events/count").json == 1 + # Re-import is idempotent. An alias of the stored name must merge into the + # existing row, never mint an empty canonical sibling. + assert c.post("/api/0/devices/import", json=envelope).status_code == 200 + alias = dict(bucket) + alias["id"] = "history" + alias["events"] = [event("source"), event("extra", 1)] + envelope["devices"]["source"]["buckets"] = {"history": alias} + assert c.post("/api/0/devices/import", json=envelope).status_code == 200 + listed = c.get("/api/0/devices/source/buckets/").json + assert set(listed) == {"history_host"} + assert "history" not in listed + assert c.get("/api/0/devices/source/buckets/history/events/count").json == 2 + labels = { + e["data"]["label"] + for e in c.get("/api/0/devices/source/buckets/history/events").json + } + assert labels == {"source", "extra"} def test_scoped_heartbeat_invalidates_after_insert_and_delete(device_app): @@ -434,6 +446,44 @@ def events_with_concurrent_write(scoped, bid, **kwargs): assert api.for_device("local").get_eventcount("b") == 1 +def test_grouped_export_does_not_block_writes(device_app): + import threading + from concurrent.futures import ThreadPoolExecutor + from unittest.mock import patch + + from aw_datastore.storages.sqlite import SqliteStorage + + c = device_app.test_client() + assert create(c, "local").status_code == 200 + entered = threading.Event() + release = threading.Event() + original = SqliteStorage.iter_events + + def paused(self, bucket_id, **kwargs): + entered.set() + assert release.wait(5) + yield from original(self, bucket_id, **kwargs) + + with patch.object(SqliteStorage, "iter_events", paused): + with ThreadPoolExecutor(max_workers=2) as pool: + export = pool.submit( + lambda: device_app.test_client().get("/api/0/devices/export") + ) + assert entered.wait(5) + write = pool.submit( + lambda: create(device_app.test_client(), "local", "during") + ) + assert write.result(timeout=5).status_code == 200 + release.set() + exported = export.result(timeout=5) + assert exported.status_code == 200 + assert "during" not in exported.json["devices"]["server-uuid"]["buckets"] + assert ( + device_app.test_client().get("/api/0/devices/local/buckets/during").status_code + == 200 + ) + + def test_sqlite_device_requests_use_request_thread_connections(device_app): from concurrent.futures import ThreadPoolExecutor From a0cdb62a347e7c8f2c6921624482a70689618dcf Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 13:55:51 +0000 Subject: [PATCH 07/11] fix(export): retain exact archive events through device transfer Git-Session-Id: 0dc232f4-6721-4652-9e55-bd51fe8eec69 --- poetry.lock | 6 ++-- pyproject.toml | 4 +-- tests/test_device_addressing.py | 50 +++++++++++++++++++++++++++++++++ tests/test_server.py | 4 +-- 4 files changed, 57 insertions(+), 7 deletions(-) diff --git a/poetry.lock b/poetry.lock index 56fb556..b7990f6 100644 --- a/poetry.lock +++ b/poetry.lock @@ -80,8 +80,8 @@ tomlkit = "*" [package.source] type = "git" url = "https://github.com/TimeToBuildBob/aw-core.git" -reference = "4ac214618dd9e93cc8e054b4e797ad70df96f641" -resolved_reference = "4ac214618dd9e93cc8e054b4e797ad70df96f641" +reference = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a" +resolved_reference = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a" [[package]] name = "black" @@ -1212,4 +1212,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = "^3.8" -content-hash = "1171ee065f23094fdcf1cf0f0c7af8010b4775bf500b31679873a64103bf5298" +content-hash = "f36f29a27a111d2f9c22963ce8a33587f6bd53b34849924fb03f9553bb7ef808" diff --git a/pyproject.toml b/pyproject.toml index 370b404..e931801 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,8 +11,8 @@ aw-server = "aw_server:main" [tool.poetry.dependencies] python = "^3.8" -# Draft dependency: ActivityWatch/aw-core#183 (stacked on schema #177). -aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4ac214618dd9e93cc8e054b4e797ad70df96f641"} +# Draft dependency: ActivityWatch/aw-core#184 (archive fidelity on #183/#177). +aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a"} aw-client = "^0.5.8" flask = "^2.2" flask-restx = "^1.0.3" diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index 182c7d7..9e9bda8 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -526,3 +526,53 @@ def write(device): results = list(pool.map(write, ("local", "peer"))) assert results == [{"label": "local"}, {"label": "peer"}] assert device_app.api._device_scope is None + + +@pytest.mark.parametrize( + "timestamp", + ["1960-01-01T00:00:00.123457+00:00", "2300-01-01T00:00:00.123457+00:00"], +) +def test_archive_export_reimport_retains_exact_event_and_owner(device_app, timestamp): + import csv + import io + + client = device_app.test_client() + assert create(client, "peer").status_code == 200 + base = "/api/0/devices/peer/buckets/window" + payload = { + "timestamp": timestamp, + "duration": 0.000003, + "data": {"text": "archive, \u00e5\nsecond line", "nested": {"x": [1, None]}}, + } + inserted = client.post(base + "/events", json=payload) + assert inserted.status_code == 200 + assert inserted.json[0]["timestamp"] == timestamp + assert inserted.json[0]["duration"] == payload["duration"] + for route in (base + "/export", "/api/0/devices/export"): + response = client.get(route) + assert response.status_code == 200 + if route == base + "/export": + exported = response.json["buckets"]["window"] + else: + envelope = response.json + exported = envelope["devices"]["peer"]["buckets"]["window"] + assert exported["events"] == [payload] + assert exported["device_id"] == "peer" + csv_response = client.get(base + "/export/csv") + assert csv_response.status_code == 200 + rows = list(csv.DictReader(io.StringIO(csv_response.text))) + assert len(rows) == 1 + assert rows[0]["timestamp"] == timestamp + assert rows[0]["duration"] == "0.000003000" + assert client.delete(base).status_code == 200 + response = client.post("/api/0/devices/import", json=envelope) + assert response.status_code == 200 + events = client.get(base + "/events").json + assert [{k: v for k, v in item.items() if k != "id"} for item in events] == [ + payload + ] + assert client.get(base + "/events/count").json == 1 + assert client.get(base).json["device_id"] == "peer" + # A repeated archive does not duplicate the newly restored event. + assert client.post("/api/0/devices/import", json=envelope).status_code == 200 + assert client.get(base + "/events/count").json == 1 diff --git a/tests/test_server.py b/tests/test_server.py index 6995160..c468b0a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -232,10 +232,10 @@ def test_export_bucket_csv(flask_client, bucket): # Newest first, union of data keys, formula-looking values neutralized assert lines[0] == "id,timestamp,duration,app,count,flag,title" assert lines[1].endswith( - ",2026-10-06T12:00:02.123000+00:00,0.000000000,'=cmd,3,null," + ",2026-10-06T12:00:02.123456+00:00,0.000000000,'=cmd,3,null," ) assert lines[2].endswith( - ',2026-10-06T12:00:00.123000+00:00,1.500000000,Firefox,,,"a, ""quoted"" title"' + ',2026-10-06T12:00:00.123456+00:00,1.500000000,Firefox,,,"a, ""quoted"" title"' ) assert lines[3] == "" From ffcd3f906910891bac1d52bdb391f2c0fc6e7f55 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 13:26:34 +0000 Subject: [PATCH 08/11] fix(api): skip duplicate new events within a device import batch Existing-bucket merge compared each incoming event only against stored history, so two identical new events in one envelope both reached create_events. Record accepted identities as they land, and execute DELETE requests before asserting their status in the HTTP tests. Git-Session-Id: 72dcece2-03b4-59da-979e-db600720e05b --- aw_server/api.py | 11 ++++++----- tests/test_device_addressing.py | 31 +++++++++++++++++++++++++++++-- 2 files changed, 35 insertions(+), 7 deletions(-) diff --git a/aw_server/api.py b/aw_server/api.py index 9f8ed67..6d062f8 100644 --- a/aw_server/api.py +++ b/aw_server/api.py @@ -262,11 +262,12 @@ def _merge_imported_events(self, bucket_id: str, events: List[Event]) -> None: self._import_event_identity(event) for event in self.db[bucket_id].get(-1, start, end) } - new_events = [ - event - for event in incoming - if self._import_event_identity(event) not in existing - ] + new_events = [] + for event in incoming: + identity = self._import_event_identity(event) + if identity not in existing: + existing.add(identity) + new_events.append(event) if new_events: self.create_events(bucket_id, new_events) diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index 9e9bda8..7cf2e68 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -58,7 +58,8 @@ def test_scoped_crud_uuid_alias_and_legacy_conflict(device_app): ] == {"label": "local"} for path in ("/buckets/", "/buckets/window", "/buckets/window/events", "/export"): assert c.get("/api/0" + path).status_code == 409 - assert c.delete("/api/0/devices/local/buckets/window").status_code == 200 + deleted = c.delete("/api/0/devices/local/buckets/window") + assert deleted.status_code == 200 assert c.get("/api/0/devices/peer/buckets/window/events/count").json == 1 @@ -274,6 +275,31 @@ def test_import_preserves_metadata_and_rejects_alias_split(device_app): assert labels == {"source", "extra"} +def test_import_skips_duplicate_new_events_in_one_batch(device_app): + c = device_app.test_client() + bucket = { + "id": "history", + "device_id": "source", + "client": "c", + "type": "t", + "hostname": "host", + "created": "2024-01-01T00:00:00Z", + "events": [event("source")], + } + envelope = {"devices": {"source": {"buckets": {"history": bucket}}}} + assert c.post("/api/0/devices/import", json=envelope).status_code == 200 + # Existing-bucket merge must skip identical new events inside one batch, + # not only events already stored from a previous import. + bucket["events"] = [event("dup", 1), event("dup", 1)] + assert c.post("/api/0/devices/import", json=envelope).status_code == 200 + assert c.get("/api/0/devices/source/buckets/history/events/count").json == 2 + labels = [ + e["data"]["label"] + for e in c.get("/api/0/devices/source/buckets/history/events").json + ] + assert sorted(labels) == ["dup", "source"] + + def test_scoped_heartbeat_invalidates_after_insert_and_delete(device_app): c = device_app.test_client() assert create(c, "local").status_code == 200 @@ -284,7 +310,8 @@ def test_scoped_heartbeat_invalidates_after_insert_and_delete(device_app): c.post(base + "/heartbeat?pulsetime=10", json=event("a", 4)).status_code == 200 ) assert c.get(base + "/events/count").json == 3 - assert c.delete(base + f"/events/{inserted}").status_code == 200 + deleted = c.delete(base + f"/events/{inserted}") + assert deleted.status_code == 200 assert c.get(base + "/events/count").json == 2 From d97c1b28134f3bf80d8dd95387d180aeaa88e0f9 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 14:03:41 +0000 Subject: [PATCH 09/11] fix(deps): include WAL-safe archive migration discovery Git-Session-Id: 0dc232f4-6721-4652-9e55-bd51fe8eec69 --- poetry.lock | 6 +++--- pyproject.toml | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/poetry.lock b/poetry.lock index b7990f6..e034e5a 100644 --- a/poetry.lock +++ b/poetry.lock @@ -80,8 +80,8 @@ tomlkit = "*" [package.source] type = "git" url = "https://github.com/TimeToBuildBob/aw-core.git" -reference = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a" -resolved_reference = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a" +reference = "538886d4063ed8df683dfcb0759e69c827824a77" +resolved_reference = "538886d4063ed8df683dfcb0759e69c827824a77" [[package]] name = "black" @@ -1212,4 +1212,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = "^3.8" -content-hash = "f36f29a27a111d2f9c22963ce8a33587f6bd53b34849924fb03f9553bb7ef808" +content-hash = "07d5b6fcfa455de65b6b4e364044e8caccc910f6cf1e908086930fe9059402cc" diff --git a/pyproject.toml b/pyproject.toml index e931801..b0440a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ aw-server = "aw_server:main" [tool.poetry.dependencies] python = "^3.8" # Draft dependency: ActivityWatch/aw-core#184 (archive fidelity on #183/#177). -aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "4a04a57bd0eaf5d2e53ac2a87f231234c389bd3a"} +aw-core = {git = "https://github.com/TimeToBuildBob/aw-core.git", rev = "538886d4063ed8df683dfcb0759e69c827824a77"} aw-client = "^0.5.8" flask = "^2.2" flask-restx = "^1.0.3" From d2381cfe9111c08daaff6f9eb3da1b01c83e4017 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 14:08:24 +0000 Subject: [PATCH 10/11] test(devices): retain three same-name histories across HTTP reload Git-Session-Id: 0dc232f4-6721-4652-9e55-bd51fe8eec69 --- tests/test_device_addressing.py | 55 +++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index 7cf2e68..e13f303 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -603,3 +603,58 @@ def test_archive_export_reimport_retains_exact_event_and_owner(device_app, times # A repeated archive does not duplicate the newly restored event. assert client.post("/api/0/devices/import", json=envelope).status_code == 200 assert client.get(base + "/events/count").json == 1 + + +def test_three_same_named_devices_survive_http_reload_and_scoped_deletion( + device_app, monkeypatch +): + client = device_app.test_client() + devices = ("local", "peer-one", "peer-two") + for device in devices: + assert create(client, device).status_code == 200 + base = f"/api/0/devices/{device}/buckets/window" + assert client.post(base + "/events", json=event(device)).status_code == 200 + device_app.api.db.storage_strategy.commit() + path = device_app.api.db.storage_strategy.conn.execute( + "PRAGMA database_list" + ).fetchone()[2] + reopened = AWFlask( + "127.0.0.1", + testing=True, + cors_origins=[], + storage_method=partial(SqliteStorage, filepath=path), + ) + try: + other = reopened.test_client() + for device in devices: + base = f"/api/0/devices/{device}/buckets/window" + assert other.get(base + "/events/count").json == 1 + assert other.get(base + "/events").json[0]["data"] == {"label": device} + result = other.post( + f"/api/0/devices/{device}/query", + json={ + "query": ['RETURN = query_bucket("window");'], + "timeperiods": ["2024-01-01T00:00:00Z/2024-01-02T00:00:00Z"], + }, + ) + assert result.status_code == 200 + assert result.json[0][0]["data"] == {"label": device} + assert other.get("/api/0/export").status_code == 409 + assert other.get("/api/0/buckets/").status_code == 409 + exported = other.get("/api/0/devices/export") + assert exported.status_code == 200 + assert set(exported.json["devices"]) == {"server-uuid", "peer-one", "peer-two"} + assert other.delete("/api/0/devices/peer-one/buckets/window").status_code == 200 + assert other.get("/api/0/devices/peer-one/buckets/window").status_code == 404 + assert client.get("/api/0/devices/peer-one/buckets/window").status_code == 404 + for device in ("local", "peer-two"): + assert ( + other.get(f"/api/0/devices/{device}/buckets/window/events/count").json + == 1 + ) + assert ( + client.get(f"/api/0/devices/{device}/buckets/window/events/count").json + == 1 + ) + finally: + reopened.api.db.storage_strategy.conn.close() From 5b12d1b109ca92749883bd761bc80fe4b5c5172f Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 23 Oct 2026 14:14:22 +0000 Subject: [PATCH 11/11] test(export): execute scoped deletes outside assertions Git-Session-Id: 0dc232f4-6721-4652-9e55-bd51fe8eec69 --- tests/test_device_addressing.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_device_addressing.py b/tests/test_device_addressing.py index e13f303..e87eeed 100644 --- a/tests/test_device_addressing.py +++ b/tests/test_device_addressing.py @@ -591,7 +591,8 @@ def test_archive_export_reimport_retains_exact_event_and_owner(device_app, times assert len(rows) == 1 assert rows[0]["timestamp"] == timestamp assert rows[0]["duration"] == "0.000003000" - assert client.delete(base).status_code == 200 + deleted = client.delete(base) + assert deleted.status_code == 200 response = client.post("/api/0/devices/import", json=envelope) assert response.status_code == 200 events = client.get(base + "/events").json @@ -644,7 +645,8 @@ def test_three_same_named_devices_survive_http_reload_and_scoped_deletion( exported = other.get("/api/0/devices/export") assert exported.status_code == 200 assert set(exported.json["devices"]) == {"server-uuid", "peer-one", "peer-two"} - assert other.delete("/api/0/devices/peer-one/buckets/window").status_code == 200 + deleted = other.delete("/api/0/devices/peer-one/buckets/window") + assert deleted.status_code == 200 assert other.get("/api/0/devices/peer-one/buckets/window").status_code == 404 assert client.get("/api/0/devices/peer-one/buckets/window").status_code == 404 for device in ("local", "peer-two"):