Crawlee Python: Dynamic Job Listings, Details, and SQLite Storage
Handle delayed job listings with PlaywrightCrawler, follow pages and details, upsert stable job records, and export CSV with failure evidence.
A loaded page can still contain only an empty listing container. Repeatedly retrying that response does not replace understanding when JavaScript supplies the data.
This tutorial connects two dynamic list pages, three job details, SQLite upserts, and CSV export. Complete teaching HTML is supplied in the article and returned through browser request routing, so no separate demo server or project download is required.
Install and select the crawler
Use Python 3.12 and Bash:
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'crawlee[playwright]==1.11.1'
python -m playwright install chromiumIf required fields already exist in response HTML, an HTTP parser such as BeautifulSoupCrawler may be simpler. This example needs a browser because JavaScript inserts the job list. See the official Crawlee Python project.
Define explicit teaching material
Create jobs.py:
import asyncio
import csv
import json
import sqlite3
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from crawlee.crawlers import PlaywrightCrawler
JOBS = {
"j1": {"title": "Python Developer", "location": "Remote", "requirements": "Python, SQL"},
"j2": {"title": "Data Engineer", "location": "Shanghai", "requirements": "ETL, SQL"},
"j3": {"title": "QA Engineer", "location": "Remote", "requirements": "Browser automation"},
}
START = "https://example.16yun.cn/jobs.html?page=1"
def teaching_html(url):
parsed = urlsplit(url)
if parsed.path == "/jobs.html":
number = int(parse_qs(parsed.query).get("page", ["1"])[0])
ids = ["j1", "j2"] if number == 1 else ["j3"]
cards = "".join(
f'<article class="job-card"><a href="/jobs/{key}">{JOBS[key]["title"]}</a></article>'
for key in ids
)
if number == 1:
cards += '<a class="next" href="/jobs.html?page=2">Next</a>'
return (
'<meta charset="utf-8"><h1>Jobs</h1><main id="results"></main>'
'<script>setTimeout(() => {'
f'document.querySelector("#results").innerHTML = {json.dumps(cards)};'
'document.querySelector("#results").dataset.ready = "true";'
'}, 50);</script>'
)
key = parsed.path.rsplit("/", 1)[-1]
job = JOBS.get(key)
if job is None:
return None
return (
'<meta charset="utf-8">'
f'<h1 class="job-title">{job["title"]}</h1>'
f'<p class="location">{job["location"]}</p>'
f'<div class="requirements">{job["requirements"]}</div>'
)The example domain is intercepted locally by the browser routing code; it is not a hosted recruitment service.
The initial list HTML has an empty container. Its script adds cards later. Fifty milliseconds describes the material's delay, not the crawler's readiness policy.
Real sites may place requirements in rich text, panels, or API responses and need different extraction rules.
Connect loading, pagination, and storage
Append to the same file:
async def main():
db = sqlite3.connect("jobs.sqlite")
db.execute("""CREATE TABLE IF NOT EXISTS jobs(
job_id TEXT PRIMARY KEY, title TEXT NOT NULL,
location TEXT NOT NULL, requirements TEXT NOT NULL, source_url TEXT NOT NULL
)""")
failures = []
crawler = PlaywrightCrawler(max_requests_per_crawl=10, max_request_retries=1)
@crawler.pre_navigation_hook
async def material(context):
async def route_request(route):
html = teaching_html(route.request.url)
await route.fulfill(
status=200 if html is not None else 404,
content_type="text/html; charset=utf-8",
body=html or "Not found",
)
await context.page.route("https://example.16yun.cn/**", route_request)
@crawler.router.default_handler
async def listing(context):
await context.page.locator('#results[data-ready="true"]').wait_for()
await context.enqueue_links(selector=".job-card a", label="DETAIL")
await context.enqueue_links(selector="a.next")
@crawler.router.handler("DETAIL")
async def detail(context):
page = context.page
title = (await page.locator(".job-title").inner_text()).strip()
location = (await page.locator(".location").inner_text()).strip()
requirements = (await page.locator(".requirements").inner_text()).strip()
if not title or not location or not requirements:
raise ValueError("职位必填字段缺失")
job_id = urlsplit(context.request.url).path.rsplit("/", 1)[-1]
with db:
db.execute("""INSERT INTO jobs VALUES (?, ?, ?, ?, ?)
ON CONFLICT(job_id) DO UPDATE SET
title=excluded.title, location=excluded.location,
requirements=excluded.requirements, source_url=excluded.source_url""",
(job_id, title, location, requirements, page.url))
@crawler.failed_request_handler
async def failed(context, error):
failures.append({"url": context.request.url, "error_type": type(error).__name__})
try:
await crawler.run([START])
rows = db.execute(
"SELECT job_id,title,location,requirements,source_url FROM jobs ORDER BY job_id"
).fetchall()
with open("jobs.csv", "w", encoding="utf-8-sig", newline="") as stream:
writer = csv.writer(stream)
writer.writerow(["job_id", "title", "location", "requirements", "source_url"])
writer.writerows(rows)
Path("failed.json").write_text(json.dumps(failures, indent=2), encoding="utf-8")
print({"stored_jobs": len(rows), "failed_requests": len(failures)})
finally:
db.close()
asyncio.run(main())Run:
python jobs.pyCrawlee owns its browser resources. The program owns and closes the SQLite connection. Do not invent a separate crawler close API.
Wait for business readiness
DOMContentLoaded does not establish that the jobs were inserted. The explicit data-ready marker connects waiting to list completion.
For a real site, use evidence such as a changed count, a completed response, or a loading marker disappearing. Background traffic makes network-idle conditions unsuitable for some applications.
Preserve pagination parameters. Collapsing page-one and page-two URLs can suppress the second page in scheduling. The request budget also counts list pages, not only jobs: this material needs five pages.
Check stored and exported data
The supplied inputs imply three unique IDs, j1 through j3, and a CSV header plus three rows. These are expected results, not execution measurements.
Check j3 to detect first-page-only collection. Compare its requirements and the location fields with the teaching input to detect extraction mistakes.
The primary key updates an existing job instead of appending a duplicate. It does not infer removal or delete historical rows missing from a later run.
Use the source's actual business identifier when adapting. Cross-site data needs a source dimension. An arbitrary URL suffix is not a universal job identifier.
CSV exports the current database, which may contain earlier observations. Long-running use needs batch or observation metadata to distinguish a current snapshot from history. See the SQLite documentation.
Adapt to an actual recruitment site
Remove the teaching interception hook, replace the entry URL, and update readiness, list, next-page, and detail selectors.
Keep a small category scope and a known page-two sample. Distinguish an explicit empty list from a timeout. Load-more buttons and cursor APIs require different pagination and termination logic.
Troubleshoot
Browser executable missing: check runtime account, environment, and cache location.
Readiness timeout: for the teaching task, verify both blocks share one file and the route hook remains present. For real pages, inspect identity and the actual ready condition.
Three stored rows despite this run failing: historical rows can remain. Reconcile failures and observation batches rather than accepting database count alone.
Duplicates after a rerun: inspect identity stability and avoid using display position or transient parameters as business keys.
Empty requirements: distinguish actual missing information from a broken selector before changing the contract.
Browser rendering works but the batch is incomplete: inspect discovery, pagination parameters, request budget, and detail failures. Readiness cannot recover URLs that never entered the queue.
Rendering is the material-acquisition stage; discovery, stable storage, and coverage checks make it a complete collection task.
Need an enterprise proxy plan?
We can tailor architecture to your target domains, concurrency, and reliability goals.