Sync a database

Most integrations don’t call the Feed API once — they mirror it into a database and keep that mirror current. The pattern is always the same: an initial backfill, then a scheduled incremental sync, plus a sweep for jobs that expired. Use id as your primary key — it’s stable across updates, so every write is a plain upsert.

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}'

from jobo_enterprise import JoboClient

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

Initial backfill

Page through the entire feed once with cursor pagination. New scans use a stable scan by default (stable_scan: true): pages follow immutable creation time, so jobs updated or expired mid-backfill can’t shift records across page boundaries — no skips, no duplicates. Three rules make the backfill robust:

Persist next_cursor after every successful batch so a crash resumes mid-stream instead of restarting from page one.
Paginate serially — one in-flight request per cursor. Cursors aren’t designed for parallel fan-out.
Send filters and batch_size only on the first request. The cursor embeds them; values sent alongside a cursor are ignored.

Stop when has_more is false.

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

The SDK’s iter_jobs handles cursor pagination for you but doesn’t expose a resume point across process restarts. For a one-off backfill that’s usually fine; for a backfill you need to resume after a crash, use the raw HTTP version and checkpoint next_cursor yourself.

Incremental sync

Run on a schedule — 15–60 minutes is a healthy range. More frequent runs add wallet usage without much fresh data. Each run:

Read your stored last_run_started_at. Record now as this_run_started_at before the first request.
Call /feed with updated_after = last_run_started_at - 15m. The 15-minute overlap protects against clock skew and late-arriving postings.
Page through with cursor until has_more is false. Upsert each job by id.
After the loop succeeds, persist this_run_started_at as the new last_run_started_at.

Use updated_after, not posted_after — posted_after only catches newly posted jobs, while updated_after also catches edits to jobs you already hold. Compare the returned updated_at against what you stored and skip the write when it hasn’t changed.

The Python SDK does not yet expose updated_after on feed.get_jobs / feed.iter_jobs. Build incremental sync against the raw HTTP API, as shown below.

Raw HTTP

"""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}))

Handling deletions

/feed only returns active jobs, so jobs that expire simply stop appearing — it never tells you what disappeared. Sweep them with GET /api/jobs/expired on the same schedule as the incremental sync:

GET /api/jobs/expired?expired_since={last_run_started_at}&batch_size=10000

Page through with cursor, and mark every returned id as expired in your store.

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

expired_since cannot be older than 7 days — the endpoint rejects an older value, and when expired_since is omitted it defaults to just the last 24 hours. If your sync stalls, or simply runs less often than weekly, expired jobs silently accumulate as dead rows in your store with no way to catch them up through this endpoint. Recover by running a full backfill against /feed and reconciling: any ID in your store that no longer appears in the feed has expired.

Backoff and rate limits

Configure a client response timeout of at least 120 seconds. The feed may spend up to 90 seconds querying the search backend before returning a retryable 503.
Retry only 503 and 429, with bounded backoff. On 503, wait the seconds named in Retry-After (currently 5) before retrying — the search backend’s circuit breaker reopens within ~30s.
On 400 (invalid cursor), drop the cursor and restart pagination — don’t loop on the same value.
On 409 feed_cursor_restart_required, the cursor is a legacy non-stable scan that reached the deep-pagination boundary. Discard it and restart with {"stable_scan": true, "batch_size": 1000}, then send only each returned cursor on serial continuation requests.
Watch X-Credits-Balance to alert before you run dry.

See Rate limits and Errors for the full picture.

Managed job sources

POST /api/jobs/feed/managed streams jobs from a private set of employers you’ve added in the Jobo portal, rather than the whole catalogue. It shares the same batch shape, cursor semantics, and metering as the main feed, with a few differences:

The locations filter is not supported on the managed endpoint — filter with sources, work_models, and posted_after instead.
created_at reflects the exact ingestion time on this endpoint. On the main feed, created_at reports indexing time for jobs under 24 hours old, which can lag ingestion slightly.
Requires a customer API key. Sandbox and marketplace keys receive 403.

See Sources for how managed job sources are added and verified.

Get started

How-to guides

Changelog

Initial backfill

Incremental sync

Handling deletions

Backoff and rate limits

Managed job sources

​Initial backfill

​Incremental sync

​Handling deletions

​Backoff and rate limits

​Managed job sources

Initial backfill

Incremental sync

Handling deletions

Backoff and rate limits

Managed job sources