> ## Documentation Index
> Fetch the complete documentation index at: https://jobo.world/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Sync a database

> Keep a downstream database of jobs in sync with Jobo using cursor pagination, incremental updates, and expiry sweeps.

Most integrations don't call the [Feed API](/docs/api-reference/feed/stream-the-job-feed) 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.

```bash theme={null}
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}'
```

```python theme={null}
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`.

<CodeGroup>
  ```python SDK theme={null}
  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
  ```

  ```python Raw HTTP (checkpointed) theme={null}
  """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
  ```
</CodeGroup>

<Note>
  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.
</Note>

***

## 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:

1. Read your stored `last_run_started_at`. Record `now` as `this_run_started_at` **before** the first request.
2. Call `/feed` with `updated_after = last_run_started_at - 15m`. The 15-minute overlap protects against clock skew and late-arriving postings.
3. Page through with `cursor` until `has_more` is `false`. Upsert each job by `id`.
4. 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.

<Note>
  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.
</Note>

```python Raw HTTP theme={null}
"""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`](/docs/api-reference/feed/list-expired-jobs) **on the same schedule** as the incremental sync:

```http theme={null}
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.

<CodeGroup>
  ```python SDK theme={null}
  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
  ```

  ```python Raw HTTP theme={null}
  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
  ```
</CodeGroup>

<Warning>
  `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.
</Warning>

***

## 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](/docs/rate-limits) and [Errors](/docs/errors) for the full picture.

***

## Managed job sources

[`POST /api/jobs/feed/managed`](/docs/api-reference/feed/stream-the-managed-feed) 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](/docs/sources) for how managed job sources are added and verified.