Pull change history
Mirror every plant and explant edit into your own database, incrementally.
Scopes: read:events (plus read:plants / read:explants to fetch full records). Endpoint: List change history.
Every edit to a plant or explant is kept as an event. Reading them incrementally is the cheapest way to keep a copy of your lab's records in a warehouse, a spreadsheet or another system.
How it works
entityis required:plantorexplant. They're separate histories, so pull each one on its own.- Events come oldest first, and
sinceis exclusive: you get events created strictly after it. - Save the
created_atof the last event you processed, and pass it back assincenext time. You only get what changed. - Page through a big catch-up with the cursor: pass
meta.next_cursorback ascursoruntil it'snull.
import os
import requests
BASE = "https://app.xplantpro.com/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['XPLANT_API_KEY']}"}
def pull(entity, since):
"""Returns new events for one entity type, and the cursor to save."""
events = []
params = {"entity": entity, "limit": 200}
if since:
params["since"] = since
while True:
body = requests.get(f"{BASE}/events", headers=HEADERS, params=params, timeout=10).json()
if not body["ok"]:
raise RuntimeError(f"{body['code']}: {body['error']}")
events.extend(body["data"])
next_cursor = (body.get("meta") or {}).get("next_cursor")
if not next_cursor:
break
params["cursor"] = next_cursor
newest = events[-1]["created_at"] if events else since
return events, newest
# Load these from your own storage between runs.
cursors = {"plant": None, "explant": None}
for entity in ("plant", "explant"):
events, cursors[entity] = pull(entity, cursors[entity])
for event in events:
apply_to_my_copy(entity, event) # your code
# Save cursors for next time.Tips
- Store the cursor only after you've applied the events. If your job crashes halfway, the next run picks up where the last successful one ended.
- Overlap, then de-duplicate. Events written together share a timestamp, and a write can land a moment after its timestamp. Start
sinceabout a minute before the newestcreated_atyou've seen, and store events keyed by their id so the overlap is harmless. - Run it on a schedule (every few minutes to hourly) rather than in a tight loop. It's well inside the rate limits either way.