How to Build a Scheduled Web Scraping Pipeline With Airflow and Scrapeless

TL;DR:

• A scraping pipeline in Airflow is three tasks — fetch, extract, store — wired by passing return values, and the full run here finished in 25.2 seconds with state=success.
• Airflow 3 renamed the scheduling parameter. schedule_interval= raises TypeError: DAG.__init__() got an unexpected keyword argument 'schedule_interval', so most tutorials written for Airflow 2 stop at the DAG definition.
• Install Airflow with its constraints file. Without one, pip install apache-airflow==3.1.2 spent over 13 minutes in dependency resolution here; with the constraints file it completed.
• You do not need Docker to run a DAG. airflow dags test executes the whole graph in-process against SQLite.
• Values passed between tasks are serialised into the metadata database — the fetch task's HTML landed there as 53,889 bytes against 2,154 for the parsed records.
• Running the same DAG twice left 20 rows rather than 40, because the write is an upsert keyed on the product title.
• Start scheduling against rendered pages on the Scrapeless free plan.

A scraper that runs when you remember to run it is a script. Turning it into something that produces a dated table every morning means deciding what a second run does to yesterday's rows and where the credentials live. Airflow handles the scheduling and the task wiring; those two decisions stay yours.

The pipeline below collects a public book catalogue on a daily schedule: a rendered fetch through the Universal Scraping API, a parse into records, and an upsert into SQLite. Every number comes from running it on Airflow 3.1.2.

Pipeline at a Glance

The flow is fetch → extract → store, and each stage hands its return value to the next. Keeping the boundaries at those three points is what lets you re-run one stage without repeating the others, and what keeps the parse logic testable without a network call.

Install Airflow Without the Long Resolve

Airflow depends on a large pinned graph, and installing it without the matching constraints file leaves pip backtracking through version combinations. A plain pip install apache-airflow==3.1.2 here was still resolving after 13 minutes; the same install with the constraints file finished normally.

python3 -m venv .venv
./.venv/bin/pip install "apache-airflow==3.1.2" \
  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.1.2/constraints-3.12.txt"
./.venv/bin/pip install "parsel==1.10.0" "requests==2.32.5"

The constraints URL encodes both the Airflow version and the Python version, and the project's installation reference treats it as part of the install rather than an optimisation. Point AIRFLOW_HOME somewhere explicit and create the metadata database:

export AIRFLOW_HOME="$PWD/home"
./.venv/bin/airflow db migrate

Airflow database tables created
DB: sqlite:////root/verify-airflow/home/airflow.db
Database migrating done!

SQLite is the default backend and it is enough for a single-machine pipeline. The scheduler needs a concurrent database to run tasks in parallel, but nothing below requires that.

Stage 1: Fetch a Rendered Page

The first stage is the one that decides whether the rest of the pipeline sees data at all. A plain HTTP client returns whatever the server sends before JavaScript runs, so the fetch task calls the Universal Scraping API, which returns the rendered document.

@task
def fetch() -> str:
    payload = json.dumps({
        "actor": "unlocker.webunlocker",
        "input": {"url": CATEGORY_URL, "js_render": True, "headless": False},
    }).encode()
    request = urllib.request.Request(
        "https://api.scrapeless.com/api/v2/unlocker/request",
        data=payload,
        headers={"Content-Type": "application/json",
                 "x-api-token": os.environ["SCRAPELESS_API_KEY"]},
    )
    with urllib.request.urlopen(request, timeout=180) as response:
        body = json.loads(response.read())
    html = body["data"]
    print(f"fetched {len(html)} chars")
    return html

fetched 50403 chars

The response envelope is {"code": ..., "data": ...}, where data carries the HTML as a string. Because the API returns JSON, the document arrives already decoded as UTF-8 — the £ in each price survives without any encoding handling in the task.

The API token comes from the environment rather than the DAG file. Airflow reads the process environment when it executes a task, so exporting it in the shell that runs the pipeline keeps the credential out of the repository. Airflow Variables and Connections are the fuller answer once more than one DAG needs it.

Stage 2: Extract Records

The extract stage takes a string and returns records. It touches no network, which means it can be exercised against a saved fixture whenever the markup changes.

