"""Dependency-free reference client for Signal Trade API integrations.""" from __future__ import annotations import json import re import time import uuid from datetime import datetime, timezone from email.utils import parsedate_to_datetime from http.client import HTTPException from typing import Any, Callable, Iterator from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode, urljoin, urlsplit, urlunsplit from urllib.request import HTTPRedirectHandler, Request, build_opener DEFAULT_BASE_URL = "https://signalautomotive.uk" PAGE_SIZE = 100 REQUEST_TIMEOUT_SECONDS = 20 MAX_REQUEST_ATTEMPTS = 4 MAX_RETRY_DELAY_SECONDS = 10 RETRYABLE_STATUSES = {429, 502, 503, 504} SUPPORTED_API_SCOPES = { "stock:read", "stock:read_write", "stock:manage", "stock:finance_read", "stock:finance_manage", "stock_review:read", "stock_review:manage", "advert:manage", "appraisal:read", "appraisal:manage", "appraisal:workflow", "appraisal:workflow_evidence", "prep:read", "prep:manage", "crm:read", "crm:manage", "deal:read", "deal:workflow", } NATIVE_SYNC_MAX_IDS = 100_000 NATIVE_SYNC_MAX_EVENTS = 1_000_000 DEAL_UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.I) def _deal_id(value: str) -> str: if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", value): raise ValueError("A saved deal or snapshot ID is required") return quote(value, safe="") def _deal_envelope(value: dict[str, Any], deal_id: str | None = None) -> dict[str, Any]: deal = value.get("deal") if (value.get("version") != "1" or not isinstance(deal, dict) or not isinstance(deal.get("id"), str) or not deal["id"] or deal_id is not None and deal["id"] != deal_id or type(deal.get("version")) is not int or not 1 <= deal["version"] <= 9007199254740991 or deal.get("kind") not in {"purchase", "sale"} or deal.get("phase") not in {"draft", "agreed", "reserved", "completed", "cancelled"} or not isinstance(value.get("source"), dict) or not isinstance(value.get("paymentEvents"), list) or not isinstance(value.get("paymentSummary"), dict) or not isinstance(value.get("snapshots"), list)): raise RuntimeError("Invalid deal detail; no local checkpoint was advanced") return value def get_deal_record(api_key: str, deal_id: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: return _deal_envelope(_request_json(base_url, api_key, f"/api/v1/deals/{_deal_id(deal_id)}"), deal_id) def _deal_write(api_key: str, body: dict[str, Any], mode: str, deal_id: str | None = None, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: if not isinstance(body, dict) or not DEAL_UUID.fullmatch(str(body.get("operationId", ""))): raise ValueError("Persist a UUID v4 operationId and the exact body before sending") payload = json.loads(json.dumps(body, allow_nan=False)) if mode != "create" and (type(payload.get("expectedVersion")) is not int or not 1 <= payload["expectedVersion"] < 9007199254740991): raise ValueError("The current expectedVersion is required") if mode == "create" and payload.get("kind") not in {"purchase", "sale"}: raise ValueError("Choose purchase or sale") path = "/api/v1/deals" if mode == "create" else f"/api/v1/deals/{_deal_id(deal_id)}" + ("/payments" if mode == "payment" else "") value = _deal_envelope(_request_json(base_url, api_key, path, method="PATCH" if mode == "change" else "POST", payload=payload, ambiguous_post_message="Deal write outcome is uncertain. Retry the persisted identical body and operationId; do not generate a replacement ID."), deal_id) receipt = value.get("operation") action = "create" if mode == "create" else "payment" if mode == "payment" else payload.get("action") if (not isinstance(receipt, dict) or receipt.get("id") != payload["operationId"] or receipt.get("action") != action or receipt.get("dealId") != value["deal"]["id"] or type(receipt.get("committedVersion")) is not int or not 1 <= receipt["committedVersion"] <= value["deal"]["version"] or mode != "create" and receipt["committedVersion"] != payload["expectedVersion"] + 1 or "snapshotId" not in receipt or not (mode == "create" and receipt["snapshotId"] is None or isinstance(receipt["snapshotId"], str) and receipt["snapshotId"]) or not _valid_iso_datetime(receipt.get("committedAt")) or type(value.get("replayed")) is not bool): raise RuntimeError("Deal receipt could not be verified. Keep and retry the identical saved request.") return value def create_deal_record(api_key: str, body: dict[str, Any], base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: return _deal_write(api_key, body, "create", base_url=base_url) def change_deal_record(api_key: str, deal_id: str, body: dict[str, Any], base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: return _deal_write(api_key, body, "change", deal_id, base_url) def record_deal_payment(api_key: str, deal_id: str, body: dict[str, Any], base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: return _deal_write(api_key, body, "payment", deal_id, base_url) def load_all_deals(api_key: str, on_deal: Callable[[dict[str, Any]], None], base_url: str = DEFAULT_BASE_URL, filters: dict[str, str] | None = None) -> dict[str, Any]: if not callable(on_deal) or filters is not None and (not isinstance(filters, dict) or any(key not in {"kind", "phase", "vehicle_id", "appraisal_id", "ids", "updated_since"} for key in filters)): raise ValueError("Use a callback and documented record filters; inventory cursors and limits belong to the helper") query = dict(filters or {}) cursor, checkpoint, count = None, None, 0 seen: set[str] = set() seen_ids: set[str] = set() while True: query["limit"] = str(PAGE_SIZE) if cursor: query["cursor"] = cursor page = _request_json(base_url, api_key, _query_path("/api/v1/deals", query)) rows = page.get("data") if (page.get("version") != "1" or not isinstance(rows, list) or not isinstance(page.get("syncCheckpoint"), str) or not page["syncCheckpoint"] or any(not isinstance(row, dict) or not isinstance(row.get("id"), str) or not row["id"] or type(row.get("version")) is not int or row["version"] < 1 for row in rows)): raise RuntimeError("Invalid deal list; initial sync cannot be confirmed") checkpoint = checkpoint or page["syncCheckpoint"] if page["syncCheckpoint"] != checkpoint: raise RuntimeError("Deal inventory checkpoint changed between pages") cursor = _next_page_cursor(page.get("pagination"), seen) for row in rows: if row["id"] in seen_ids: raise RuntimeError("Deal inventory repeated a record; reconcile before saving its checkpoint") seen_ids.add(row["id"]) on_deal(row) count += 1 if cursor is None: return {"syncCheckpoint": checkpoint, "recordsUpdated": count} def get_deal_snapshot(api_key: str, deal_id: str, snapshot_id: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: page = _request_json(base_url, api_key, f"/api/v1/deals/{_deal_id(deal_id)}/snapshots/{_deal_id(snapshot_id)}") snapshot = page.get("snapshot") if (page.get("version") != "1" or not isinstance(snapshot, dict) or snapshot.get("id") != snapshot_id or snapshot.get("dealId") != deal_id or type(snapshot.get("version")) is not int or snapshot["version"] < 1 or type(page.get("dealVersion")) is not int or not snapshot["version"] <= page["dealVersion"] <= 9007199254740991 or snapshot.get("kind") not in {"purchase", "sale"} or snapshot.get("phase") not in {"draft", "agreed", "reserved", "completed", "cancelled"} or not isinstance(page.get("snapshotHash"), str) or not re.fullmatch(r"[0-9a-f]{64}", page["snapshotHash"])): raise RuntimeError("Invalid saved deal revision") return snapshot def sync_deal_records(api_key: str, checkpoint: str, on_record: Callable[[dict[str, Any], dict[str, Any]], None], on_delete: Callable[[str, dict[str, Any]], None], save_checkpoint: Callable[[str], None], base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: """Callbacks durably and idempotently apply records/removals before returning.""" if not isinstance(checkpoint, str) or not checkpoint or not all(map(callable, [on_record, on_delete, save_checkpoint])): raise ValueError("The completed inventory checkpoint and all three durable callbacks are required") cursor, last_sequence, count = None, 0, 0 seen: set[str] = set() while True: query = {"limit": str(PAGE_SIZE), "checkpoint": checkpoint} if cursor: query["cursor"] = cursor page = _request_json(base_url, api_key, _query_path("/api/v1/deals/events", query)) events = page.get("data") if page.get("version") != "1" or not isinstance(events, list): raise RuntimeError("Invalid deal event page; checkpoint was not advanced") next_cursor = _next_page_cursor(page.get("pagination"), seen) if ((next_cursor is not None and page.get("nextCheckpoint") is not None) or (next_cursor is None and (not isinstance(page.get("nextCheckpoint"), str) or not page["nextCheckpoint"]))): raise RuntimeError("Invalid deal event checkpoint; checkpoint was not advanced") for event in events: if (not isinstance(event, dict) or not isinstance(event.get("id"), str) or not event["id"] or not isinstance(event.get("dealId"), str) or not event["dealId"] or event.get("action") not in {"upsert", "delete"} or event.get("kind") not in {"purchase", "sale"} or type(event.get("dealVersion")) is not int or not 1 <= event["dealVersion"] <= 9007199254740991 or type(event.get("sequence")) is not int or not last_sequence < event["sequence"] <= 9007199254740991 or not _valid_iso_datetime(event.get("recordedAt"))): raise RuntimeError("Invalid deal event ordering; checkpoint was not advanced") last_sequence = event["sequence"] latest = {event["dealId"]: event for event in events} for event in sorted(latest.values(), key=lambda item: item["sequence"]): if event["action"] == "delete": on_delete(event["dealId"], event) else: on_record(get_deal_record(api_key, event["dealId"], base_url), event) count += 1 if next_cursor is None: save_checkpoint(page["nextCheckpoint"]) return {"checkpoint": page["nextCheckpoint"], "recordsUpdated": count} cursor = next_cursor SITE_ACTIONS = { "trade_site_created", "trade_site_renamed", "trade_site_archived", "trade_site_restored", } STOCK_EVENT_TYPES = { "stock_vehicle_created": "stock_vehicle", "stock_vehicle_updated": "stock_vehicle", "stock_advert_created": "stock_vehicle", "stock_advert_updated": "stock_vehicle", "stock_advert_published": "stock_vehicle", "stock_advert_paused": "stock_vehicle", "stock_advert_photo_added": "stock_vehicle", "stock_advert_photos_reordered": "stock_vehicle", "stock_advert_photo_removed": "stock_vehicle", "stock_vehicle_added": "vehicle", "stock_api_created": "vehicle", "stock_api_updated": "vehicle", "stock_csv_imported": "vehicle", "vehicle_record_updated": "vehicle", "vehicle_sale_recorded": "vehicle", "stock_record_deleted": "vehicle", "trade_site_created": "business", "trade_site_renamed": "business", "trade_site_archived": "business", "trade_site_restored": "business", } def _origin(url: str) -> tuple[str, str, int | None]: parsed = urlsplit(url) default_port = 443 if parsed.scheme.lower() == "https" else 80 return parsed.scheme.lower(), (parsed.hostname or "").lower(), parsed.port or default_port def _valid_iso_datetime(value: object) -> bool: if not isinstance(value, str) or not re.fullmatch( r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", value, ): return False try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.tzinfo is not None except (TypeError, ValueError, OverflowError): return False def _iso_time(value: str) -> datetime: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) class _SameOriginRedirectHandler(HTTPRedirectHandler): """Do not forward the bearer key to a different origin after a redirect.""" def redirect_request(self, request, file_pointer, code, message, headers, new_url): destination = urljoin(request.full_url, new_url) if _origin(request.full_url) != _origin(destination): raise HTTPError( request.full_url, code, "Cross-origin redirect refused to protect the Signal Trade API key", headers, file_pointer, ) return super().redirect_request(request, file_pointer, code, message, headers, destination) _opener = build_opener(_SameOriginRedirectHandler()) def _retry_delay(retry_after: str | None, attempt: int) -> float: if retry_after: try: seconds = float(retry_after) except ValueError: try: retry_at = parsedate_to_datetime(retry_after) if retry_at.tzinfo is None: retry_at = retry_at.replace(tzinfo=timezone.utc) seconds = (retry_at - datetime.now(timezone.utc)).total_seconds() except (TypeError, ValueError, OverflowError): seconds = -1 if seconds >= 0: return min(seconds, MAX_RETRY_DELAY_SECONDS) return min(0.25 * (2 ** (attempt - 1)), MAX_RETRY_DELAY_SECONDS) def _request_json( base_url: str, api_key: str, path: str, *, method: str = "GET", payload: dict[str, Any] | None = None, ambiguous_post_message: str | None = None, before_attempt: Callable[[], None] | None = None, ) -> dict[str, Any]: parsed_base = urlsplit(base_url) if not parsed_base.hostname or (parsed_base.scheme != "https" and not (parsed_base.scheme == "http" and parsed_base.hostname in {"localhost", "127.0.0.1", "::1"})): raise ValueError("Signal Trade API requests require HTTPS") if parsed_base.username or parsed_base.password: raise ValueError("base_url must not contain embedded credentials") if not api_key: raise ValueError("api_key is required") if method not in {"GET", "POST", "PATCH", "DELETE"}: raise ValueError("method must be GET, POST, PATCH or DELETE") if method == "GET" and payload is not None: raise ValueError("GET requests cannot include a payload") url = urljoin(urlunsplit((parsed_base.scheme, parsed_base.netloc, "/", "", "")), path) data = None if payload is None else json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") request = Request(url, data=data, method=method, headers={ "Authorization": f"Bearer {api_key}", "Accept": "application/json", **({"Content-Type": "application/json"} if data is not None else {}), }) last_error: Exception | None = None for attempt in range(1, MAX_REQUEST_ATTEMPTS + 1): if before_attempt is not None: before_attempt() try: with _opener.open(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: status = response.status headers = response.headers raw_body = response.read() if 200 <= status < 300: body = json.loads(raw_body.decode("utf-8")) if not isinstance(body, dict): raise ValueError("Invalid JSON object response") except HTTPError as error: status = error.code retry_after = error.headers.get("Retry-After") if error.headers else None error.close() if method != "GET" or status not in RETRYABLE_STATUSES or attempt == MAX_REQUEST_ATTEMPTS: raise RuntimeError(f"Signal Trade API returned HTTP {status}") from error time.sleep(_retry_delay(retry_after, attempt)) continue except (URLError, TimeoutError, OSError, HTTPException, ValueError) as error: if method != "GET": if ambiguous_post_message: raise RuntimeError(ambiguous_post_message) from error raise RuntimeError( "Signal Trade API write outcome could not be confirmed. Read the record before retrying." ) from error last_error = error if attempt == MAX_REQUEST_ATTEMPTS: break time.sleep(_retry_delay(None, attempt)) continue if method == "GET" and status in RETRYABLE_STATUSES and attempt < MAX_REQUEST_ATTEMPTS: time.sleep(_retry_delay(headers.get("Retry-After"), attempt)) continue if status < 200 or status >= 300: raise RuntimeError(f"Signal Trade API returned HTTP {status}") try: body = json.loads(raw_body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise RuntimeError("Signal Trade API returned invalid JSON") from error if not isinstance(body, dict): raise RuntimeError("Signal Trade API returned an unexpected response") return body reason = "request timed out or failed" if last_error is None else str(last_error) raise RuntimeError(f"Signal Trade API request failed after {MAX_REQUEST_ATTEMPTS} attempts: {reason}") from last_error def _query_path(path: str, values: dict[str, str]) -> str: return f"{path}?{urlencode(values)}" def _next_page_cursor(pagination: object, seen_cursors: set[str]) -> str | None: if not isinstance(pagination, dict) or "nextCursor" not in pagination: raise RuntimeError("Signal Trade returned missing or invalid pagination.nextCursor; sync cannot continue safely") next_cursor = pagination["nextCursor"] if next_cursor is not None: if not isinstance(next_cursor, str) or not next_cursor: raise RuntimeError("Signal Trade returned invalid pagination.nextCursor; sync cannot continue safely") if next_cursor in seen_cursors: raise RuntimeError("Signal Trade repeated a pagination cursor; sync cannot continue safely") seen_cursors.add(next_cursor) return next_cursor def _pages( base_url: str, api_key: str, path: str, *, page_size_parameter: str = "limit", initial_query: dict[str, str] | None = None, ) -> Iterator[dict[str, Any]]: cursor: str | None = None seen_cursors: set[str] = set() while True: query = dict(initial_query or {}) query[page_size_parameter] = str(PAGE_SIZE) if cursor: query["cursor"] = cursor page = _request_json(base_url, api_key, _query_path(path, query)) if path == "/api/v1/stock" and page.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") if not isinstance(page.get("data"), list) or any( not isinstance(record, dict) or not isinstance(record.get("id"), str) or not record["id"] for record in page["data"] ) or not isinstance(page.get("pagination"), dict): raise RuntimeError("Signal Trade returned an invalid record page") next_cursor = _next_page_cursor(page["pagination"], seen_cursors) yield page cursor = next_cursor if not cursor: return def get_stock_event_checkpoint( api_key: str, base_url: str = DEFAULT_BASE_URL, ) -> str: """Capture this timestamp before the initial stock load.""" page = _request_json(base_url, api_key, "/api/v1/stock/events?limit=1") if page.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") checkpoint = page.get("pollStartedAt") if not _valid_iso_datetime(checkpoint): raise RuntimeError("Signal Trade did not return pollStartedAt") return checkpoint def get_trade_api_account( api_key: str, base_url: str = DEFAULT_BASE_URL, *, transport: Callable[[str], dict[str, Any]] | None = None, ) -> dict[str, Any]: """Verify the dealership and scope attached to an API key, without loading records.""" response = transport("/api/v1/account") if transport is not None else _request_json(base_url, api_key, "/api/v1/account") business = response.get("business") key = response.get("apiKey") if ( response.get("version") != "1" or not isinstance(business, dict) or not isinstance(business.get("id"), str) or not business["id"] or not isinstance(business.get("name"), str) or not business["name"] or not isinstance(key, dict) or not isinstance(key.get("id"), str) or not key["id"] or not isinstance(key.get("name"), str) or not isinstance(key.get("tokenPrefix"), str) or not re.fullmatch(r"sig_live_[A-Za-z0-9_-]{8}", key["tokenPrefix"]) or key["tokenPrefix"] != api_key[:17] or not isinstance(key.get("scope"), str) or key["scope"] not in SUPPORTED_API_SCOPES or not _valid_iso_datetime(response.get("generatedAt")) ): raise RuntimeError("Signal Trade returned an invalid API account response") return response def get_stock_sites( api_key: str, base_url: str = DEFAULT_BASE_URL, ) -> list[dict[str, Any]]: """Read the active branch list without loading the full stock inventory.""" response = _request_json(base_url, api_key, "/api/v1/stock?limit=1") if response.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") sites = response.get("sites") if not isinstance(sites, list) or any( not isinstance(site, dict) or not isinstance(site.get("id"), str) or not site["id"] for site in sites ): raise RuntimeError("Signal Trade returned an invalid branch list") return sites def load_all_stock( api_key: str, on_vehicle: Callable[[dict[str, Any]], None], on_sites: Callable[[list[dict[str, Any]]], None] | None = None, base_url: str = DEFAULT_BASE_URL, ) -> int: """Load every current stock page; callers should upsert by Signal record ID.""" if not callable(on_vehicle): raise TypeError("on_vehicle is required") loaded = 0 seen_record_ids: set[str] = set() for page in _pages(base_url, api_key, "/api/v1/stock"): page_ids = [vehicle["id"] for vehicle in page["data"]] if len(set(page_ids)) != len(page_ids) or seen_record_ids.intersection(page_ids): raise RuntimeError("Signal Trade returned duplicate stock IDs during the full load") seen_record_ids.update(page_ids) sites = page.get("sites") if on_sites: if not isinstance(sites, list) or any( not isinstance(site, dict) or not isinstance(site.get("id"), str) or not site["id"] for site in sites ) or len({site["id"] for site in sites}) != len(sites): raise RuntimeError("Signal Trade returned an invalid branch list") on_sites(sites) for vehicle in page["data"]: if not isinstance(vehicle, dict) or not isinstance(vehicle.get("id"), str): raise RuntimeError("Signal Trade returned an invalid or unexpected stock record") on_vehicle(vehicle) loaded += 1 return loaded def get_stock_vehicle(api_key: str, vehicle_id: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: """Read the current record before an update or sale; retain its updatedAt.""" if not isinstance(vehicle_id, str) or not vehicle_id or len(vehicle_id) > 80: raise ValueError("vehicle_id must be a valid Signal stock ID") response = _request_json(base_url, api_key, f"/api/v1/stock/{quote(vehicle_id, safe='')}") if response.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") vehicle = response.get("vehicle") if not isinstance(vehicle, dict) or vehicle.get("id") != vehicle_id or not isinstance(vehicle.get("updatedAt"), str): raise RuntimeError("Signal Trade returned an invalid stock record") return vehicle def create_stock_vehicle( api_key: str, create_id: str | None, vehicle: dict[str, Any], base_url: str = DEFAULT_BASE_URL, *, operation_id: str | None = None, ) -> dict[str, Any]: """Create stock with a caller-persisted UUID v4 for safe retry after an uncertain result.""" if (create_id is None) == (operation_id is None): raise ValueError("Supply exactly one saved create_id or operation_id") if create_id is not None and (not isinstance(create_id, str) or not UUID_V4_PATTERN.fullmatch(create_id)): raise ValueError("create_id must be a stable UUID v4 saved by the calling integration") if not isinstance(vehicle, dict): raise TypeError("vehicle must be a stock create dictionary") if {"id", "createId", "operationId", "saleId", "expectedUpdatedAt"}.intersection(vehicle): raise ValueError("Supply retry identifiers separately from vehicle fields") response = _request_json( base_url, api_key, "/api/v1/stock", method="POST", payload={**vehicle, **({} if create_id is None else {"createId": create_id}), **_operation_field(operation_id)}, ambiguous_post_message=( "Signal Trade could not confirm stock creation. Retry the same vehicle with the same saved " "create_id, or reconcile that create_id before changing the request." ), ) result = response.get("vehicle") if not isinstance(result, dict) or not isinstance(result.get("id"), str) or not result["id"]: raise RuntimeError("Signal Trade returned no created stock vehicle") if not isinstance(response.get("created"), bool): raise RuntimeError("Signal Trade returned an invalid stock creation response") if operation_id is not None: validate_stock_write_receipt(response, operation_id, "stock_create", response["vehicle"]["id"]) return response def create_stock_vehicles( api_key: str, vehicles: list[dict[str, Any]], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Create up to 25 stock records; persist a unique UUID v4 create_id for each before sending.""" if not isinstance(vehicles, list) or not 1 <= len(vehicles) <= 25: raise ValueError("vehicles must contain between 1 and 25 stock records") seen_create_ids: set[str] = set() records: list[dict[str, Any]] = [] for index, item in enumerate(vehicles): if not isinstance(item, dict): raise TypeError(f"vehicles[{index}] must be a dictionary") create_id = item.get("create_id") operation_id = item.get("operation_id") vehicle = item.get("vehicle") if (create_id is None) == (operation_id is None): raise ValueError(f"vehicles[{index}] needs exactly one saved retry identifier") retry_id = create_id if create_id is not None else operation_id if not isinstance(retry_id, str) or not UUID_V4_PATTERN.fullmatch(retry_id): raise ValueError(f"vehicles[{index}] retry identifier must be a saved UUID v4") normalized_create_id = retry_id.lower() if normalized_create_id in seen_create_ids: raise ValueError("Each vehicle in one batch must have a different create_id") seen_create_ids.add(normalized_create_id) if not isinstance(vehicle, dict): raise TypeError(f"vehicles[{index}].vehicle must be a stock create dictionary") if {"id", "createId", "operationId", "saleId", "expectedUpdatedAt"}.intersection(vehicle): raise ValueError(f"vehicles[{index}].vehicle must not include retry identifiers; supply them separately") records.append({**vehicle, **({} if create_id is None else {"createId": create_id}), **_operation_field(operation_id)}) response = _request_json( base_url, api_key, "/api/v1/stock/bulk", method="POST", payload={"records": records}, ambiguous_post_message=( "Signal Trade could not confirm the bulk creation. Retry the identical batch with the same saved " "create_id for each vehicle, or reconcile those IDs first." ), ) results = response.get("results") counts = (response.get("created"), response.get("replayed"), response.get("rejected"), response.get("notProcessed")) if ( response.get("version") != "1" or response.get("requested") != len(records) or not isinstance(results, list) or len(results) != len(records) or any(not isinstance(count, int) or isinstance(count, bool) for count in counts) or sum(counts) != len(records) ): raise RuntimeError("Signal Trade returned an invalid bulk stock creation response") for index, row in enumerate(results): if ( not isinstance(row, dict) or row.get("index") != index or row.get("status") not in {"created", "replayed", "rejected", "not_processed"} or not (isinstance(row.get("httpStatus"), int) or row.get("httpStatus") is None) or ( row.get("status") in {"created", "replayed"} and (not isinstance(row.get("vehicle"), dict) or not row["vehicle"].get("id")) ) or (row.get("status") == "replayed" and row.get("replayed") is not True) ): raise RuntimeError("Signal Trade returned an invalid bulk stock creation result row") if vehicles[index].get("operation_id") is not None and row["status"] in {"created", "replayed"}: validate_stock_write_receipt(row, vehicles[index]["operation_id"], "stock_create", row["vehicle"]["id"]) return response def update_stock_vehicle( api_key: str, vehicle_id: str, changes: dict[str, Any], expected_updated_at: str, base_url: str = DEFAULT_BASE_URL, *, operation_id: str | None = None, ) -> dict[str, Any]: """Update stock; after 409 reload the record and reconcile before resending.""" if not isinstance(vehicle_id, str) or not vehicle_id or len(vehicle_id) > 80: raise ValueError("vehicle_id must be a valid Signal stock ID") if not isinstance(changes, dict) or not changes: raise ValueError("changes must contain at least one stock field") if {"id", "createId", "operationId", "saleId", "expectedUpdatedAt"}.intersection(changes): raise ValueError("Supply retry identifiers and expected_updated_at separately from changes") if changes.get("status") == "sold": raise ValueError("Record a completed sale with record_stock_sale and a stable sale_id") if not isinstance(expected_updated_at, str): raise TypeError("expected_updated_at must be the exact updatedAt value read from Signal") try: parsed = datetime.fromisoformat(expected_updated_at.replace("Z", "+00:00")) except ValueError as error: raise ValueError("expected_updated_at must be an ISO date-time") from error if parsed.tzinfo is None: raise ValueError("expected_updated_at must include a timezone") canonical = parsed.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") if expected_updated_at != canonical: raise ValueError("expected_updated_at must exactly match the canonical updatedAt value read from Signal") payload = {**changes, "expectedUpdatedAt": expected_updated_at, **_operation_field(operation_id)} response = _request_json( base_url, api_key, f"/api/v1/stock/{quote(vehicle_id, safe='')}", method="PATCH", payload=payload, ) if response.get("id") != vehicle_id or response.get("updated") is not True: raise RuntimeError("Signal Trade returned an invalid stock update response") if operation_id is not None: validate_stock_write_receipt(response, operation_id, "stock_update", vehicle_id) return response def update_stock_vehicles( api_key: str, updates: list[dict[str, Any]], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Apply up to 25 version-checked updates in one request; reconcile rejected rows individually.""" if not isinstance(updates, list) or not 1 <= len(updates) <= 25: raise ValueError("updates must contain between 1 and 25 stock changes") records: list[dict[str, Any]] = [] for index, item in enumerate(updates): if not isinstance(item, dict): raise TypeError(f"updates[{index}] must be a dictionary") vehicle_id = item.get("vehicle_id") expected_updated_at = item.get("expected_updated_at") changes = item.get("changes") if not isinstance(vehicle_id, str) or not vehicle_id or len(vehicle_id) > 80: raise ValueError(f"updates[{index}].vehicle_id must be a valid Signal stock ID") if not isinstance(changes, dict) or not changes: raise ValueError(f"updates[{index}].changes must contain at least one stock field") if {"id", "expectedUpdatedAt", "saleId", "createId", "operationId"}.intersection(changes): raise ValueError(f"updates[{index}].changes must not include id, expectedUpdatedAt or saleId") if changes.get("status") == "sold": raise ValueError(f"updates[{index}] is a sale; record it separately with record_stock_sale") if not isinstance(expected_updated_at, str): raise TypeError(f"updates[{index}].expected_updated_at must be the exact updatedAt value read from Signal") try: parsed = datetime.fromisoformat(expected_updated_at.replace("Z", "+00:00")) except ValueError as error: raise ValueError(f"updates[{index}].expected_updated_at must be an ISO date-time") from error if parsed.tzinfo is None: raise ValueError(f"updates[{index}].expected_updated_at must include a timezone") canonical = parsed.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") if expected_updated_at != canonical: raise ValueError(f"updates[{index}].expected_updated_at must exactly match the canonical updatedAt value read from Signal") records.append({**changes, "id": vehicle_id, "expectedUpdatedAt": expected_updated_at, **_operation_field(item.get("operation_id"))}) response = _request_json( base_url, api_key, "/api/v1/stock/bulk", method="PATCH", payload={"records": records}, ambiguous_post_message=( "Signal Trade could not confirm the bulk update. Read every affected record before retrying individual reconciled changes." ), ) results = response.get("results") counts = (response.get("updated"), response.get("rejected"), response.get("notProcessed")) if ( response.get("version") != "1" or response.get("requested") != len(records) or not isinstance(results, list) or len(results) != len(records) or any(not isinstance(count, int) or isinstance(count, bool) for count in counts) or sum(counts) != len(records) ): raise RuntimeError("Signal Trade returned an invalid bulk stock update response") for index, row in enumerate(results): if ( not isinstance(row, dict) or row.get("index") != index or row.get("status") not in {"updated", "rejected", "not_processed"} or not (isinstance(row.get("httpStatus"), int) or row.get("httpStatus") is None) or (row.get("status") == "updated" and row.get("id") != updates[index]["vehicle_id"]) ): raise RuntimeError("Signal Trade returned an invalid bulk stock update result row") if updates[index].get("operation_id") is not None and row["status"] == "updated": validate_stock_write_receipt(row, updates[index]["operation_id"], "stock_update", updates[index]["vehicle_id"]) return response def _operation_field(operation_id: str | None) -> dict[str, str]: if operation_id is None: return {} if not isinstance(operation_id, str) or not UUID_V4_PATTERN.fullmatch(operation_id): raise ValueError("operation_id must be a saved UUID v4") return {"operationId": operation_id.lower()} def validate_stock_write_receipt(response: dict[str, Any], operation_id: str, action: str, vehicle_id: str) -> dict[str, Any]: """Confirm the original save independently of the freshly authorized current record.""" _operation_field(operation_id) receipt = response.get("operation") result = receipt.get("result") if isinstance(receipt, dict) else None allowed = {"vehicleId", "recordedAt", "vehicleUpdatedAt", "advertId", "advertVersion", "photoId", "photoIds", "changedFields"} advert_action = action.startswith("advert_") or action in {"photos_reorder", "photo_remove"} if (not isinstance(receipt, dict) or receipt.get("id") != operation_id.lower() or receipt.get("action") != action or not isinstance(response.get("replayed"), bool) or not isinstance(result, dict) or result.get("vehicleId") != vehicle_id or not _valid_iso_datetime(result.get("recordedAt")) or not _valid_stock_receipt_stamp(result.get("vehicleUpdatedAt")) or not set(result).issubset(allowed) or advert_action and (not isinstance(result.get("advertId"), str) or not result["advertId"] or type(result.get("advertVersion")) is not int or result["advertVersion"] < 1) or action == "photos_reorder" and (not isinstance(result.get("photoIds"), list) or any(not isinstance(item, str) or not item for item in result["photoIds"]) or len(set(result["photoIds"])) != len(result["photoIds"])) or action == "photo_remove" and (not isinstance(result.get("photoId"), str) or not result["photoId"])): raise RuntimeError("Signal Trade did not return the original operation confirmation. Retain the saved request for reconciliation.") return receipt def _valid_stock_receipt_stamp(value: object) -> bool: return _valid_iso_datetime(value) or (isinstance(value, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d+)?", value) is not None and _valid_iso_datetime(value.replace(" ", "T") + "Z")) def _advert_path(vehicle_id: str) -> str: if not isinstance(vehicle_id, str) or not vehicle_id or len(vehicle_id) > 80: raise ValueError("vehicle_id must be a valid Signal stock ID") return f"/api/v1/stock/{quote(vehicle_id, safe='')}/advert" def get_stock_advert(api_key: str, vehicle_id: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any] | None: response = _request_json(base_url, api_key, _advert_path(vehicle_id)) advert = response.get("advert") if response.get("version") != "1" or "advert" not in response or (advert is not None and (not isinstance(advert, dict) or advert.get("vehicleId") != vehicle_id or type(advert.get("version")) is not int)): raise RuntimeError("Signal Trade returned an invalid advert record") return advert def _advert_write(api_key: str, vehicle_id: str, operation_id: str, action: str, method: str, payload: dict[str, Any], base_url: str, suffix: str = "") -> dict[str, Any]: if operation_id is None: raise ValueError("A saved operation_id is required for confirmed advert writes") response = _request_json(base_url, api_key, _advert_path(vehicle_id) + suffix, method=method, payload={**payload, **_operation_field(operation_id)}, ambiguous_post_message="Advert write could not be confirmed. Retry the identical saved body and operation_id, including its original version.") advert = response.get("advert") if response.get("version") != "1" or "advert" not in response or advert is not None and (not isinstance(advert, dict) or advert.get("vehicleId") != vehicle_id or type(advert.get("version")) is not int): raise RuntimeError("Signal Trade returned an invalid current advert") receipt = validate_stock_write_receipt(response, operation_id, action, vehicle_id) if ("version" in payload and receipt["result"]["advertVersion"] != payload["version"] + 1 or action == "photos_reorder" and receipt["result"]["photoIds"] != payload["photoIds"] or action == "photo_remove" and quote(receipt["result"]["photoId"], safe="") != suffix.rsplit("/", 1)[-1]): raise RuntimeError("Signal Trade returned a confirmation for different advert or photo changes") return response def create_stock_advert(api_key: str, vehicle_id: str, operation_id: str, base_url: str = DEFAULT_BASE_URL, *, expected_vehicle_updated_at: str | None = None) -> dict[str, Any]: payload = {} if expected_vehicle_updated_at is None else {"expectedVehicleUpdatedAt": expected_vehicle_updated_at} return _advert_write(api_key, vehicle_id, operation_id, "advert_create", "POST", payload, base_url) def change_stock_advert(api_key: str, vehicle_id: str, operation_id: str, action: str, version: int, base_url: str = DEFAULT_BASE_URL, *, fields: dict[str, Any] | None = None, confirmed: bool | None = None, expected_vehicle_updated_at: str | None = None) -> dict[str, Any]: if action not in {"save", "publish", "pause"} or type(version) is not int or version < 1: raise ValueError("A valid advert action and version are required") if fields is not None and (not isinstance(fields, dict) or {"operationId", "action", "version"}.intersection(fields)): raise ValueError("Advert fields must not replace the operation identity or version") payload = {**(fields or {}), "action": action, "version": version} if confirmed is not None: payload["confirmed"] = confirmed if expected_vehicle_updated_at is not None: payload["expectedVehicleUpdatedAt"] = expected_vehicle_updated_at return _advert_write(api_key, vehicle_id, operation_id, "advert_" + action, "PATCH", payload, base_url) def reorder_stock_advert_photos(api_key: str, vehicle_id: str, operation_id: str, version: int, photo_ids: list[str], base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: if type(version) is not int or version < 1 or not isinstance(photo_ids, list) or not 1 <= len(photo_ids) <= 8 or any(not isinstance(item, str) or not item for item in photo_ids) or len(set(photo_ids)) != len(photo_ids): raise ValueError("Include a version and every photo ID exactly once") return _advert_write(api_key, vehicle_id, operation_id, "photos_reorder", "PATCH", {"version": version, "photoIds": photo_ids}, base_url, "/photos") def remove_stock_advert_photo(api_key: str, vehicle_id: str, operation_id: str, version: int, photo_id: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: if type(version) is not int or version < 1 or not isinstance(photo_id, str) or not photo_id: raise ValueError("A photo ID and current advert version are required") return _advert_write(api_key, vehicle_id, operation_id, "photo_remove", "DELETE", {"version": version}, base_url, "/photos/" + quote(photo_id, safe="")) def upload_stock_advert_photo(api_key: str, vehicle_id: str, upload_id: str, data: bytes, media_type: str, base_url: str = DEFAULT_BASE_URL) -> dict[str, Any]: """Persist upload_id and exact original bytes before sending; never blindly retry an upload.""" if not api_key or not isinstance(upload_id, str) or not UUID_V4_PATTERN.fullmatch(upload_id): raise ValueError("api_key and a saved upload_id UUID v4 are required") if not isinstance(data, bytes) or not 1 <= len(data) <= 4 * 1024 * 1024 or media_type not in {"image/jpeg", "image/png", "image/webp"}: raise ValueError("Supply up to 4 MB of JPG, PNG or WebP bytes with the matching media type") base = urlsplit(base_url) if base.username or base.password or not base.hostname or (base.scheme != "https" and not (base.scheme == "http" and base.hostname in {"localhost", "127.0.0.1", "::1"})): raise ValueError("base_url requires HTTPS except on loopback, without embedded credentials") boundary = "signal" + uuid.uuid4().hex raw = (f'--{boundary}\r\nContent-Disposition: form-data; name="uploadId"\r\n\r\n{upload_id.lower()}\r\n' f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="vehicle-photo"\r\nContent-Type: {media_type}\r\n\r\n').encode() + data + f'\r\n--{boundary}--\r\n'.encode() request = Request(urljoin(base_url, _advert_path(vehicle_id) + "/photos"), data=raw, method="POST", headers={"Authorization": "Bearer " + api_key, "Accept": "application/json", "Content-Type": "multipart/form-data; boundary=" + boundary}) try: with _opener.open(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: result = json.loads(response.read().decode("utf-8")) except HTTPError as error: code = error.code error.close() raise RuntimeError(f"Signal Trade API returned HTTP {code}") from error except (URLError, TimeoutError, OSError, HTTPException, ValueError) as error: raise RuntimeError("Photo upload could not be confirmed. Retry the same saved upload_id and original bytes.") from error upload = result.get("upload") if isinstance(result, dict) else None current = result.get("advert") if isinstance(result, dict) else None if (not isinstance(result, dict) or result.get("version") != "1" or not isinstance(upload, dict) or upload.get("id") != upload_id.lower() or upload.get("photoId") != upload_id.lower() or upload.get("mediaType") != media_type or upload.get("size") != len(data) or not _valid_iso_datetime(upload.get("createdAt")) or type(upload.get("advertVersion")) is not int or upload["advertVersion"] < 1 or type(upload.get("position")) is not int or upload["position"] < 0 or not isinstance(result.get("replayed"), bool) or not isinstance(result.get("created"), bool) or not isinstance(result.get("removed"), bool) or not isinstance(current, dict) or current.get("vehicleId") != vehicle_id or not isinstance(current.get("id"), str) or not current["id"] or type(current.get("version")) is not int or current["version"] < upload["advertVersion"] or not isinstance(current.get("photos"), list) or any(not isinstance(photo, dict) or not isinstance(photo.get("id"), str) or not photo["id"] or type(photo.get("position")) is not int or photo["position"] < 0 for photo in current["photos"]) or len({photo["id"] for photo in current["photos"]}) != len(current["photos"]) or result["removed"] == any(photo["id"] == upload["photoId"] for photo in current["photos"])): raise RuntimeError("Signal Trade returned an invalid original photo upload confirmation") return result def record_stock_sale( api_key: str, vehicle_id: str, sale_id: str, sale: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Record a sale using a caller-persisted UUID v4; retry only identical details with the same ID.""" if not isinstance(vehicle_id, str) or not vehicle_id or len(vehicle_id) > 80: raise ValueError("vehicle_id must be a valid Signal stock ID") if not isinstance(sale_id, str) or not UUID_V4_PATTERN.fullmatch(sale_id): raise ValueError("sale_id must be a stable UUID v4 saved by the calling integration") if not isinstance(sale, dict) or not {"expectedUpdatedAt", "soldPricePence", "soldDate", "soldPriceVatBasis"}.issubset(sale) or not isinstance(sale.get("expectedUpdatedAt"), str): raise ValueError("sale must include expectedUpdatedAt from the current record, soldPricePence, soldDate and soldPriceVatBasis") response = _request_json( base_url, api_key, f"/api/v1/stock/{quote(vehicle_id, safe='')}", method="PATCH", payload={**sale, "status": "sold", "saleId": sale_id}, ambiguous_post_message=( "Signal Trade could not confirm the sale. Retry the identical sale details with the same saved sale_id." ), ) if response.get("id") != vehicle_id or response.get("sold") is not True or response.get("saleId") != sale_id: raise RuntimeError("Signal Trade returned an invalid stock sale response") return response def _load_records( api_key: str, path: str, on_record: Callable[[dict[str, Any]], None], base_url: str, *, page_size_parameter: str = "limit", query: dict[str, str] | None = None, ) -> int: if not callable(on_record): raise TypeError("on_record is required") loaded = 0 for page in _pages( base_url, api_key, path, page_size_parameter=page_size_parameter, initial_query=query, ): for record in page["data"]: on_record(record) loaded += 1 return loaded def load_all_appraisals( api_key: str, on_appraisal: Callable[[dict[str, Any]], None], base_url: str = DEFAULT_BASE_URL, ) -> int: """Load saved appraisals using an appraisal-scoped API key.""" return _load_records(api_key, "/api/v1/appraisals", on_appraisal, base_url) def calculate_trade_valuation( api_key: str, request_body: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Calculate a Signal trade estimate without saving an appraisal record.""" if not isinstance(request_body, dict) or not request_body: raise TypeError("request_body must be a non-empty valuation input dictionary") result = _request_json( base_url, api_key, "/api/v1/valuation", method="POST", payload=request_body, ambiguous_post_message=( "Signal Trade could not confirm the valuation response. " "This request does not save an appraisal; you can repeat the same calculation inputs." ), ) valuation = result.get("valuation") if ( result.get("version") != "1" or not isinstance(result.get("cached"), bool) or not _valid_iso_datetime(result.get("generatedAt")) or not isinstance(valuation, dict) or not isinstance(valuation.get("id"), str) or not isinstance(valuation.get("vehicle"), dict) or not isinstance(valuation.get("prices"), dict) or not isinstance(valuation.get("buyingPlan"), dict) or not isinstance(valuation.get("evidence"), dict) or not isinstance(valuation.get("interpretation"), str) ): raise RuntimeError("Signal Trade returned an invalid valuation response") return result def load_all_preparation( api_key: str, on_job: Callable[[dict[str, Any]], None], base_url: str = DEFAULT_BASE_URL, ) -> int: """Load preparation jobs using a preparation-scoped API key.""" return _load_records(api_key, "/api/v1/preparation", on_job, base_url) def load_all_crm( api_key: str, kind: str, on_record: Callable[[dict[str, Any]], None], base_url: str = DEFAULT_BASE_URL, ) -> int: """Load one CRM record kind using a CRM-scoped key.""" kinds = {"customers", "opportunities", "tasks", "activities"} if kind not in kinds: raise ValueError("kind must be customers, opportunities, tasks or activities") return _load_records( api_key, "/api/v1/crm", on_record, base_url, page_size_parameter="page_size", query={"kind": kind, "filter": "all"}, ) CRM_KINDS = {"customers", "opportunities", "tasks", "activities"} CRM_MUTABLE_KINDS = {"customers", "opportunities", "tasks"} UUID_V4_PATTERN = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.IGNORECASE) def get_crm_assignees(api_key: str, base_url: str = DEFAULT_BASE_URL) -> list[dict[str, Any]]: """List current team members available for CRM task assignment; emails are omitted.""" response = _request_json(base_url, api_key, "/api/v1/crm/assignees") members = response.get("data") if not isinstance(members, list) or any( not isinstance(member, dict) or not isinstance(member.get("userId"), str) or not member["userId"] or not isinstance(member.get("displayName"), str) or member.get("role") not in ("principal", "manager", "sales_buyer") for member in members ): raise RuntimeError("Signal Trade returned an invalid CRM assignee directory") return members def get_crm_record( api_key: str, kind: str, record_id: str, base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Fetch one current CRM record by kind and ID for safe reconciliation.""" if not isinstance(kind, str) or kind not in CRM_KINDS: raise ValueError("kind must be customers, opportunities, tasks or activities") if not isinstance(record_id, str) or not record_id or len(record_id) > 80: raise ValueError("record_id must be a valid CRM record ID") response = _request_json( base_url, api_key, f"/api/v1/crm/{quote(record_id, safe='')}?{urlencode({'kind': kind})}", ) result = response.get("record") if response.get("kind") != kind or not isinstance(result, dict) or result.get("id") != record_id: raise RuntimeError("Signal Trade returned an unexpected CRM record") return result def create_crm_record( api_key: str, kind: str, create_id: str, record: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Create a CRM record. Retry only identical details with the same UUID v4 create_id.""" if not isinstance(kind, str) or kind not in CRM_KINDS: raise ValueError("kind must be customers, opportunities, tasks or activities") if not isinstance(create_id, str) or not UUID_V4_PATTERN.fullmatch(create_id): raise ValueError("create_id must be a stable UUID v4") if not isinstance(record, dict): raise TypeError("record must be a dictionary") response = _request_json( base_url, api_key, "/api/v1/crm", method="POST", payload={**record, "entity": kind, "createId": create_id}, ) result = response.get("record") if not isinstance(result, dict) or not isinstance(result.get("id"), str) or not result["id"]: raise RuntimeError("Signal Trade returned no created CRM record") return result def update_crm_record( api_key: str, kind: str, record_id: str, version: int, changes: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Update a mutable CRM record using its current positive version; writes are never retried.""" if not isinstance(kind, str) or kind not in CRM_MUTABLE_KINDS: raise ValueError("kind must be customers, opportunities or tasks; activities are append-only") if not isinstance(record_id, str) or not record_id or len(record_id) > 80: raise ValueError("record_id must be a valid CRM record ID") if isinstance(version, bool) or not isinstance(version, int) or version < 1: raise ValueError("version must be the current positive record version") if not isinstance(changes, dict) or not changes: raise ValueError("changes must contain at least one CRM field") response = _request_json( base_url, api_key, f"/api/v1/crm/{quote(record_id, safe='')}", method="PATCH", payload={**changes, "entity": kind, "version": version}, ) result = response.get("record") if not isinstance(result, dict) or result.get("id") != record_id: raise RuntimeError("Signal Trade returned an unexpected updated CRM record") return result def create_crm_task( api_key: str, create_id: str, task: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Create a CRM follow-up task; reuse create_id only for an identical retry.""" return create_crm_record(api_key, "tasks", create_id, task, base_url) def update_crm_task( api_key: str, task_id: str, version: int, changes: dict[str, Any], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Update a task with its current version; fetch it again after HTTP 409 or timeout.""" return update_crm_record(api_key, "tasks", task_id, version, changes, base_url) EVENT_FEEDS = {"appraisals", "preparation", "crm"} CRM_KIND_BY_EVENT = { "customer": "customers", "opportunity": "opportunities", "task": "tasks", "activity": "activities", } EVENT_ACTIONS = { "appraisals": { "appraisal_record": {"appraisal_record_created", "appraisal_record_updated"}, }, "preparation": { "preparation_job": {"preparation_job_created", "preparation_job_updated"}, }, "crm": { "customer": {"customer_created", "customer_updated"}, "opportunity": {"opportunity_created", "opportunity_updated"}, "task": {"task_created", "task_updated"}, "activity": {"activity_created", "activity_updated"}, }, } def get_trade_event_checkpoint( api_key: str, feed: str, base_url: str = DEFAULT_BASE_URL, ) -> str: """Capture a server checkpoint before a full sync for the selected feed.""" if feed not in EVENT_FEEDS: raise ValueError("feed must be appraisals, preparation or crm") result = _request_json(base_url, api_key, f"/api/v1/{feed}/events?limit=1") checkpoint = result.get("pollStartedAt") if not _valid_iso_datetime(checkpoint): raise RuntimeError(f"Signal Trade did not return a valid {feed} poll checkpoint") return checkpoint def sync_trade_records( api_key: str, feed: str, checkpoint: str | None, on_record: Callable[[dict[str, Any], dict[str, Any]], None], save_checkpoint: Callable[[str], None], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Fetch current records changed since a checkpoint and save it last. The callback receives ``(record, event)``. Make writes safe to replay by Signal record ID and CRM kind: a failed request or callback leaves the checkpoint unchanged so the same window is delivered again. """ if feed not in EVENT_FEEDS: raise ValueError("feed must be appraisals, preparation or crm") if not callable(on_record) or not callable(save_checkpoint): raise TypeError("on_record and save_checkpoint are required") events_by_record: dict[str, dict[str, Any]] = {} events_processed = 0 next_checkpoint: str | None = None cursor: str | None = None seen_cursors: set[str] = set() while True: query = {"limit": str(PAGE_SIZE)} if checkpoint: query["updated_since"] = checkpoint if cursor: query["cursor"] = cursor page = _request_json( base_url, api_key, _query_path(f"/api/v1/{feed}/events", query), ) poll_started_at = page.get("pollStartedAt") records = page.get("data") pagination = page.get("pagination") if ( not _valid_iso_datetime(poll_started_at) or not isinstance(records, list) or not isinstance(pagination, dict) ): raise RuntimeError(f"Signal Trade returned an invalid {feed} event page") if next_checkpoint and next_checkpoint != poll_started_at: raise RuntimeError(f"Signal Trade changed the {feed} checkpoint while paging") next_checkpoint = poll_started_at next_cursor = _next_page_cursor(pagination, seen_cursors) for event in records: if ( not isinstance(event, dict) or not isinstance(event.get("id"), str) or not event["id"] or not isinstance(event.get("recordId"), str) or not event["recordId"] or not isinstance(event.get("type"), str) or not isinstance(event.get("action"), str) or not event["action"] or not _valid_iso_datetime(event.get("createdAt")) ): raise RuntimeError(f"Signal Trade returned a malformed {feed} event") if feed == "appraisals" and event["type"] != "appraisal_record": raise RuntimeError("Signal Trade returned an unsupported appraisal event") if feed == "preparation" and event["type"] != "preparation_job": raise RuntimeError("Signal Trade returned an unsupported preparation event") if feed == "crm" and event["type"] not in CRM_KIND_BY_EVENT: raise RuntimeError("Signal Trade returned an unsupported CRM event") if event["action"] not in EVENT_ACTIONS[feed].get(event["type"], set()): raise RuntimeError(f"Signal Trade returned an unsupported {feed} event action") kind = CRM_KIND_BY_EVENT.get(event["type"], "") key = f"{kind}:{event['recordId']}" if kind else event["recordId"] events_by_record[key] = event events_processed += len(records) cursor = next_cursor if not cursor: break if not next_checkpoint: raise RuntimeError(f"Signal Trade did not return a {feed} checkpoint") groups: dict[str, list[dict[str, Any]]] = {} for event in events_by_record.values(): group = event["type"] if feed == "crm" else feed groups.setdefault(group, []).append(event) records_by_key: dict[str, dict[str, Any]] = {} for group, group_events in groups.items(): for start in range(0, len(group_events), PAGE_SIZE): batch = group_events[start : start + PAGE_SIZE] query = {"ids": ",".join(event["recordId"] for event in batch)} if feed == "appraisals": path = "/api/v1/appraisals" elif feed == "preparation": path = "/api/v1/preparation" else: path = "/api/v1/crm" query["kind"] = CRM_KIND_BY_EVENT[group] response = _request_json(base_url, api_key, _query_path(path, query)) records = response.get("data") if not isinstance(records, list): raise RuntimeError(f"Signal Trade returned no current {feed} batch records") if _next_page_cursor(response.get("pagination"), set()) is not None: raise RuntimeError(f"Signal Trade returned a nonterminal current {feed} batch; checkpoint was not advanced") requested_ids = {event["recordId"] for event in batch} returned_ids: set[str] = set() for record in records: if ( not isinstance(record, dict) or not isinstance(record.get("id"), str) or record["id"] not in requested_ids or record["id"] in returned_ids ): raise RuntimeError(f"Signal Trade returned an invalid {feed} batch record") returned_ids.add(record["id"]) records_by_key[f"{group}:{record['id']}"] = record for event in batch: if f"{group}:{event['recordId']}" not in records_by_key: raise RuntimeError(f"Signal Trade returned no current {feed} record for event {event['id']}") for event in events_by_record.values(): group = event["type"] if feed == "crm" else feed record = records_by_key[f"{group}:{event['recordId']}"] on_record(record, event) save_checkpoint(next_checkpoint) return { "feed": feed, "checkpoint": next_checkpoint, "eventsProcessed": events_processed, "recordsUpdated": len(events_by_record), } def sync_stock( api_key: str, checkpoint: str | None, on_vehicle: Callable[[dict[str, Any]], None], save_checkpoint: Callable[[str], None], on_delete: Callable[[str], None] | None = None, on_sites_changed: Callable[[], None] | None = None, base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Apply one stock event poll and save its checkpoint only after all work succeeds. Make on_vehicle and on_delete safe to replay by Signal record ID. Every touched ID is hydrated; a current row wins over historical deletion events. Missing rows require deletion evidence from the completed event poll. If a read or callback fails, the checkpoint stays unchanged and can be retried. """ if not callable(on_vehicle) or not callable(save_checkpoint): raise TypeError("on_vehicle and save_checkpoint are required") if checkpoint is not None and not _valid_iso_datetime(checkpoint): raise ValueError("checkpoint must be a valid ISO date-time") touched_record_ids: dict[str, None] = {} deletion_evidenced_ids: set[str] = set() events_processed = 0 sites_changed = False cursor: str | None = None seen_cursors: set[str] = set() next_checkpoint: str | None = None while True: query = {"limit": str(PAGE_SIZE)} if checkpoint: query["updated_since"] = checkpoint if cursor: query["cursor"] = cursor page = _request_json(base_url, api_key, _query_path("/api/v1/stock/events", query)) if page.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") page_checkpoint = page.get("pollStartedAt") if not _valid_iso_datetime(page_checkpoint): raise RuntimeError("Signal Trade did not return pollStartedAt; checkpoint was not advanced") if checkpoint and _iso_time(page_checkpoint) < _iso_time(checkpoint): raise RuntimeError("Signal Trade returned a backward stock checkpoint; checkpoint was not advanced") if next_checkpoint and next_checkpoint != page_checkpoint: raise RuntimeError("Signal Trade changed pollStartedAt while paging; checkpoint was not advanced") next_checkpoint = page_checkpoint events = page.get("data") pagination = page.get("pagination") if not isinstance(events, list) or not isinstance(pagination, dict): raise RuntimeError("Signal Trade returned an invalid stock event page") next_cursor = _next_page_cursor(pagination, seen_cursors) events_processed += len(events) for event in events: if ( not isinstance(event, dict) or not isinstance(event.get("id"), str) or not event["id"] or not isinstance(event.get("action"), str) or STOCK_EVENT_TYPES.get(event["action"]) != event.get("type") or not isinstance(event.get("recordId"), str) or not event["recordId"] or not _valid_iso_datetime(event.get("createdAt")) ): raise RuntimeError("Signal Trade returned a malformed or unsupported stock event") if _iso_time(event["createdAt"]) > _iso_time(page_checkpoint) or (checkpoint and _iso_time(event["createdAt"]) < _iso_time(checkpoint)): raise RuntimeError("Signal Trade returned a stock event outside the poll window; checkpoint was not advanced") if event.get("action") in SITE_ACTIONS: if event["type"] != "business": raise RuntimeError("Signal Trade returned an invalid branch event") sites_changed = True continue if event["type"] == "business": raise RuntimeError("Signal Trade returned an invalid stock event type") record_id = event.get("recordId") if not isinstance(record_id, str) or not record_id: raise RuntimeError("Signal Trade returned a stock event without a record ID") if event.get("action") == "stock_record_deleted": deletion_evidenced_ids.add(record_id) touched_record_ids[record_id] = None cursor = next_cursor if not cursor: break if sites_changed and not callable(on_sites_changed): raise TypeError("on_sites_changed is required when branch settings change") ids_to_load = list(touched_record_ids) vehicles_updated = 0 vehicles_deleted = 0 for offset in range(0, len(ids_to_load), PAGE_SIZE): ids = ids_to_load[offset:offset + PAGE_SIZE] path = _query_path("/api/v1/stock", {"ids": ",".join(ids)}) result = _request_json(base_url, api_key, path) if result.get("version") != "1": raise RuntimeError("Signal Trade returned an invalid stock API version") records = result.get("data") if not isinstance(records, list): raise RuntimeError("Signal Trade returned an invalid current-stock response") if _next_page_cursor(result.get("pagination"), set()) is not None: raise RuntimeError("Signal Trade returned a nonterminal current-stock batch; checkpoint was not advanced") requested_ids = set(ids) records_by_id: dict[str, dict[str, Any]] = {} for vehicle in records: if ( not isinstance(vehicle, dict) or not isinstance(vehicle.get("id"), str) or vehicle["id"] not in requested_ids or vehicle["id"] in records_by_id ): raise RuntimeError("Signal Trade returned a stock record without an ID") records_by_id[vehicle["id"]] = vehicle missing_ids = requested_ids - records_by_id.keys() if not missing_ids.issubset(deletion_evidenced_ids): raise RuntimeError("Signal Trade returned an incomplete current-stock batch; checkpoint was not advanced") if missing_ids and not callable(on_delete): raise TypeError("on_delete is required when current stock contains evidenced deletions") # Validate the entire batch before callbacks. A current row wins over # historical deletion evidence, including recreation at the same time. for record_id in ids: if record_id in records_by_id: on_vehicle(records_by_id[record_id]) vehicles_updated += 1 else: on_delete(record_id) # type: ignore[misc] vehicles_deleted += 1 if sites_changed: on_sites_changed() # type: ignore[misc] if not next_checkpoint: raise RuntimeError("Signal Trade did not return pollStartedAt; checkpoint was not advanced") save_checkpoint(next_checkpoint) return { "checkpoint": next_checkpoint, "eventsProcessed": events_processed, "vehiclesUpdated": vehicles_updated, "vehiclesDeleted": vehicles_deleted, "sitesChanged": sites_changed, } def sync_stock_deletions( api_key: str, checkpoint: str | None, on_delete: Callable[[str], None], save_checkpoint: Callable[[str], None], base_url: str = DEFAULT_BASE_URL, ) -> dict[str, Any]: """Poll the retained ID-only deletion feed using its independent checkpoint. Omit checkpoint on the first call to replay the complete retained history. Keep callbacks idempotent because a failure before checkpoint storage replays the same deletions on the next call. """ if not api_key or not callable(on_delete) or not callable(save_checkpoint): raise TypeError("api_key, on_delete and save_checkpoint are required") cursor: str | None = None seen_cursors: set[str] = set() next_checkpoint: str | None = None deletions_processed = 0 vehicles_deleted = 0 while True: query = {"limit": str(PAGE_SIZE)} if checkpoint: query["updated_since"] = checkpoint if cursor: query["cursor"] = cursor page = _request_json(base_url, api_key, _query_path("/api/v1/stock/deletions", query)) page_checkpoint = page.get("pollStartedAt") if not _valid_iso_datetime(page_checkpoint): raise RuntimeError("Signal Trade did not return a valid deletion pollStartedAt; checkpoint was not advanced") if next_checkpoint and next_checkpoint != page_checkpoint: raise RuntimeError("Signal Trade changed the deletion checkpoint while paging; checkpoint was not advanced") next_checkpoint = page_checkpoint records = page.get("data") pagination = page.get("pagination") if not isinstance(records, list) or not isinstance(pagination, dict): raise RuntimeError("Signal Trade returned an invalid stock deletion page") next_cursor = _next_page_cursor(pagination, seen_cursors) deleted_ids: set[str] = set() for deletion in records: if ( not isinstance(deletion, dict) or not isinstance(deletion.get("id"), str) or not deletion["id"] or not _valid_iso_datetime(deletion.get("deletedAt")) ): raise RuntimeError("Signal Trade returned an invalid stock deletion; checkpoint was not advanced") deleted_ids.add(deletion["id"]) deletions_processed += len(records) for record_id in deleted_ids: on_delete(record_id) vehicles_deleted += len(deleted_ids) cursor = next_cursor if not cursor: break if not next_checkpoint: raise RuntimeError("Signal Trade did not return pollStartedAt; checkpoint was not advanced") save_checkpoint(next_checkpoint) return { "checkpoint": next_checkpoint, "deletionsProcessed": deletions_processed, "vehiclesDeleted": vehicles_deleted, } # Native inspection sync is separate from the old figure-import API and never # requests private files. Workflow/evidence scope is required by these routes. def _native_id(value: Any) -> str: if not _native_record_id(value): raise ValueError("A saved appraisal ID is required") return quote(value, safe="") def _native_record_id(value: Any) -> bool: return isinstance(value, str) and 0 < len(value.encode("utf-16-le", errors="surrogatepass")) // 2 <= 80 and re.search(r"[\s,\x00-\x1f\x7f-\x9f]", value) is None def _native_token(value: Any) -> bool: return isinstance(value, str) and 0 < len(value) <= 1024 def _native_next_cursor(pagination: Any, seen: set[str]) -> str | None: cursor = _next_page_cursor(pagination, seen) if cursor is not None and not _native_token(cursor): raise RuntimeError("Invalid native pagination cursor; no checkpoint was advanced") return cursor def _native_summary(row: Any) -> bool: if not isinstance(row, dict): return False linked = row.get("linkedStock") valid_id = _native_record_id return (valid_id(row.get("id")) and _valid_iso_datetime(row.get("createdAt")) and _valid_iso_datetime(row.get("updatedAt")) and isinstance(row.get("status"), str) and bool(row["status"]) and "mode" in row and row["mode"] in {"quick", "full", None} and "sourceAppraisalId" in row and (row["sourceAppraisalId"] is None or valid_id(row["sourceAppraisalId"])) and "continuationAppraisalId" in row and (row["continuationAppraisalId"] is None or valid_id(row["continuationAppraisalId"])) and "managerReviewStatus" in row and row["managerReviewStatus"] in {"unreviewed", "requested", "reviewed", None} and "linkedStock" in row and (linked is None or isinstance(linked, dict) and valid_id(linked.get("id")) and _valid_iso_datetime(linked.get("updatedAt")) and isinstance(linked.get("stockStatus"), str))) def _native_page(page: dict[str, Any]) -> None: rows = page.get("data") if (page.get("version") != "1" or not isinstance(rows, list) or len(rows) > PAGE_SIZE or not _native_token(page.get("syncCheckpoint")) or not _valid_iso_datetime(page.get("pollStartedAt")) or not _valid_iso_datetime(page.get("generatedAt")) or any(not _native_summary(row) for row in rows)): raise RuntimeError("Invalid native appraisal inventory; no checkpoint was advanced") def _native_detail(result: dict[str, Any], appraisal_id: str) -> dict[str, Any]: if not isinstance(result, dict): raise RuntimeError("Invalid native appraisal detail; no checkpoint was advanced") row = result.get("appraisal") if (result.get("version") != "1" or not isinstance(row, dict) or row.get("id") != appraisal_id or not _valid_iso_datetime(row.get("updatedAt")) or not _valid_iso_datetime(row.get("createdAt")) or not isinstance(row.get("evidence"), list) or not isinstance(row.get("facts"), dict) or "input" not in row or row["input"] is not None and not isinstance(row["input"], dict) or "vehicle" not in result or result["vehicle"] is not None and (not isinstance(result["vehicle"], dict) or not _native_record_id(result["vehicle"].get("id")) or not _valid_iso_datetime(result["vehicle"].get("updatedAt"))) or not _valid_iso_datetime(result.get("generatedAt")) or not _native_seller_submission(row.get("sellerSubmission"))): raise RuntimeError("Invalid native appraisal detail; no checkpoint was advanced") return result def _native_seller_submission(value: Any) -> bool: if value is None: return True if not isinstance(value, dict) or value.get("source") != "seller_link" or not _valid_iso_datetime(value.get("submittedAt")) or not isinstance(value.get("answers"), dict): return False fields = {"mileage", "keyCount", "startsAndDrives", "warningLights", "knownDamage", "mechanicalIssues", "serviceHistory", "v5cAvailable", "missingEquipment"} answers = value["answers"] return (set(answers) == fields and all(answers[name] is None or type(answers[name]) is int and 0 <= answers[name] <= maximum for name, maximum in [("mileage", 2000000), ("keyCount", 20)]) and all(isinstance(answers[name], str) and answers[name] in {"yes", "no", "unsure"} for name in fields - {"mileage", "keyCount"})) def get_native_appraisal(api_key: str, appraisal_id: str, base_url: str = DEFAULT_BASE_URL, *, transport: Callable[[str], dict[str, Any]] | None = None) -> dict[str, Any]: """Read current private inspection facts, without downloading files.""" request = transport or create_native_appraisal_sync_transport(api_key, base_url) return _native_detail(request(f"/api/v1/appraisals/native/{_native_id(appraisal_id)}"), appraisal_id) def get_native_appraisals(api_key: str, ids: list[str], base_url: str = DEFAULT_BASE_URL, *, transport: Callable[[str], dict[str, Any]] | None = None) -> list[dict[str, Any]]: """Read one to ten complete records, all or nothing, in requested ID order.""" if (not isinstance(ids, list) or not 1 <= len(ids) <= 10 or any(not _native_record_id(record_id) for record_id in ids) or len(set(ids)) != len(ids)): raise ValueError("Supply one to ten distinct native appraisal IDs") request = transport or create_native_appraisal_sync_transport(api_key, base_url) result = request(_query_path("/api/v1/appraisals/native/batch", {"ids": ",".join(ids)})) rows = result.get("data") if result.get("version") != "1" or not isinstance(rows, list) or len(rows) != len(ids) or not _valid_iso_datetime(result.get("generatedAt")): raise RuntimeError("Invalid native detail batch; no checkpoint was advanced") return [_native_detail(row, record_id) for row, record_id in zip(rows, ids)] def _native_batch_size(value: Any) -> int: if type(value) is not int or not 1 <= value <= 10: raise ValueError("detail_batch_size must be a whole number from one to ten") return value def create_native_appraisal_sync_transport(api_key: str, base_url: str = DEFAULT_BASE_URL, *, request_spacing_ms: int = 600, clock: Callable[[], float] | None = None, sleep: Callable[[float], None] = time.sleep) -> Callable[[str], dict[str, Any]]: """Share one GET pacing context across account/discovery/batch/feed reads. Default <=100 requests/minute, including retry attempts. Fake clock/sleep or explicit zero spacing supports local fixtures. Hourly quota stays separate. """ if not api_key or type(request_spacing_ms) is not int or not 0 <= request_spacing_ms <= 60000 or not callable(sleep): raise ValueError("Use a native key, bounded request spacing and clock/sleep functions") now = clock or (lambda: time.monotonic() * 1000) if not callable(now): raise ValueError("A receiver clock function is required") last_started: float | None = None def pace() -> None: nonlocal last_started current = now() if not isinstance(current, (int, float)) or current != current or abs(current) == float("inf"): raise RuntimeError("Invalid native receiver clock") delay = 0 if last_started is None else max(0, last_started + request_spacing_ms - current) if delay: sleep(delay / 1000) last_started = now() def request(path: str) -> dict[str, Any]: if not isinstance(path, str) or not (path == "/api/v1/account" or path.startswith("/api/v1/appraisals/native")): raise ValueError("Native sync transport accepts account and native GET routes only") return _request_json(base_url, api_key, path, before_attempt=pace) return request def load_all_native_appraisals(api_key: str, on_appraisal: Callable[[dict[str, Any]], None], base_url: str = DEFAULT_BASE_URL, *, detail_batch_size: int = 10, transport: Callable[[str], dict[str, Any]] | None = None) -> dict[str, Any]: """Stream exact current detail. Save the inventory checkpoint only after catch-up.""" if not callable(on_appraisal): raise TypeError("on_appraisal must durably apply the current record") _native_batch_size(detail_batch_size) request = transport or create_native_appraisal_sync_transport(api_key, base_url) cursor, checkpoint, poll_started_at, count = None, None, None, 0 seen: set[str] = set() seen_ids: set[str] = set() while True: query = {"limit": str(PAGE_SIZE)} if cursor: query["cursor"] = cursor page = request(_query_path("/api/v1/appraisals/native", query)) _native_page(page) checkpoint = checkpoint or page["syncCheckpoint"] poll_started_at = poll_started_at or page["pollStartedAt"] if checkpoint != page["syncCheckpoint"] or poll_started_at != page["pollStartedAt"]: raise RuntimeError("Native inventory window changed; no checkpoint was advanced") cursor = _native_next_cursor(page.get("pagination"), seen) for row in page["data"]: if row["id"] in seen_ids: raise RuntimeError("Native inventory repeated a record; no checkpoint was advanced") seen_ids.add(row["id"]) if len(seen_ids) > NATIVE_SYNC_MAX_IDS: raise RuntimeError("Native inventory exceeds the reference-client record bound; no checkpoint was advanced") for offset in range(0, len(page["data"]), detail_batch_size): details = get_native_appraisals(api_key, [row["id"] for row in page["data"][offset:offset + detail_batch_size]], base_url, transport=request) for detail in details: on_appraisal(detail) count += 1 if cursor is None: return {"syncCheckpoint": checkpoint, "recordsUpdated": count} def sync_native_appraisal_records(api_key: str, checkpoint: str, on_record: Callable[[dict[str, Any], dict[str, Any]], None], on_delete: Callable[[str, dict[str, Any]], None], save_checkpoint: Callable[[str], None], base_url: str = DEFAULT_BASE_URL, *, detail_batch_size: int = 10, transport: Callable[[str], dict[str, Any]] | None = None) -> dict[str, Any]: """Drain a fixed window and hydrate all touched IDs before checkpointing. Current presence wins over deletion/recreation. Missing IDs need explicit delete evidence. Use idempotent callbacks and one local transaction around this call, as the runnable SQLite example does. File bytes are not loaded. """ if not _native_token(checkpoint) or not all(map(callable, [on_record, on_delete, save_checkpoint])): raise ValueError("The completed native inventory checkpoint and all three durable callbacks are required") _native_batch_size(detail_batch_size) request = transport or create_native_appraisal_sync_transport(api_key, base_url) cursor, next_checkpoint, poll_started_at, last_sequence, events_processed = None, None, None, 0, 0 seen: set[str] = set() seen_event_ids: set[str] = set() touched: dict[str, dict[str, Any]] = {} while True: query = {"limit": str(PAGE_SIZE), "checkpoint": checkpoint} if cursor: query["cursor"] = cursor page = request(_query_path("/api/v1/appraisals/native/events", query)) events = page.get("data") if (page.get("version") != "1" or not isinstance(events, list) or len(events) > PAGE_SIZE or not _valid_iso_datetime(page.get("pollStartedAt")) or not _valid_iso_datetime(page.get("generatedAt"))): raise RuntimeError("Invalid native appraisal event page; no checkpoint was advanced") poll_started_at = poll_started_at or page["pollStartedAt"] if poll_started_at != page["pollStartedAt"]: raise RuntimeError("Native event window changed; no checkpoint was advanced") cursor = _native_next_cursor(page.get("pagination"), seen) if (cursor is not None and page.get("nextCheckpoint") is not None or cursor is None and not _native_token(page.get("nextCheckpoint"))): raise RuntimeError("Invalid native event checkpoint; no checkpoint was advanced") for event in events: if (not isinstance(event, dict) or not isinstance(event.get("id"), str) or not 0 < len(event["id"]) <= 80 or event["id"] in seen_event_ids or not _native_record_id(event.get("appraisalId")) or event.get("action") not in {"upsert", "delete"} or type(event.get("sequence")) is not int or not last_sequence < event["sequence"] <= 9007199254740991 or not _valid_iso_datetime(event.get("recordedAt"))): raise RuntimeError("Invalid native event identity or ordering; no checkpoint was advanced") last_sequence = event["sequence"] seen_event_ids.add(event["id"]) events_processed += 1 previous = touched.get(event["appraisalId"], {}) touched[event["appraisalId"]] = {"event": event, "deleteEvidence": previous.get("deleteEvidence", False) or event["action"] == "delete"} if len(touched) > NATIVE_SYNC_MAX_IDS or events_processed > NATIVE_SYNC_MAX_EVENTS: raise RuntimeError("Native event window exceeds the reference-client bound; no checkpoint was advanced") if cursor is None: next_checkpoint = page["nextCheckpoint"] break records_updated, records_deleted = 0, 0 ids = list(touched) for offset in range(0, len(ids), PAGE_SIZE): batch = ids[offset:offset + PAGE_SIZE] expected = set(batch) page = request(_query_path("/api/v1/appraisals/native", {"limit": str(PAGE_SIZE), "ids": ",".join(batch)})) _native_page(page) if _native_next_cursor(page.get("pagination"), set()) is not None: raise RuntimeError("Native current-ID batch is nonterminal; no checkpoint was advanced") current: set[str] = set() for row in page["data"]: if row["id"] not in expected or row["id"] in current: raise RuntimeError("Native current-ID batch contains duplicate or unexpected records; no checkpoint was advanced") current.add(row["id"]) for appraisal_id in batch: if appraisal_id not in current and not touched[appraisal_id]["deleteEvidence"]: raise RuntimeError("A changed native appraisal is missing without deletion evidence; no checkpoint was advanced") present = [appraisal_id for appraisal_id in batch if appraisal_id in current] for start in range(0, len(present), detail_batch_size): for detail in get_native_appraisals(api_key, present[start:start + detail_batch_size], base_url, transport=request): on_record(detail, touched[detail["appraisal"]["id"]]["event"]) records_updated += 1 for appraisal_id in batch: if appraisal_id not in current: on_delete(appraisal_id, touched[appraisal_id]["event"]) records_deleted += 1 save_checkpoint(next_checkpoint) return {"checkpoint": next_checkpoint, "eventsProcessed": events_processed, "recordsUpdated": records_updated, "recordsDeleted": records_deleted}