Introduction
While job boards show you a list of current openings, they don’t preserve the history of those openings. This means you can’t see how listings for a role have changed across different regions or which companies are ramping up amidst the mass layoffs.
What do you know? You can build out that history yourself.
In this article, you’ll build a Python script that snapshots listings for a role across 5 different cities every day so you can plot a chart of the changes as the data grows.
For the job data, you’ll use the SearchApi Google Jobs API. The API returns structured JSON, which means you don’t need to parse any markup.
TL;DR
Google Jobs is stateless, meaning it only shows current openings for today. It has no history. That history is what you’ll build, and you’ll do that by pulling and recording daily snapshots of the listed jobs.
- What you’ll build: A Python script that pulls listings for a particular role across 5 different cities and stores the listings in an SQLite database.
- The cost: SearchApi’s 100-search free plan is enough to build and test the script against real data. In production, though, you’ll need to upgrade. The Developer plan ($40/mo) is sufficient for most production use cases.
- What to look out for: If you keep hitting your
max_pageslimit, know that you’re measuring your own fetch settings instead of the market. Because a parser that fails silently will hand you a confident but wrong answer about the API.
Prerequisites
To follow along with this tutorial, you’ll need:
- Python 3.11 or later
- a SearchApi API key; you can get your API key when you create a free account
Add the following libraries to your requirements.txt file so your runs are reproducible.
requests==2.32.3
pandas==2.2.3
matplotlib==3.9.2
SearchApi gives you 100 free searches, which is enough during development and testing of this Python script. But in production, you’ll easily burn through those 100 searches—5 cities at 2 pages each for a month uses 300 searches.
So when taking your script to production, you want to upgrade to at least the Developer plan, which costs $40/mo and gives you 10,000 searches in the same timeframe. And with that quota, 300 searches is only but a small fraction. This means you can build more similar pipelines while still under the same Developer plan.
Rates are at the time of writing this article. Check the SearchApi pricing page for the latest prices.
Export your API key as an environment variable so you don’t include it in your repo.
export SEARCHAPI_KEY="your_key_here"
Understanding the Google Jobs response
Before you write a single line of code, make a standalone request to the Google Jobs API and read the response.
The following cURL request searches for “backend engineer” roles in Austin and formats the response using Python’s JSON formatter:
curl -G "https://www.searchapi.io/api/v1/search" \
--data-urlencode "engine=google_jobs" \
--data-urlencode "q=backend engineer" \
--data-urlencode "location=Austin,Texas,United States" \
--data-urlencode "gl=us" \
--data-urlencode "api_key=$SEARCHAPI_KEY" | python -m json.tool | head -60
The response has 4 top-level fields:
-
search_metadata: carries the request ID and timings -
search_parameters: includes the location of the job search (it’s a canonical location Google resolves your input to) -
jobs: holds the actual listings data (title, description, location, company name, etc.) -
pagination: holds the token for the next page
The jobs field contains the following fields (kept the ones that matter for this tutorial):
{
"position": 4,
"title": "Java Backend Engineer",
"company_name": "Aurum Groups",
"location": "Fort Worth, TX",
"via": "via Indeed",
"description": "<p>Position :- Java Backend Engineer</p>\n<p>Location :- Fort Worth, TX (Onsite) )Only Local)</p>\n<p>Mandatory Technical Skills:</p>\n<p>Core Java</p>\n<p>Java 11/…",
"extensions": [
"3 days ago",
"125K a year",
"Full-time",
"No degree mentioned",
"Health insurance"
],
"detected_extensions": {
"posted_at": "3 days ago",
"salary": "125K a year",
"schedule": "Full-time",
"no_degree_mentioned": true,
"health_insurance": true
},
"apply_link": "https://www.indeed.com/viewjob?jk=eed5ed5afc3d407f",
"sharing_link": "https://www.google.com/search?ibp=htl;jobs&…&htidocid=OmZ9vxiPLgLgQpOSAAAAAA%3D%3D…"
}
There are three fields that define the build of the Python script:
First, the location field. You can see the location property reads “Fort Worth, TX”, even though you requested the job postings in Austin. Well, that’s because Google resolved the search to a metro region instead of the city.
Second, description comes in as HTML markup. You’ll have to strip the tags when parsing to get the clean text.
The detected_extensions field doesn’t have any property that indicates whether a role is remote or on-site. That’s why the remote classifier, which we’ll build later, relies on the parsed job description, because it’ll likely contain information about what type of job it is.
Some fields aren’t present between listings in one run or between consecutive runs. For example, I ran the query of this article one more time using the same job title and location. One of the listings (position 4) contained a posted_at field, while another (position 1) didn’t.
Before jumping into building a schema for extraction, you want to know that listings with the barest data stay at the top. You may want to rethink using job[0] schema as a reference schema; otherwise, you’ll build the wrong schema.
For counting job listings and analyzing trends, you only need the following fields:
titlecompany_namelocationdetected_extensionssharing_link
Ignore everything else, except for the job description, which you’ll use to get the skills required for the job.
There are two details that'll break your pipeline if you miss them:
Whatever posted_at reads for a listing is only relative to the time you fetched it. Store the snapshot timestamp next to each row if you want a date you can work with.
The response doesn’t come with a job ID field, which makes it difficult to uniquely identify job listings and causes duplicate records of the same listing. To solve this, you’ll use the htidocid key—Google’s internal document ID for each listing—as the unique identifier and deduplication key. Without doing so, you run the risk of polluting your dataset.
For pagination, you’ll pass the token value of the pagination.next_page_token property on the next request.
Fetching postings for one city
The city fetch takes 2 functions:
- one to supply the key
- one to scrape the pages, handle failures, and stay within the rate limit
import os
import time
import requests
SEARCHAPI_URL = "https://www.searchapi.io/api/v1/search"
def api_key() -> str:
"""Read the key at request time so imports stay side-effect free."""
key = os.environ.get("SEARCHAPI_KEY")
if not key:
raise SystemExit("Set SEARCHAPI_KEY before fetching.")
return key
def fetch_city(query: str, location: str, max_pages: int = 2, pause: float = 1.5,
max_retries: int = 2) -> tuple[list[dict], str | None]:
"""Fetch job listings for one query/location pair. Returns (jobs, location_used)."""
jobs: list[dict] = []
token: str | None = None
location_used: str | None = None
page, retries = 0, 0
while page < max_pages:
params = {
"engine": "google_jobs",
"q": query,
"location": location,
"gl": "us",
"hl": "en",
"api_key": api_key(),
}
if token:
params["next_page_token"] = token
try:
resp = requests.get(SEARCHAPI_URL, params=params, timeout=95)
except requests.RequestException as exc:
print(f" [{location}] page {page + 1} network error: {exc}")
break
if resp.status_code == 429:
if retries >= max_retries:
print(f" [{location}] still limited after {max_retries} retries, stopping")
break
retries += 1
print(f" [{location}] rate limited, retry {retries} in 60s")
time.sleep(60)
continue # same page, token unchanged
if resp.status_code != 200:
print(f" [{location}] page {page + 1} HTTP {resp.status_code}: {resp.text[:200]}")
break
retries = 0
data = resp.json()
location_used = location_used or data.get("search_parameters", {}).get("location_used")
page_jobs = data.get("jobs", [])
jobs.extend(page_jobs)
print(f" [{location}] page {page + 1}: {len(page_jobs)} jobs")
page += 1
token = data.get("pagination", {}).get("next_page_token")
if not token or not page_jobs:
break
time.sleep(pause)
return jobs, location_used
The fetch_city function fetches job listings for a role and location from the Google Jobs API. It accepts 3 parameters: max_pages, pause, and max_retries for pagination, throttling, and retries, respectively.
The fetch_city function is simply a paginated loop, running the first call as-is, and then appending next_page_token to the next call to fetch the next page.
Network errors break the loop and return whatever was collected. This means even with network errors, any scraped listing is returned. For 429 rate limit errors, the function pauses for 60 seconds before retrying the request up to max_retries.
The function returns jobs and location_used. location_used is important to know which location Google resolved the search to. Without it, you could see a demand shift in your chart that isn’t necessarily true.
Normalizing the messy parts
This is the part that separates a tracker that works from a script that hands you confident nonsense. Three problems to solve, in ascending order of how much they'll annoy you.
Relative dates into absolute dates
The parser converts Google's relative strings into ISO dates and tags each one with a precision flag, since some strings are far vaguer than others.
import re
from datetime import date, datetime, timedelta
RELATIVE_RE = re.compile(r"(\d+)\s*(\+?)\s*(minute|hour|day|week|month)s?\s+ago", re.I)
FLOOR_HINT = re.compile(r"\b(over|more than|at least)\b", re.I)
IMMEDIATE = {"just posted", "just now", "today", "posted today"}
UNIT_DELTA = {
"minute": lambda n: timedelta(minutes=n),
"hour": lambda n: timedelta(hours=n),
"day": lambda n: timedelta(days=n),
"week": lambda n: timedelta(weeks=n),
"month": lambda n: timedelta(days=30 * n),
}
def parse_posted_at(raw: str | None, snapshot: date) -> tuple[str | None, str]:
"""Convert '3 days ago' into an ISO date plus a precision flag."""
if not raw:
return None, "missing"
text = raw.strip().lower()
if text in IMMEDIATE:
return snapshot.isoformat(), "day"
match = RELATIVE_RE.search(text)
if not match:
return None, "unparsed"
n, plus, unit = int(match.group(1)), match.group(2), match.group(3).lower()
posted = snapshot - UNIT_DELTA[unit](n)
if plus == "+" or FLOOR_HINT.search(text):
precision = "floor" # true date is this old or older
elif unit in ("minute", "hour", "day"):
precision = "day"
else:
precision = "approx" # weeks and months round hard
return posted.isoformat(), precision
Since the value of posted_at is only relative to the time you fetched the listing, you want to filter out roles that have been posted for more than 30 days. Why? Because “30+ days ago” can mean 40 days ago or even 300 days ago. precision != "floor" removes such listings so you don’t have stale listings in your dataset.
Like I mentioned earlier, some job listings don’t come back with a posted_at field. For the request behind this article, the number of rows with that field is 18. For yours, it could be up to 35% of the listings. So, you need to account for the missing rows before you trust your freshness metric.
Detecting genuinely remote roles
You may want to use the location property to determine if a role is remote or not. Please don't. Because an onsite role in San Francisco and a remote-first role that lists San Francisco for tax reasons will both say “San Francisco, CA", which can make you misclassify remote roles.
All job listings for this tutorial didn't have detected_extensions.work_from_home, which is a dead end itself.
So how do you find real remote roles? Well, the classifier below layers 3 signals to return one of 4 labels to classify each listing.
TAG_RE = re.compile(r"<[^>]+>") # descriptions arrive as HTML
REMOTE_LOC = re.compile(r"\b(remote|anywhere|work from home|telecommute)\b", re.I)
HYBRID_HINT = re.compile(
r"\b(hybrid|"
r"\d+\s*days?\s*(?:per|a)\s*week\s*(?:in|at|from)|"
r"return[- ]to[- ]office)\b", re.I
)
ONSITE_HINT = re.compile(r"\b(on-?site|in[- ]office|in[- ]person)\b", re.I)
def classify_work_mode(job: dict) -> str:
"""Return one of: remote, hybrid, onsite, unknown."""
flagged = job.get("detected_extensions", {}).get("work_from_home")
location = job.get("location") or ""
head = TAG_RE.sub(" ", job.get("description") or "")[:2000]
location_says_remote = bool(REMOTE_LOC.search(location))
hybrid = bool(HYBRID_HINT.search(head))
onsite = bool(ONSITE_HINT.search(head))
if flagged is True or location_says_remote:
return "hybrid" if (hybrid or onsite) else "remote"
if hybrid:
return "hybrid"
if onsite or flagged is False:
return "onsite"
return "unknown"
Keep these 3 things in mind when reading the result of the job type classifier:
You want to keep the ONSITE_HINT and HYBRID_HINT regexes separate because a phrase like “you’ll work onsite with the platform team” clearly describes an onsite job, not a hybrid one. And merging the two regexes would label the role as hybrid, which can inflate the count of hybrid roles—what this project exists to measure exactly. If a role is already flagged for a remote work mode, you merge the regexes to check for mentions like “hybrid” or “onsite” so you can downgrade it to hybrid.
Descriptions arrive as HTML, and the tags take up to 7% of the text. You strip the tags before cutting the text down to 2000 characters, because otherwise a phrase describing the work mode sitting near the cutoff will fall outside the 2000-character window and the listing will be wrongly labelled as unknown.
We scan for the first 2000 characters of the description because the top of a listing describes the role itself, and the bottom part is about the company and role benefits. If you scan for everything, you’ll pick up generic texts like “hybrid schedules” and mislabel listings. And if more and more listings come back as unknown, changing the character window is the first place to start making adjustments. But it comes at the cost of more false matches.
Salary out of free text
When looking for a listing’s salary, the parser looks at two places:
-
detected_extensions.salary: currency symbol is optional because we know whatever is there is the salary. - Job description: the currency symbol is mandatory, or else the parser would match version numbers like “Python 3.2.2” or dates as salaries.
In both cases, HTML is stripped and the detection period is annualized.
AMOUNT = r"(\d{1,3}(?:,\d{3})*(?:\.\d+)?)\s*([KkMm])?"
SEPARATOR = r"\s*(?:-|to|\u2013|\u2014)\s*"
PERIOD = r"(?:\s*(?:per|a|an|/)\s*(hour|hr|year|yr|month|mo|week|wk))?"
# The structured field is known to be a salary, so a currency symbol is optional there.
# Free description text needs the "$" anchor to avoid matching version numbers and dates.
FIELD_RANGE = re.compile(r"\$?\s?" + AMOUNT + SEPARATOR + r"\$?\s?" + AMOUNT + PERIOD, re.I)
FIELD_SINGLE = re.compile(r"\$?\s?" + AMOUNT + PERIOD, re.I)
DESC_RANGE = re.compile(r"\$\s?" + AMOUNT + SEPARATOR + r"\$?\s?" + AMOUNT + PERIOD, re.I)
DESC_SINGLE = re.compile(r"\$\s?" + AMOUNT + PERIOD, re.I)
# Dropping the "$" requirement on the field pass lets non-USD amounts through,
# so reject them outright rather than storing foreign currency as dollars.
NON_USD = re.compile(
r"[\u00a3\u20ac\u00a5\u20b9]|\b(?:EUR|GBP|CAD|AUD|INR)\b|\b(?:C|A|CA|AU|NZ)\$", re.I)
ANNUALIZE = {"hour": 2080, "hr": 2080, "week": 52, "wk": 52,
"month": 12, "mo": 12, "year": 1, "yr": 1}
def _to_number(amount: str, suffix: str | None) -> float:
value = float(amount.replace(",", ""))
if suffix and suffix.lower() == "k":
value *= 1_000
elif suffix and suffix.lower() == "m":
value *= 1_000_000
return value
def _extract(text: str, rng: re.Pattern, single: re.Pattern) -> tuple[float, float] | None:
match = rng.search(text)
if match:
low = _to_number(match.group(1), match.group(2))
high = _to_number(match.group(3), match.group(4))
period = (match.group(5) or "year").lower()
else:
match = single.search(text)
if not match:
return None
low = high = _to_number(match.group(1), match.group(2))
period = (match.group(3) or "year").lower()
factor = ANNUALIZE.get(period, 1)
low, high = low * factor, high * factor
if low > high:
low, high = high, low
if not (10_000 <= low <= 2_000_000):
return None
return low, high
def parse_salary(job: dict) -> dict:
"""Annualized salary range from the structured field, falling back to the description."""
field = job.get("detected_extensions", {}).get("salary")
if field and not NON_USD.search(field):
found = _extract(field, FIELD_RANGE, FIELD_SINGLE)
if found:
return {"salary_min": found[0], "salary_max": found[1],
"salary_source": "detected_extensions"}
description = TAG_RE.sub(" ", job.get("description") or "")[:4000]
if description: ""
found = _extract(description, DESC_RANGE, DESC_SINGLE)
if found:
return {"salary_min": found[0], "salary_max": found[1],
"salary_source": "description"}
return {"salary_min": None, "salary_max": None, "salary_source": None}
The wrong outputs in the table below is from real strings, including this project’s snapshot, not assumptions.
| Input | Output | Problem |
|---|---|---|
$132,500 - $157,500 a year |
132500 to 157500 | correct |
$55 - $70 an hour |
114400 to 145600 | annualizes at 2,080 hours, wrong for part-time or short contracts |
Up to $180,000 a year |
180000 to 180000 | a ceiling recorded as a point value |
125K a year (live, no currency symbol) |
125000 to 125000 | correct only after the field pass drops the $ requirement |
£50,000 - £65,000 a year |
discarded by the currency guard | without that guard the range regex fails on the second £, the single-amount pattern then matches 50,000, and a range silently becomes a point value |
Equity grant of $2,000 - $8,000. Base salary $140,000 - $170,000. |
discarded | equity wins the regex race, then fails the sanity floor and takes the real salary with it |
The parser finds the equity range first because it appears first. The 10_000 <= low <= 2_000_000 check rejects it as being too small to be the salary of the listing. But the real salary after the equity is never checked. Since that check weeds out equities, it also weeds out listings with disclosed salaries. That’s the tradeoff: rather lose a good data than store a bad one. If you aren’t comfortable with the tradeoff, you can decide to scan the whole thing.
We include NON_USD because of the £ row. Without it, the range pattern matcher fails on the second £, the single-amount pattern grabs “50,000”, and a salary range in pounds becomes one number in dollars. If you want to keep multiple currencies, store the currency symbol with the exchange rate next to each row.
In an earlier version of this parser, only 9% of the 104 listings disclosed salaries and none of the salary disclosure was in the detected_extensions.salary, even though the salary numbers were there. The issue is the text in that field looks like this: "125K a year". No currency symbol or anything. So, the old $-anchored didn’t match it as a valid salary.
You want to compare the field coverage against the raw response rather than your parser's output, because the two numbers answer different questions.
Also, measure both rates on your own database because coverage changes with the role, city, and time.
has_salary = df.salary_min.notna().mean() * 100
from_field = df[df.salary_min.notna()].salary_source.eq("detected_extensions").mean() * 100
print(f"{has_salary:.0f}% disclose; {from_field:.0f}% of those via detected_extensions")
Whatever number you land on, salary analysis on Google Jobs only describes the employers who choose to disclose. That group skews toward places with pay transparency laws, so treat the rate as a signal about disclosure norms, not about what the market actually pays.
Storing snapshots over time
The value of this project lies in the comparison of roles over time. That’s why the best approach is to store one job per row per day. This will make it easier to compare yesterday’s listings with today’s, and give insight into listing trend.
import hashlib
import sqlite3
from pathlib import Path
from urllib.parse import unquote
DB_PATH = Path(__file__).with_name("jobs.db") # independent of cwd
HTIDOCID_RE = re.compile(r"htidocid=([^&#]+)")
SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshots (
snapshot_date TEXT NOT NULL,
job_key TEXT NOT NULL,
query TEXT NOT NULL,
city TEXT NOT NULL,
title TEXT,
company TEXT,
location TEXT,
via TEXT,
posted_at_raw TEXT,
posted_date TEXT,
posted_precision TEXT,
work_mode TEXT,
schedule TEXT,
salary_raw TEXT,
salary_min REAL,
salary_max REAL,
salary_source TEXT,
position INTEGER,
PRIMARY KEY (snapshot_date, city, query, job_key)
);
CREATE INDEX IF NOT EXISTS idx_key_date ON snapshots(job_key, snapshot_date);
CREATE INDEX IF NOT EXISTS idx_city_date ON snapshots(city, snapshot_date);
"""
def job_key(job: dict) -> str:
"""Google's htidocid, or a content hash when the sharing link is missing."""
match = HTIDOCID_RE.search(job.get("sharing_link") or "")
if match:
return unquote(match.group(1))
seed = "|".join([
(job.get("title") or "").strip().lower(),
(job.get("company_name") or "").strip().lower(),
(job.get("location") or "").strip().lower(),
])
return "h:" + hashlib.sha1(seed.encode()).hexdigest()[:16]
def migrate(conn) -> None:
"""CREATE TABLE IF NOT EXISTS skips existing tables, so add columns explicitly."""
existing = {row[1] for row in conn.execute("PRAGMA table_info(snapshots)")}
for column, decl in [("salary_raw", "TEXT")]:
if column not in existing:
conn.execute(f"ALTER TABLE snapshots ADD COLUMN {column} {decl}")
conn.commit()
def save_snapshot(conn, rows: list[dict]) -> int:
if not rows:
return 0
columns = list(rows[0].keys())
placeholders = ", ".join("?" * len(columns))
sql = (f"INSERT OR REPLACE INTO snapshots ({', '.join(columns)}) "
f"VALUES ({placeholders})")
conn.executemany(sql, [tuple(r[c] for c in columns) for r in rows])
conn.commit()
return len(rows)
If the scheduled runner dies partway through a run and a rerun would’ve double-counted every records before the crash. The composite key (snapshot_date, city, query, job_key) plus INSERT OR REPLACE turns every run into an overwrite.
If you add salary_raw record to your database will break the old database because of column mismatch. The migrate function solves that problem by creating a new column for any that’s missing. And this makes it safe to run on every startup.
Schedule the script for early morning so each snapshot samples a comparable point in the posting cycle. This cron line runs it at 06:15 every morning and logs everything to a file.
15 6 * * * cd /home/you/jobtracker && /usr/bin/env SEARCHAPI_KEY=xxx ./venv/bin/python tracker.py >> run.log 2>&1
GitHub Actions is a good alternative as it gives free hosting and a version history of every snapshot—the workflow commits the database back to the repo after each run.
name: job-snapshot
on:
schedule:
- cron: "15 6 * * *"
workflow_dispatch:
jobs:
snapshot:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.11"
- run: pip install -r requirements.txt
- run: python tracker.py
env:
SEARCHAPI_KEY: ${{ secrets.SEARCHAPI_KEY }}
- run: |
git config user.name "job-bot"
git config user.email "bot@users.noreply.github.com"
git add jobs.db
git commit -m "snapshot $(date -u +%F)" || exit 0
git push
GitHub's scheduled runners aren't fully reliable. Under heavy usage they can start up to an hour late, and during outages they sometimes skip a run entirely. So plot your chart using the actual date of each snapshot instead of assuming there's one for every day. Otherwise, a missed run will look like a quiet day in the market.
Running across multiple cities
The runner loops the fetcher over your city list, normalizes each result through the 3 parsers, and writes one batch per city.
from datetime import UTC, datetime # UTC requires Python 3.11+
CITIES = [
"Austin,Texas,United States",
"Denver,Colorado,United States",
"Seattle,Washington,United States",
"Atlanta,Georgia,United States",
"New York,New York,United States",
]
QUERY = "backend engineer"
def run_snapshot(cities: list[str] = CITIES, max_pages: int = 2) -> None:
snapshot = datetime.now(tz=UTC).date()
conn = sqlite3.connect(DB_PATH)
conn.executescript(SCHEMA)
migrate(conn)
for city in cities:
jobs, location_used = fetch_city(QUERY, city, max_pages=max_pages)
posted = [parse_posted_at(
j.get("detected_extensions", {}).get("posted_at"), snapshot) for j in jobs]
rows = [{
"snapshot_date": snapshot.isoformat(),
"job_key": job_key(j),
"query": QUERY,
"city": city.split(",")[0],
"title": j.get("title"),
"company": j.get("company_name"),
"location": j.get("location"),
"via": (j.get("via") or "").replace("via ", ""),
"posted_at_raw": j.get("detected_extensions", {}).get("posted_at"),
"posted_date": p[0],
"posted_precision": p[1],
"work_mode": classify_work_mode(j),
"schedule": j.get("detected_extensions", {}).get("schedule"),
"salary_raw": j.get("detected_extensions", {}).get("salary"),
"position": j.get("position"),
**parse_salary(j),
} for j, p in zip(jobs, posted)]
if rows:
print(f"{city.split(',')[0]}: saved {save_snapshot(conn, rows)} rows "
f"(resolved as {location_used})")
time.sleep(2)
conn.close()from datetime import UTC, datetime # UTC requires Python 3.11+
CITIES = [
"Austin,Texas,United States",
"Denver,Colorado,United States",
"Seattle,Washington,United States",
"Atlanta,Georgia,United States",
"New York,New York,United States",
]
QUERY = "backend engineer"
def run_snapshot(cities: list[str] = CITIES, max_pages: int = 2) -> None:
snapshot = datetime.now(tz=UTC).date()
conn = sqlite3.connect(DB_PATH)
conn.executescript(SCHEMA)
migrate(conn)
for city in cities:
jobs, location_used = fetch_city(QUERY, city, max_pages=max_pages)
posted = [parse_posted_at(
j.get("detected_extensions", {}).get("posted_at"), snapshot) for j in jobs]
rows = [{
"snapshot_date": snapshot.isoformat(),
"job_key": job_key(j),
"query": QUERY,
"city": city.split(",")[0],
"title": j.get("title"),
"company": j.get("company_name"),
"location": j.get("location"),
"via": (j.get("via") or "").replace("via ", ""),
"posted_at_raw": j.get("detected_extensions", {}).get("posted_at"),
"posted_date": p[0],
"posted_precision": p[1],
"work_mode": classify_work_mode(j),
"schedule": j.get("detected_extensions", {}).get("schedule"),
"salary_raw": j.get("detected_extensions", {}).get("salary"),
"position": j.get("position"),
**parse_salary(j),
} for j, p in zip(jobs, posted)]
if rows:
print(f"{city.split(',')[0]}: saved {save_snapshot(conn, rows)} rows "
f"(resolved as {location_used})")
time.sleep(2)
conn.close()
Your monthly quota works out to cities * queries * pages * days searches. For the run behind this article, the number of searches made is 300. That’s 3% of the Developer plan. If you scale up to 5 titles across 10 cities at 3 pages each, your search is only 4,500. That’s 45% of the same Developer plan.
So you should worry about staying within your monthly allocation rather than what any individual run costs.
But before committing to 2 pages, check whether page 2 actually returns new listings. If not, using 2 pages will cost you unnecessary extra searches.
jobs, _ = fetch_city(QUERY, CITIES[0], max_pages=2)
keys = [job_key(j) for j in jobs]
print(f"{len(keys)} fetched, {len(set(keys))} distinct")
If the numbers are close, then querying 2 pages is worth it. But if the second page mostly repeats the listings from the first, set max_pages to 1. Page 2 for the “backend engineer” search across all 5 cities returned new listings, that’s why we stuck to 2 pages. Long-tailed job titles will often fit into a single page, so run the checks to know you aren’t wasting your search quota.
Putting it together
The snippets above assemble into 2 files, plus the workflow and a database that appears on first run.
jobtracker/
├── tracker.py # fetch, normalize, store (everything through run_snapshot)
├── analyze.py # remote-share chart, plus trend queries later
├── tests/test_tracker.py # parser unit tests, no API key needed
├── requirements.txt
├── jobs.db # created on first run
└── .github/workflows/snapshot.yml
All imports for tracker.py go at the top, in this order, and cover every function in the piece.
import hashlib
import os
import re
import sqlite3
import time
from datetime import UTC, date, datetime, timedelta
from pathlib import Path
from urllib.parse import unquote
import requests
Here are the constants (SEARCHAPI_URL, DB_PATH, SCHEMA, CITIES, QUERY), regex patterns, and function in the order they appear above, and an entry point that accepts 2 flags:
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser(description="Snapshot Google Jobs postings.")
ap.add_argument("--one-city", action="store_true",
help="fetch only the first city in CITIES")
ap.add_argument("--max-pages", type=int, default=2,
help="pages per city (1 halves your quota use)")
args = ap.parse_args()
run_snapshot(cities=CITIES[:1] if args.one_city else CITIES,
max_pages=args.max_pages)
Run python tracker.py --one-city --max-pages 1 in your terminal to confirm your key works and rows are recorded in the database.
analyze.py requires sqlite3, pandas, matplotlib.pyplot, and its own copies of DB_PATH and QUERY. Because api_key() reads the environment lazily, parsers can be tested against fixture strings with no network or key, so the test suite costs nothing to run.
Analyzing the data
The remote-share chart presents a readable data from day one. You get to see how your classifier is doing. But Listings per city needs a couple of weeks of history and higher pagination ceiling to give you any meaningful data.
The script loads all snapshots but only plots a chart of the latest snapshot. Each bar shows the raw count with the percentage, so you can see the numbers for each city. The script uses matplotlib's Agg backend so it runs without display under cron jobs or GitHub Actions, creates images/ if there's none, and stops if the database is empty.
import sqlite3
from pathlib import Path
import matplotlib
matplotlib.use("Agg") # headless, so this runs under cron and Actions
import matplotlib.pyplot as plt
import pandas as pd
DB_PATH = Path(__file__).with_name("jobs.db") # same constants as tracker.py
QUERY = "backend engineer"
IMAGES = Path(__file__).with_name("images") # same cwd independence as DB_PATH
IMAGES.mkdir(exist_ok=True)
conn = sqlite3.connect(DB_PATH)
df = pd.read_sql_query("SELECT * FROM snapshots", conn, parse_dates=["snapshot_date"])
if df.empty:
raise SystemExit("No rows yet. Run tracker.py first.")
latest = df[df.snapshot_date == df.snapshot_date.max()]
totals = latest.groupby("city")["job_key"].nunique()
remote = (latest[latest.work_mode == "remote"]
.groupby("city")["job_key"].nunique()
.reindex(totals.index, fill_value=0))
share = (remote / totals * 100).sort_values()
fig, ax = plt.subplots(figsize=(8, 4.5))
share.plot.barh(ax=ax, color="#3b6ea5")
ax.bar_label(
ax.containers[0],
labels=[f"{share[c]:.0f}% (n={remote[c]}/{totals[c]})" for c in share.index],
padding=4, fontsize=9,
)
ax.set_xlim(0, max(share.max() * 1.5, 10))
ax.set_title(f'Genuinely remote "{QUERY}" postings, {latest.snapshot_date.max():%d %b %Y}')
ax.set_xlabel("% of listings in that city")
ax.set_ylabel("")
ax.spines[["top", "right"]].set_visible(False)
fig.tight_layout()
fig.savefig(IMAGES / "remote_share.png", dpi=150)
You know, without n=7/20 next to the percentage, a reader will assume the sample is big enough to compare cities. Also, put the counts on the chart, not in a caption. So if the charts get shared, the numbers will still be there.
What the first snapshot showed
The first run across 5 cities returned 104 rows. On its own, it doesn’t count for much. You start to see trend after days or even weeks of running the script.
Of the 104 rows, only 9% had their salaries recorded. This number isn’t entirely accurate because the parser rejected any number without a $ attached to it. Well, the real rate is a lot higher. And you can’t fix it in retrospect because the old schema saved the parser’s output and threw away the raw string. The real number starts with the first snapshot written under the salary_raw column.
18 of the 104 listings didn’t have a posting date. For those records, the snapshot date is the only timestamp you’ve got. This means if you built your freshness metric on the posted_at property, you’re only measuring for a subset of the entire listings.
Austin had 25 counts. Denver, New York, and Seattle each had 20, while Atlanta had 19. That’s 104 total. Since each page collects 10 results and you scrape only 2 pages, the counts are based on what your scraper collected, not what exists out there. Only Atlanta returned the exact count of listings in that city. If you want to use the count as the demand signal, you’ll have to increase the number of pages you scrape (max_pages) until your scraper starts returning fewer listings than the pages.
Across all 5 cities, unknown makes up the biggest bucket—53% in Atlanta, 68% in Austin. Real remote roles appeared in 35% of New York listings, 30% of Denver’s, 25% of Seattle’s, 21% of Atlanta’s, and 20% of Austin’s. Only Atlanta (11%) and Austin (8%) had strictly onsite listings. The other didn’t.
You don't want to read too much into these percentages. Each one is based on just 4 to 7 remote listings, so one extra posting will shift a bar by 4 to 5 points. New York beats Austin by only two listings (7 v 5), which is well within the noise.
Most of the listings landed in unknown because their job descriptions never mentioned work arrangements in the first 2000 characters. This means the remote share depends on what you divide by. If it's out of all listings, remote is 20-35%. If it's out of only the listings the classifier could label, remote is 44-88%.
The onsite bucket is almost empty for the same "job description" reason. An employer who's hiring in a city assumes you'll come in, so the job description doesn't say "onsite". Until you expand what ONSITE_HINT matches, the onsite count will be pretty minimal.
Trend claims need 2 things this run doesn't have: a lifted pagination ceiling and several weeks of history. Both arrive on their own once the scheduler runs. The queries waiting for them are below.
What arrives with history
This is where the snapshot design earns its keep. New postings on any given day are just the keys present today and absent yesterday. That's it. The delta query needs at least 2 snapshots to run, but give it a couple of weeks before the output is worth reading.
by_day = df.groupby("snapshot_date")["job_key"].apply(set)
churn = pd.DataFrame({
"new": [len(by_day.iloc[i] - by_day.iloc[i - 1]) for i in range(1, len(by_day))],
"gone": [len(by_day.iloc[i - 1] - by_day.iloc[i]) for i in range(1, len(by_day))],
}, index=by_day.index[1:])
print(churn.describe())
The postings-per-city line chart belongs here too. It only becomes meaningful once your counts stop hitting the pagination ceiling covered in "Limitations.”
daily = (df.groupby(["snapshot_date", "city"])["job_key"]
.nunique().unstack(fill_value=0).sort_index())
fig, ax = plt.subplots(figsize=(11, 5))
daily.plot(ax=ax, marker="o", linewidth=1.8, markersize=4)
ax.set_title(f'Live "{QUERY}" postings by city')
ax.set_ylabel("distinct postings")
ax.set_xlabel("")
ax.grid(alpha=0.25)
ax.legend(title=None, frameon=False, ncol=5)
fig.tight_layout()
fig.savefig(IMAGES / "postings_by_city.png", dpi=150)
Median listing lifespan comes out of the same data. Group by job_key, take the first-seen and last-seen snapshot date for each posting, and you get the median.
Wrapping up
And that’s it. You’ve built a collector that snapshots Google Jobs postings across 5 cities daily. It turns relative dates, work mode, and salary into something you can query and read, and stores every snapshot instead of reseting it like standard job boards do. This is the magic behind analyzing job trends across different regions.
There are parts you’ll have to tinker to get right for your taste. But that’ll be as the collector runs and you see what comes back.
Otherwise, you just need to be patient and allow the scheduler to run a few weeks to build up a solid data you can analyze.
The entire code, including a suite of 39 unit tests are on GitHub. Get your free SearchApi key and have your first snapshot in 15 minutes.
Top comments (6)
really helpful!
Thanks
Wow…this is pretty detailed and hands-on.
Thanks for checking
Some comments may only be visible to logged-in visitors. Sign in to view all comments.