Jobs API Ingestion Guide: Reliable Historical Backfills and Incremental Syncs

Table of contents

If you are integrating the Jobs endpoint for the first time, there are two proven sync patterns: a date-based approach and an ID-based approach. Both are valid, but they solve slightly different operational concerns. The examples below are designed to be directly executable in Python.

Start with the right mental model

The Jobs API endpoint is GET https://jobdataapi.com/api/jobs/ and returns paginated JSON with count, next, previous, and results. You authenticate using Authorization: Api-Key YOUR_API_KEY in your request header.

A useful way to think about ingestion is that there are two different timelines in your data feed. One is a publication timeline (published) and one is an insertion timeline (id growth). Publication time reflects when a listing was posted, while IDs let you track what entered the dataset since your last run. Many integration issues happen when teams assume those timelines are always equivalent.

This is exactly why there are two recommended sync patterns. Date-based sync is usually easier for humans to reason about and is often enough for job boards and dashboards. ID-based sync is more deterministic for "only new since last checkpoint" pipelines and catches late-added records even if their published date is in the past.

Plan and filtering behavior you should verify before coding

Before implementation, verify your subscription tier because this changes available filtering behavior.

With API access lite, slicing parameters are blocked on /api/jobs/ (published_since, published_until, min_id, max_id, min_age, max_age), and behavior is limited to latest-window feed access.

With API access (or higher), those slicing parameters are available and are what power reliable backfills and incrementals.

Also important: if no slicing parameter is provided, /api/jobs/ applies an implicit max_age=90 window. For non-lite plans, as soon as you provide any slicing parameter, that default window is not auto-applied. This prevents your explicit date or ID query from being silently narrowed.

In practical terms, always make your sync intent explicit in query params. Do not rely on defaults for production ingestion.

Option 1: Date-based sync (the easiest place to start)

The date-based strategy is often the best first implementation because it aligns with how teams schedule operations: initial historical import once, then daily refresh. It is simple to explain to non-engineers and easy to audit at a high level.

For production, an initial sync often starts with published_since=2024-01-01 (or another start date based on your product goals), a high page size, and full sequential pagination. For fast local testing, this tutorial intentionally caps the initial sync to the first 100,000 recent jobs so you can validate behavior quickly. For recurring updates, run a daily job with max_age=2 and skip any job ID that already exists in your store.

Why the overlap (max_age=2) matters: real-world schedulers, deployments, and network paths occasionally fail. If your daily pull has a 24-48 hour overlap, a delayed run is much less likely to create permanent gaps. You will naturally re-encounter some listings, so your importer should skip IDs that already exist instead of storing duplicates and cleaning them up later. This is usually the best trade-off for resilience.

Date-based Python example (runnable)

Save as date_sync_example.py:

#!/usr/bin/env python3
import argparse
import json
import os
import sqlite3
import time
from datetime import date
from typing import Dict, Iterator, List

import requests

API_URL = 'https://jobdataapi.com/api/jobs/'
DB_PATH = 'job_sync_date_based.sqlite3'
INITIAL_SYNC_LIMIT = 100_000


def get_api_key() -> str:
    key = os.getenv('JOBDATA_API_KEY', '').strip()
    if not key:
        raise RuntimeError('Missing JOBDATA_API_KEY environment variable')
    return key


def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        '''
        CREATE TABLE IF NOT EXISTS jobs (
            id INTEGER PRIMARY KEY,
            published TEXT,
            title TEXT,
            company_name TEXT,
            country_codes TEXT,
            raw_json TEXT,
            imported_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
        '''
    )
    conn.execute(
        '''
        CREATE TABLE IF NOT EXISTS sync_runs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            mode TEXT NOT NULL,
            started_at TEXT DEFAULT CURRENT_TIMESTAMP,
            imported_count INTEGER NOT NULL,
            notes TEXT
        )
        '''
    )
    conn.commit()
    return conn