@task
def extract(html: str) -> list[dict]:
    sel = Selector(text=html)
    rows = []
    for card in sel.css("article.product_pod"):
        rows.append({
            "title": card.css("h3 a::attr(title)").get(),
            "price": card.css("p.price_color::text").get(),
            "rating": (card.css("p.star-rating::attr(class)").get() or "").replace("star-rating", "").strip(),
            "in_stock": "In stock" in (card.css("p.instock.availability::text").getall() or [""])[-1],
        })
    print(f"extracted {len(rows)} records")
    return rows

extracted 20 records

Selecting each card first and then querying inside it keeps fields from the same product together. Selecting all titles and all prices as two flat lists and zipping them produces silently mismatched pairs the moment one card is missing a field.

Stage 3: Store With an Upsert

A scheduled pipeline writes the same rows again tomorrow, so the write has to be defined in terms of what makes a record unique rather than appending blindly.

@task
def store(rows: list[dict]) -> int:
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""CREATE TABLE IF NOT EXISTS books (
        title TEXT PRIMARY KEY, price TEXT, rating TEXT,
        in_stock INTEGER, scraped_at TEXT)""")
    now = datetime.utcnow().isoformat(timespec="seconds")
    conn.executemany(
        "INSERT OR REPLACE INTO books VALUES (?,?,?,?,?)",
        [(r["title"], r["price"], r["rating"], int(r["in_stock"]), now) for r in rows],
    )
    conn.commit()
    count = conn.execute("SELECT COUNT(*) FROM books").fetchone()[0]
    conn.close()
    print(f"stored; table now holds {count} rows")
    return count

stored; table now holds 20 rows

title is the primary key, and SQLite's INSERT OR REPLACE behaviour deletes the conflicting row before inserting the new one. The stored table after the run:

title                                  price     rating stock
A Murder in Time                       £16.64    One    1
A Study in Scarlet (Sherlock Holmes    £16.73    Two    1
A Time of Torment (Charlie Parker #1   £48.35    Five   1
Boar Island (Anna Pigeon #19)          £59.48    Three  1
Delivering the Truth (Quaker Midwife   £20.89    Four   1
Hide Away (Eve Duncan #20)             £11.84    One    1

Wiring the Stages Into a DAG

The DAG declaration carries the schedule; the last line declares the dependency chain by passing return values.

@dag(
    dag_id="books_pipeline",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 9, 1, tz="UTC"),
    catchup=False,
    tags=["scraping"],
)
def books_pipeline():
    ...
    store(extract(fetch()))


books_pipeline()

schedule takes a standard cron expression, so 0 6 * * * is 06:00 daily in the DAG's timezone. Setting start_date with an explicit zone matters more than it looks: a DAG pinned to a zone that observes daylight saving shifts its real execution time twice a year, and the offsets come from the IANA time zone database. Pinning to UTC keeps the interval fixed. catchup=False matters for scrapers specifically: with it set to True, deploying a DAG whose start_date is a month back queues a run for every missed interval, and a scraper cannot recover a page as it looked three weeks ago anyway.

Writing store(extract(fetch())) is the whole dependency graph. Airflow reads the call nesting and builds the edges.

Run It Without Docker

Most Airflow walkthroughs start with a Docker Compose file and a Postgres container. For developing a DAG, airflow dags test runs the entire graph in-process, without starting the scheduler, the webserver, or a container.

export AIRFLOW_HOME="$PWD/home"
export SCRAPELESS_API_KEY="your-api-key"
./.venv/bin/airflow dags test books_pipeline

fetched 50403 chars
extracted 20 records
stored; table now holds 20 rows
DagRun Finished: dag_id=books_pipeline, run_duration=25.211182, state=success

That run is the fastest feedback loop available while the parse logic is still moving. The scheduler is what you start once the DAG is doing what you want.

Scheduling against pages that render client-side? The Scrapeless free plan covers enough requests to get a daily DAG running end to end.

Running Twice Should Not Double Your Rows

A scheduled pipeline runs unattended, so the interesting test is the second run rather than the first. Executing the same DAG again:

stored; table now holds 20 rows

Twenty, not forty. The upsert keyed on title replaced each row and refreshed its scraped_at, which is what makes the DAG safe to trigger manually while debugging. A plain INSERT would have doubled the table and left no way to tell which copy was current.

If the history matters — tracking how a price moves — the key becomes the pair of title and run date rather than the title alone, and yesterday's row stays.

The XCom Boundary

Return values move between tasks through XCom, and XCom values are serialised into the metadata database. The three stages here stored:

task_id      serialized bytes
fetch        53889
extract      2154
store        2

The HTML is 25 times the size of the records parsed out of it, and it sits in the Airflow database rather than in memory. That is fine at this size and stops being fine as pages get larger or runs get more frequent.

The shape that scales is to parse early and pass small: have the fetch task write the raw document to object storage and return a key, then let extract read it back. The stage boundaries stay identical and the metadata database keeps holding record-sized values instead of documents.

Airflow 2 Code Will Not Run Here Unchanged

Two changes catch anyone following an older tutorial, and only one of them announces itself.

from airflow.decorators import dag, task   # deprecated
from airflow.sdk import dag, task          # Airflow 3

The old import path still resolves and emits DeprecatedImportWarning: The airflow.decorators.dag attribute is deprecated. Please use 'airflow.sdk.dag'. The scheduling parameter is a hard failure:

TypeError: DAG.__init__() got an unexpected keyword argument 'schedule_interval'

schedule_interval= became schedule=. Since almost every published Airflow scraping tutorial predates version 3, that TypeError is the first thing many readers hit — and it surfaces at DAG-parse time, before any scraping code runs.

Conclusion

The pipeline is three functions and a decorator: fetch returns HTML, extract returns records, store returns a count, and store(extract(fetch())) is the graph. What makes it a pipeline rather than a script is the upsert that survives a second run, the schedule expressed as cron, and the stage boundary that lets you re-parse without re-fetching.

Start with airflow dags test and SQLite, keep documents out of XCom once they grow, and pin the install with the constraints file so the first thing you fight is the markup rather than the dependency resolver. For the storage end of the same problem, our DuckDB and Parquet pipeline covers columnar output, and pricing lists what a daily schedule costs in requests.

Ready to put a scraper on a schedule? Start with the Scrapeless free plan and point the fetch stage at your own target.

FAQ

Q: Do I need Docker to run Airflow for scraping?

No. airflow dags test <dag_id> runs the full graph in-process against the default SQLite database, which is how the 25.2-second run above executed. Docker and Postgres become useful when you want the scheduler running unattended with tasks executing in parallel, because SQLite does not support the concurrent writes that requires.

Q: Where should the API key live in an Airflow DAG?

Not in the DAG file. Reading it from the process environment keeps it out of version control, and Airflow Variables or Connections are the better home once several DAGs need the same credential — both are stored in the metadata database and referenced by name rather than pasted into code.

Q: How do I stop a scheduled scraper from creating duplicate rows?

Define what makes a record unique and write against that key. Here title is the primary key and the write is INSERT OR REPLACE, so the second run left 20 rows rather than 40. If you want history instead of current state, widen the key to include the run date so each day's snapshot is its own row.

Q: Is it safe to pass scraped HTML between Airflow tasks?

At small sizes, yes — the 50,403-character document above serialised to 53,889 bytes in the metadata database and moved between tasks without trouble. It stops being a good idea as documents or run frequency grow, because every value lives in that database. Writing the document to object storage and passing a key keeps the same task boundaries with record-sized XCom values.

Q: Why does my Airflow tutorial code raise a TypeError on schedule_interval?

Because it was written for Airflow 2. Version 3 renamed the parameter to schedule, and passing the old name raises TypeError: DAG.__init__() got an unexpected keyword argument 'schedule_interval' while the DAG is being parsed. The companion change is the import: airflow.decorators still works but warns, and airflow.sdk is the current path.

Q: Should catchup be enabled for a scraping DAG?

Usually not. With catchup=True, a DAG whose start_date is in the past queues one run per missed interval as soon as it is deployed. That behaviour suits reprocessing a dated dataset, but a scraper reads whatever the page shows now, so those runs would collect today's data and label it with historical dates.