Delta Sync nutzen, um geänderte Tabellen effizient zu erkennen

import os

import requests

API_KEY = os.environ["TRASSETS_API_KEY"]
BASE_URL = "https://api.trassets.ai"
HEADERS = {"X-API-Key": API_KEY}


def poll_changes(since: str, etag: str | None = None):
    """Poll /v1/changes and return (changed_tables, new_etag, new_since).

    changed_tables is an empty list on a 304 Not Modified response, and
    new_since stays unchanged then.
    """
    headers = dict(HEADERS)
    if etag:
        headers["If-None-Match"] = etag

    resp = requests.get(
        f"{BASE_URL}/v1/changes", headers=headers, params={"since": since}, timeout=30
    )

    if resp.status_code == 304:
        return [], etag, since

    resp.raise_for_status()
    changed = resp.json()
    if not changed:
        return [], resp.headers.get("ETag"), since
    # last_sync_at is UTC without a suffix; the API expects it with "Z"
    new_since = max(t["last_sync_at"] for t in changed) + "Z"
    return changed, resp.headers.get("ETag"), new_since


if __name__ == "__main__":
    since = "2026-07-01T00:00:00Z"
    etag = None

    changed, etag, since = poll_changes(since, etag)
    for table in changed:
        print(f"Changed: {table['table_name']} -> {table['endpoint_url']}")
    if not changed:
        print("No changes since last poll.")

    # Persist `since` and `etag` and pass them into the next scheduled run.