def fetch_pages(params: Dict[str, str], api_key: str) -> Iterator[List[Dict]]:
    headers = {'Authorization': f'Api-Key {api_key}'}
    session = requests.Session()
    page = 1

    print(f'[fetch] starting pagination with params={params}')

    while True:
        query = dict(params)
        query['page'] = page

        for attempt in range(5):
            print(f'[fetch] requesting page={page}, attempt={attempt + 1}')
            resp = session.get(API_URL, headers=headers, params=query, timeout=60)
            if resp.status_code == 429:
                wait_s = min(2 ** attempt, 30)
                print(f'[fetch] rate-limited (429) on page={page}, sleeping {wait_s}s')
                time.sleep(wait_s)
                continue
            resp.raise_for_status()
            payload = resp.json()
            print(
                f"[fetch] page={page} ok, results={len(payload.get('results', []))}, "
                f"has_next={payload.get('next') is not None}"
            )
            break
        else:
            raise RuntimeError(f'Failed after retries for page={page}')

        results = payload.get('results', [])
        if not results:
            print(f'[fetch] page={page} returned no results, stopping')
            return

        yield results

        if not payload.get('next'):
            print(f'[fetch] page={page} was final page, stopping')
            return

        page += 1
        time.sleep(0.12)  # Keep calls sequential and modest.


def insert_new_jobs(conn: sqlite3.Connection, jobs: List[Dict]) -> int:
    inserted = 0
    skipped_existing = 0
    for job in jobs:
        job_id = int(job['id'])
        exists = conn.execute('SELECT 1 FROM jobs WHERE id = ?', (job_id,)).fetchone()
        if exists:
            skipped_existing += 1
            continue

        company_name = (job.get('company') or {}).get('name')
        country_codes = '|'.join(
            c.get('code') for c in (job.get('countries') or []) if c.get('code')
        )

        conn.execute(
            '''
            INSERT INTO jobs (id, published, title, company_name, country_codes, raw_json)
            VALUES (?, ?, ?, ?, ?, ?)
            ''',
            (
                job_id,
                job.get('published'),
                job.get('title'),
                company_name,
                country_codes,
                json.dumps(job),
            ),
        )
        inserted += 1

    conn.commit()
    print(
        f'[import] batch done: received={len(jobs)}, inserted={inserted}, '
        f'skipped_existing={skipped_existing}'
    )
    return inserted


def run_initial_sync(conn: sqlite3.Connection, api_key: str) -> int:
    params = {
        # Fast test bootstrap: recent window + hard cap for quick validation.
        # For a real historical backfill, switch to published_since and remove the cap.
        'max_age': '30',
        'page_size': '5000',
        # Optional market bucket example (North America + selected Europe):
        # 'country_code': 'US|CA|GB|DE|FR|NL',
    }

    print(f'[sync] starting date-based initial sync, limit={INITIAL_SYNC_LIMIT}')

    total_inserted = 0
    processed = 0
    for page_jobs in fetch_pages(params, api_key):
        remaining = INITIAL_SYNC_LIMIT - processed
        if remaining <= 0:
            print('[sync] reached initial sync limit before processing next page chunk')
            break

        chunk = page_jobs[:remaining]
        total_inserted += insert_new_jobs(conn, chunk)
        processed += len(chunk)
        print(f'[sync] initial progress: processed={processed}/{INITIAL_SYNC_LIMIT}')

        if processed >= INITIAL_SYNC_LIMIT:
            print(f'[sync] initial sync reached hard limit={INITIAL_SYNC_LIMIT}, stopping')
            break

    conn.execute(
        'INSERT INTO sync_runs (mode, imported_count, notes) VALUES (?, ?, ?)',
        ('date_initial', total_inserted, f'test bootstrap: max_age=30, limit={INITIAL_SYNC_LIMIT}'),
    )
    conn.commit()
    print(f'[sync] date-based initial complete, inserted={total_inserted}')
    return total_inserted


