Sync a database

# First request: filters and batch_size
curl -X POST "https://connect.jobo.world/api/jobs/feed" \
  -H "X-Api-Key: $JOBO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"batch_size": 1000}'

# Every later request: only the next_cursor from the previous response
curl -X POST "https://connect.jobo.world/api/jobs/feed" \
  -H "X-Api-Key: $JOBO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"cursor": "<next_cursor>"}'

import os
from jobo_enterprise import JoboClient

with JoboClient(api_key=os.environ["JOBO_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, os, time, pathlib, requests

API_KEY = os.environ["JOBO_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, attempts=5):
    for attempt in range(attempts):
        r = requests.post(f"{BASE}/feed",
                          headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                          json=body, timeout=120)  # the feed can take up to 90 s
        if r.status_code not in (429, 500, 503) or attempt == attempts - 1:
            r.raise_for_status(); return r.json()
        # Honor Retry-After on 429/503; otherwise back off 1 s, 2 s, 4 s … (30 s cap).
        time.sleep(int(r.headers.get("Retry-After", min(2 ** attempt, 30))))

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
import os
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=os.environ["JOBO_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

    # None on the first run falls back to the last 24 hours.
    for job_id in client.feed.iter_expired_job_ids(expired_since=last_run_at, batch_size=10000):
        store.mark_expired(job_id)

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

"""Incremental sync: pages new and updated jobs since the last run using
   updated_after. The cursor is checkpointed so a crash resumes mid-stream.
   Pages serially (one in-flight request per cursor) and retries 429, 500 and 503.
   Sweep expired IDs afterwards: see Handling deletions."""
import json, os, time, datetime as dt, requests, pathlib

API_KEY = os.environ["JOBO_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}
# A resumed run keeps the start time of the run it continues.
this_run_at = state.get("this_run_at") or dt.datetime.now(dt.timezone.utc).isoformat()

def post(body, attempts=5):
    for attempt in range(attempts):
        r = requests.post(f"{BASE}/feed",
                          headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                          json=body, timeout=120)  # the feed can take up to 90 s
        if r.status_code not in (429, 500, 503) or attempt == attempts - 1:
            r.raise_for_status(); return r.json()
        # Honor Retry-After on 429/503; otherwise back off 1 s, 2 s, 4 s … (30 s cap).
        time.sleep(int(r.headers.get("Retry-After", min(2 ** attempt, 30))))

# 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, "this_run_at": this_run_at}))  # 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}))

import os
from jobo_enterprise import JoboClient

last_run_at = store.get_watermark()  # ← your state: a datetime within the last 7 days, None on first run

with JoboClient(api_key=os.environ["JOBO_API_KEY"]) as client:
    # None falls back to the last 24 hours.
    for job_id in client.feed.iter_expired_job_ids(expired_since=last_run_at, batch_size=10000):
        store.mark_expired(job_id)  # ← your DB; job_id is a UUID

# Reuses BASE, API_KEY and state from the incremental sample. Run it before
# committing the new watermark, as the SDK sample does.
cursor = None
while True:
    # None on the first run falls back to the last 24 hours.
    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, timeout=30)
    r.raise_for_status()
    data = r.json()
    for job_id in data["job_ids"]:
        store.mark_expired(job_id)  # ← your DB
    cursor = data["next_cursor"]
    if not data["has_more"]:
        break