Job market analysis in Python

# Python 3.10+
pip install requests pandas

# fetch.py — pull a bounded dataset with polite rate-limit handling.
import os, time, requests

API_BASE = "https://api.joblistingsapi.com/v1"
API_KEY = os.environ["JLA_API_KEY"]
SESSION = requests.Session()
SESSION.headers["X-API-Key"] = API_KEY


def fetch_jobs(max_rows: int = 1000, page_size: int = 100, **filters) -> list[dict]:
    """Page through /v1/jobs with offset, capping the pull at max_rows.

    The free tier allows 10 requests/minute and 5,000/month, so a 1,000-row
    pull is ten requests — well inside the free budget. On a 429 we honor the
    Retry-After header instead of guessing a backoff.
    """
    jobs: list[dict] = []
    offset = 0

    while len(jobs) < max_rows:
        limit = min(page_size, max_rows - len(jobs))
        res = SESSION.get(
            f"{API_BASE}/jobs",
            params={"limit": limit, "offset": offset, **filters},
        )

        # Rate limited: wait exactly as long as the server asks, then retry.
        if res.status_code == 429:
            wait = int(res.headers.get("Retry-After", "5"))
            time.sleep(wait)
            continue

        res.raise_for_status()
        body = res.json()
        batch = body["jobs"]
        if not batch:
            break  # reached the end of the dataset before max_rows

        jobs.extend(batch)
        offset += len(batch)

        # Be a good citizen between pages even when not rate limited.
        time.sleep(0.2)

    return jobs[:max_rows]


if __name__ == "__main__":
    rows = fetch_jobs(max_rows=1000, country="GB")
    print(f"pulled {len(rows)} jobs")

# analyze.py — normalize the rows into a flat DataFrame.
import pandas as pd
from fetch import fetch_jobs

rows = fetch_jobs(max_rows=1000, country="GB")
df = pd.json_normalize(rows)

# Salary is a nested object that is null on most postings. Pull the fields we
# need to flat columns; rows without disclosed pay become NaN, which is correct.
for field in ("min", "max", "currency"):
    col = f"salary.{field}"
    df[f"salary_{field}"] = df[col] if col in df.columns else pd.NA

# Parse the listing timestamp so we can resample by time. Some postings have no
# listed_at; coerce keeps those as NaT instead of raising.
df["listed_at"] = pd.to_datetime(df["listed_at"], errors="coerce", utc=True)

df["has_salary"] = df["salary_min"].notna()
print(df[["title", "company", "seniority", "role_category", "has_salary"]].head())

# 1. Postings by seniority — where is the hiring weighted?
by_seniority = df["seniority"].value_counts(dropna=False)
print(by_seniority)
# senior    412
# mid       301
# junior    188
# lead       74
# NaN        25  (postings where the source did not declare a level)

# 2. Salary disclosure rate by role category — who actually posts pay?
# Group by category and take the mean of the boolean has_salary column: the
# mean of a 0/1 series is the share that is True, i.e. the disclosure rate.
disclosure = (
    df.groupby("role_category")["has_salary"]
    .mean()
    .sort_values(ascending=False)
    .mul(100)
    .round(1)
)
print(disclosure)
# engineering    34.2
# data           29.8
# sales          18.1
# (each value is the percent of postings in that category that disclosed pay)

# 3. Remote share over time — is remote rising or falling week to week?
# Resample the postings into weekly buckets and take the mean of is_remote,
# which (being boolean) gives the remote share of each week's postings.
weekly_remote = (
    df.dropna(subset=["listed_at"])
    .set_index("listed_at")
    .sort_index()["is_remote"]
    .resample("W")
    .mean()
    .mul(100)
    .round(1)
)
print(weekly_remote)
# Caveat: the live dataset is a rolling 21-day window, so this resample only
# spans ~3 weeks. For a real time series, snapshot on a schedule (see below).