def run_daily_sync(conn: sqlite3.Connection, api_key: str) -> int:
    params = {
        'max_age': '2',
        'page_size': '5000',
        # Optional APAC bucket example (AU/NZ):
        # 'country_code': 'AU|NZ',
    }

    print('[sync] starting date-based daily incremental sync')

    total_inserted = 0
    for page_jobs in fetch_pages(params, api_key):
        total_inserted += insert_new_jobs(conn, page_jobs)

    conn.execute(
        'INSERT INTO sync_runs (mode, imported_count, notes) VALUES (?, ?, ?)',
        ('date_daily', total_inserted, f'max_age=2 run_date={date.today().isoformat()}'),
    )
    conn.commit()
    print(f'[sync] date-based daily complete, inserted={total_inserted}')
    return total_inserted


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Date-based Jobdata sync runner')
    parser.add_argument(
        '--mode',
        choices=['initial', 'daily', 'both'],
        default='daily',
        help='Which sync step to run',
    )
    args = parser.parse_args()

    api_key = get_api_key()
    conn = get_conn()

    print('[main] running date-based script')

    if args.mode in ('initial', 'both'):
        initial = run_initial_sync(conn, api_key)
        print(f'Initial sync imported {initial} new jobs')

    if args.mode in ('daily', 'both'):
        daily = run_daily_sync(conn, api_key)
        print(f'Daily sync imported {daily} new jobs')

Run:

python -m pip install requests
export JOBDATA_API_KEY='YOUR_REAL_KEY'
python date_sync_example.py --mode initial
python date_sync_example.py --mode daily

or

python date_sync_example.py --mode both

Option 2: ID-based sync (the precise incremental model)

The ID-based strategy is best when you want strict checkpoint-based ingestion. Instead of asking "what was published recently," you ask "what has been added since the last record I processed?" This keeps daily sync deterministic and is especially useful for analytics pipelines and systems where missed records are costly.

The practical rule is straightforward: persist last_seen_id after each successful run and request min_id=last_seen_id+1 next time. You still keep an "ignore existing IDs" check as a safety net, but the working model is incremental by insertion order.

Why this catches edge cases better

If a listing enters the provider dataset later, its published date may be old, but its id still places it in your "new since last sync" window. This is the key reason mature ingestion systems often prefer ID checkpoints over date-only windows.

ID-based Python example (runnable)

Save as id_sync_example.py:

#!/usr/bin/env python3
import argparse
import json
import os
import sqlite3
import time
from typing import Dict, Iterator, List

import requests

API_URL = 'https://jobdataapi.com/api/jobs/'
DB_PATH = 'job_sync_id_based.sqlite3'
INITIAL_SYNC_LIMIT = 100_000


def get_api_key() -> str:
    key = os.getenv('JOBDATA_API_KEY', '').strip()
    if not key:
        raise RuntimeError('Missing JOBDATA_API_KEY environment variable')
    return key


def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        '''
        CREATE TABLE IF NOT EXISTS jobs (
            id INTEGER PRIMARY KEY,
            published TEXT,
            title TEXT,
            raw_json TEXT,
            imported_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
        '''
    )
    conn.execute(
        '''
        CREATE TABLE IF NOT EXISTS sync_state (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        )
        '''
    )
    conn.commit()
    return conn


def get_state_int(conn: sqlite3.Connection, key: str, default: int = 0) -> int:
    row = conn.execute('SELECT value FROM sync_state WHERE key = ?', (key,)).fetchone()
    return int(row[0]) if row else default


def set_state_int(conn: sqlite3.Connection, key: str, value: int) -> None:
    conn.execute(
        '''
        INSERT INTO sync_state (key, value)
        VALUES (?, ?)
        ON CONFLICT(key) DO UPDATE SET value = excluded.value
        ''',
        (key, str(value)),
    )
    conn.commit()


def fetch_pages(params: Dict[str, str], api_key: str) -> Iterator[List[Dict]]:
    headers = {'Authorization': f'Api-Key {api_key}'}
    session = requests.Session()
    page = 1

    print(f'[fetch] starting pagination with params={params}')

    while True:
        query = dict(params)
        query['page'] = page

        for attempt in range(5):
            print(f'[fetch] requesting page={page}, attempt={attempt + 1}')
            resp = session.get(API_URL, headers=headers, params=query, timeout=60)
            if resp.status_code == 429:
                wait_s = min(2 ** attempt, 30)
                print(f'[fetch] rate-limited (429) on page={page}, sleeping {wait_s}s')
                time.sleep(wait_s)
                continue
            resp.raise_for_status()
            payload = resp.json()
            print(
                f"[fetch] page={page} ok, results={len(payload.get('results', []))}, "
                f"has_next={payload.get('next') is not None}"
            )
            break
        else:
            raise RuntimeError(f'Failed after retries page={page}')

        results = payload.get('results', [])
        if not results:
            print(f'[fetch] page={page} returned no results, stopping')
            return

        yield results

        if not payload.get('next'):
            print(f'[fetch] page={page} was final page, stopping')
            return

        page += 1
        time.sleep(0.12)


