Sync a database

from jobo_enterprise import JoboClient

with JoboClient(api_key="YOUR_API_KEY") as client:
    for job in client.feed.iter_jobs(work_models=["remote"], batch_size=1000):
        store.upsert(job.id, job)  # ← your DB

"""Backfills the full feed, checkpointing next_cursor after every batch so a
   crash resumes mid-stream instead of restarting from page one."""
import json, time, pathlib, requests

API_KEY = "YOUR_API_KEY"
BASE = "https://connect.jobo.world/api/jobs"
STATE = pathlib.Path("backfill_state.json")

state = json.loads(STATE.read_text()) if STATE.exists() else {"cursor": None}

def post(body):
    while True:
        r = requests.post(f"{BASE}/feed",
                          headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                          json=body)
        if r.status_code == 503:
            time.sleep(int(r.headers.get("Retry-After", "5"))); continue
        r.raise_for_status(); return r.json()

cursor = state["cursor"]
while True:
    # First request carries filters + batch_size. Every later request sends ONLY the cursor.
    body = {"cursor": cursor} if cursor else {"batch_size": 1000}
    data = post(body)

    for job in data["jobs"]:
        store.upsert(job["id"], job)  # ← your DB

    cursor = data["next_cursor"]
    STATE.write_text(json.dumps({"cursor": cursor}))  # checkpoint after every batch
    if not data["has_more"]:
        break

STATE.unlink(missing_ok=True)  # backfill complete

"""Incremental sync through the client. iter_jobs walks the cursor for you;
   the watermark is persisted only after the loop finishes cleanly."""
import datetime as dt
from jobo_enterprise import JoboClient

OVERLAP = dt.timedelta(minutes=15)

last_run_at = store.get_watermark()               # ← your state, None on first run
this_run_at = dt.datetime.now(dt.timezone.utc)

with JoboClient(api_key="YOUR_API_KEY") as client:
    for job in client.feed.iter_jobs(
        updated_after=last_run_at - OVERLAP if last_run_at else None,
        batch_size=1000,
    ):
        store.upsert(job.id, job)                 # ← your DB

    for job_id in client.feed.iter_expired_job_ids():   # defaults to 24h back
        store.mark_expired(job_id)

# Commit the new checkpoint only after the run succeeds.
store.set_watermark(this_run_at)

"""Incremental sync: pages new/updated jobs since the last run using updated_after,
   then sweeps expired IDs. Cursor is checkpointed so a crash resumes mid-stream.
   Pages serially — one in-flight request per cursor — and retries only 503."""
import json, time, datetime as dt, requests, pathlib

API_KEY = "YOUR_API_KEY"
BASE = "https://connect.jobo.world/api/jobs"
STATE = pathlib.Path("sync_state.json")
OVERLAP = dt.timedelta(minutes=15)

state = json.loads(STATE.read_text()) if STATE.exists() else {"last_run_at": None, "cursor": None}
this_run_at = dt.datetime.now(dt.timezone.utc).isoformat()

def post(body):
    while True:
        r = requests.post(f"{BASE}/feed",
                          headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                          json=body)
        if r.status_code == 503:
            time.sleep(int(r.headers.get("Retry-After", "5"))); continue
        r.raise_for_status(); return r.json()

# Page the feed with updated_after = last_run - overlap. A continuation body
# carries ONLY the cursor — it embeds the first request's filters and batch
# size, and other values are ignored.
updated_after = None
if state["last_run_at"]:
    updated_after = (dt.datetime.fromisoformat(state["last_run_at"]) - OVERLAP).isoformat()

cursor = state.get("cursor")
while True:
    if cursor:
        body = {"cursor": cursor}
    else:
        body = {"batch_size": 1000}
        if updated_after:
            body["updated_after"] = updated_after
    data = post(body)

    for job in data["jobs"]:
        store.upsert(job["id"], job)  # ← your DB

    cursor = data["next_cursor"]
    STATE.write_text(json.dumps({**state, "cursor": cursor}))  # checkpoint
    if not data["has_more"]:
        break

# Commit the new checkpoint only after the run succeeds.
STATE.write_text(json.dumps({"last_run_at": this_run_at, "cursor": None}))

from jobo_enterprise import JoboClient
import datetime as dt

with JoboClient(api_key="YOUR_API_KEY") as client:
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=1)
    for job_id in client.feed.iter_expired_job_ids(expired_since=since):
        store.mark_expired(job_id)  # ← your DB; job_id is a UUID

cursor = None
while True:
    params = {"expired_since": state["last_run_at"], "batch_size": 10000}
    if cursor:
        params["cursor"] = cursor
    r = requests.get(f"{BASE}/expired", headers={"X-Api-Key": API_KEY}, params=params).json()
    for job_id in r["job_ids"]:
        store.mark_expired(job_id)  # ← your DB
    cursor = r["next_cursor"]
    if not r["has_more"]:
        break