def insert_if_new(conn: sqlite3.Connection, job: Dict) -> bool:
    job_id = int(job['id'])
    exists = conn.execute('SELECT 1 FROM jobs WHERE id = ?', (job_id,)).fetchone()
    if exists:
        return False

    conn.execute(
        'INSERT INTO jobs (id, published, title, raw_json) VALUES (?, ?, ?, ?)',
        (job_id, job.get('published'), job.get('title'), json.dumps(job)),
    )
    return True


def initial_bootstrap(conn: sqlite3.Connection, api_key: str) -> int:
    params = {
        # Fast test bootstrap: recent window + hard cap for quick validation.
        # For a real historical backfill, switch to published_since and remove the cap.
        'max_age': '30',
        'page_size': '5000',
        # Optional market bucket example (North America + selected Europe):
        # 'country_code': 'US|CA|GB|DE|FR|NL',
    }

    imported = 0
    skipped_existing = 0
    processed = 0
    highest_id = get_state_int(conn, 'last_seen_id', 0)

    print(f'[sync] starting ID bootstrap, limit={INITIAL_SYNC_LIMIT}')
    print(f'[sync] current last_seen_id before bootstrap={highest_id}')

    for page_jobs in fetch_pages(params, api_key):
        remaining = INITIAL_SYNC_LIMIT - processed
        if remaining <= 0:
            print('[sync] reached bootstrap limit before processing next page chunk')
            break

        chunk = page_jobs[:remaining]
        for job in chunk:
            if insert_if_new(conn, job):
                imported += 1
            else:
                skipped_existing += 1
            highest_id = max(highest_id, int(job['id']))
        processed += len(chunk)
        conn.commit()
        print(
            f'[sync] bootstrap progress: processed={processed}/{INITIAL_SYNC_LIMIT}, '
            f'imported={imported}, skipped_existing={skipped_existing}, '
            f'highest_id={highest_id}'
        )

        if processed >= INITIAL_SYNC_LIMIT:
            print(f'[sync] bootstrap reached hard limit={INITIAL_SYNC_LIMIT}, stopping')
            break

    set_state_int(conn, 'last_seen_id', highest_id)
    print(f'[sync] bootstrap complete, final last_seen_id={highest_id}, imported={imported}')
    return imported


def daily_incremental(conn: sqlite3.Connection, api_key: str) -> int:
    last_seen_id = get_state_int(conn, 'last_seen_id', 0)
    params = {
        'min_id': str(last_seen_id + 1),
        'page_size': '5000',
        # Optional APAC bucket example (AU/NZ):
        # 'country_code': 'AU|NZ',
    }

    imported = 0
    skipped_existing = 0
    highest_id = last_seen_id

    print(f'[sync] starting ID incremental, last_seen_id={last_seen_id}')

    for page_jobs in fetch_pages(params, api_key):
        for job in page_jobs:
            if insert_if_new(conn, job):
                imported += 1
            else:
                skipped_existing += 1
            highest_id = max(highest_id, int(job['id']))
        conn.commit()
        print(
            f'[sync] incremental page processed: imported={imported}, '
            f'skipped_existing={skipped_existing}, highest_id={highest_id}'
        )

    if highest_id > last_seen_id:
        set_state_int(conn, 'last_seen_id', highest_id)
        print(f'[sync] last_seen_id advanced: {last_seen_id} -> {highest_id}')
    else:
        print('[sync] no new IDs found; last_seen_id unchanged')

    return imported


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='ID-based Jobdata sync runner')
    parser.add_argument(
        '--mode',
        choices=['bootstrap', 'incremental', 'both'],
        default='incremental',
        help='Which sync step to run',
    )
    args = parser.parse_args()

    api_key = get_api_key()
    conn = get_conn()

    print('[main] running ID-based script')

    if args.mode in ('bootstrap', 'both'):
        initial_count = initial_bootstrap(conn, api_key)
        print(f'Initial bootstrap imported {initial_count} jobs')

    if args.mode in ('incremental', 'both'):
        daily_count = daily_incremental(conn, api_key)
        print(f'ID incremental imported {daily_count} jobs')

Run:

python -m pip install requests
export JOBDATA_API_KEY='YOUR_REAL_KEY'
python id_sync_example.py --mode bootstrap
python id_sync_example.py --mode incremental

or

python id_sync_example.py --mode both

Regional strategy: why country buckets are usually better than broad region filters

For many commercial datasets, global region filters are good for rough segmentation but too broad for operational pipelines. If your product needs one stream for North America plus selected European markets and another for APAC, split by country_code instead of relying only on region_id.

The API supports multi-code values with |, so you can build explicit market buckets such as country_code=US|CA|GB|DE|FR|NL and country_code=AU|NZ. This gives better control over quality checks, refresh cadence, and downstream reporting. It also makes failures easier to isolate: if one regional pipeline has issues, the others continue normally.

Data handling and reliability practices that prevent expensive rework

Always store the API id with every listing and treat it as your permanent identity key. During imports, first check whether that ID already exists and skip it immediately if it does. This avoids duplicate writes during overlap windows and retries. Persist sync checkpoints (last_seen_id, run time, or both) in durable storage so jobs can resume cleanly after crashes.

Keep requests sequential and moderate. The platform guidance calls out avoiding parallel bursts, and rate-limit responses (429) should be expected and handled with retries and backoff. This is not a sign of failure; it is normal API citizenship and helps both stability and fairness.

Use high page_size for large backfills to reduce request overhead, but test first with smaller pages and narrower filters before scaling up. Early dry-runs catch schema assumptions and parser bugs while the blast radius is still small.

Finally, remember that ingestion is not only about fetching data; it is about producing predictable downstream state. The safest approach is idempotent writes, deterministic checkpoints, and clear run logging.

Which approach should you choose?

If your goal is to launch quickly with a robust daily feed, start with the date-based model (published_since for initial load, then daily max_age=2 while skipping existing IDs during import). If your goal is strict incremental guarantees and easier auditability, use ID-based incrementals (min_id=last_seen_id+1) after bootstrap. In this tutorial code, bootstrap is intentionally capped at 100,000 recent jobs to make testing fast; remove that cap when moving to full production sync.

Many teams begin with date-based ingestion, then add ID-based daily checkpoints once the product matures. That progression is practical and usually minimizes time-to-value without sacrificing long-term correctness.

Common pitfalls for new API consumers

The most frequent mistakes are trying to use slicing filters on access lite, relying on implicit defaults, and running pages in parallel because it "looks faster" in quick tests. Those choices often cause silent data gaps, 429 instability, or hard-to-debug inconsistencies.

The better pattern is explicit filters, sequential pagination, retry/backoff, overlap where useful, and a strict "skip existing IDs" check at write time. If you implement those from day one, your pipeline stays boring in the best possible way.

Operational details that are easy to overlook

Two non-obvious details can save a lot of cleanup work. First, schedule your ingestion in UTC and store run timestamps in UTC so day boundaries are unambiguous across regions and daylight-saving changes. This keeps daily windows and post-run reporting consistent even when your user-facing product is localized.

Second, treat your API key as production credential material from the start. Keep it in environment variables or secret managers, avoid committing it into scripts, and rotate it quickly if you suspect leakage. Most ingestion failures are recoverable; exposed credentials are more disruptive and harder to unwind.

Minimal test plan before going live

Before connecting this to production jobs, run a small-scope validation with a narrow country or other kind of query set and low page size. Confirm authentication, pagination completion, "ignore existing ID" behavior on reruns, and checkpoint advancement only after successful writes. Then increase page size and widen geographic scope.

A small test now is cheaper than a full historical re-import after launch.

Primary references:

• https://jobdataapi.com/c/jobs-api-endpoint-documentation/
• https://jobdataapi.com/c/date-id-and-age-slicing-parameters-documentation/
• https://jobdataapi.com/c/multi-value-parameters-documentation/
• https://jobdataapi.com/accounts/pricing/
• https://jobdataapi.com/llms-full.txt

Related Docs