Why Every Data Engineer Should Learn Selenium in 2026

# If the data is in the raw HTML, you do not need a browser.
# pip install requests beautifulsoup4
import requests
from bs4 import BeautifulSoup

url = "https://example.com/reports"
headers = {"User-Agent": "data-team-bot/1.0 (contact: data@yourcompany.com)"}
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()

soup = BeautifulSoup(resp.text, "html.parser")
rows = soup.select("table tbody tr")
print(f"Rows found without a browser: {len(rows)}")
# 0 rows here but data visible in Chrome? It is loaded by JavaScript.

# Python's standard library can read robots.txt rules.
from urllib.robotparser import RobotFileParser

agent = "data-team-bot"
rp = RobotFileParser()
rp.set_url("https://example.com/robots.txt")
rp.read()

for path in ["/reports", "/admin", "/search"]:
    allowed = rp.can_fetch(agent, "https://example.com" + path)
    print(f"{path:<10} allowed: {allowed}")

delay = rp.crawl_delay(agent)
print("Crawl-delay:", delay if delay else "not set - still go slow")

# pip install selenium   (Selenium 4.6+ manages the driver itself)
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

opts = webdriver.ChromeOptions()
opts.add_argument("--headless=new")
driver = webdriver.Chrome(options=opts)
records = []
try:
    for page in range(1, 4):
        driver.get(f"https://example.com/prices?page={page}")
        WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, "table tbody tr")))
        for row in driver.find_elements(By.CSS_SELECTOR, "table tbody tr"):
            cells = [c.text.strip() for c in row.find_elements(By.TAG_NAME, "td")]
            if len(cells) >= 3:
                records.append({"item": cells[0], "region": cells[1], "price": cells[2]})
        time.sleep(5)  # be polite between pages
finally:
    driver.quit()

# Continue from the collector: turn records into a clean table.
import pandas as pd

df = pd.DataFrame(records)
df["price"] = pd.to_numeric(
    df["price"].str.replace(r"[^0-9.]", "", regex=True), errors="coerce")
df["collected_at"] = pd.Timestamp.now(tz="Asia/Kolkata")

# Basic quality checks before loading anywhere
assert not df.empty, "No rows collected - page layout may have changed"
assert df["item"].notna().all(), "Missing item names"
bad = df["price"].isna().sum()
print(f"{len(df)} rows, {bad} rows with unreadable price")

df.to_parquet("raw_prices.parquet", index=False)  # needs pyarrow

# test_dashboard.py  - run with: pytest -q test_dashboard.py
import os
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

DASH_URL = os.environ["DASH_URL"]
EXPECTED_TOTAL = float(os.environ["EXPECTED_TOTAL"])  # from warehouse SQL

def test_revenue_tile_matches_warehouse():
    opts = webdriver.ChromeOptions()
    opts.add_argument("--headless=new")
    driver = webdriver.Chrome(options=opts)
    try:
        driver.get(DASH_URL)
        tile = WebDriverWait(driver, 30).until(
            EC.visibility_of_element_located((By.CSS_SELECTOR, "[data-test='total-revenue']")))
        shown = float(tile.text.replace(",", "").strip())
        assert abs(shown - EXPECTED_TOTAL) < 0.01, f"Dashboard {shown} vs warehouse {EXPECTED_TOTAL}"
    finally:
        driver.quit()

# Last task in an Airflow DAG (Airflow 2.x)
from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="sales_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:
    load = BashOperator(task_id="load_sales", bash_command="python load_sales.py")
    check_dash = BashOperator(
        task_id="check_dashboard",
        bash_command="pytest -q /opt/checks/test_dashboard.py",
    )
    load >> check_dash

WEB DATA SOURCE CHECKLIST (fill before writing code)

Source URL        : ______________________________
Owner / contact   : ______________________________
[ ] Official API available?          -> use it
[ ] Bulk download / CSV export?      -> use it
[ ] Terms of use allow automated access?  Y / N
[ ] robots.txt allows these paths?        Y / N
[ ] Login, paywall or CAPTCHA present?    -> stop, ask owner
[ ] Personal data involved?               -> legal review
Fields needed     : ______________________________
Refresh frequency : daily / weekly / one-off
Approved by       : ______________  Date: ________

# collector_skeleton.py - structure for a maintainable collector
import logging, time
from selenium import webdriver

logging.basicConfig(level=logging.INFO)
PAUSE_SECONDS = 5

def make_driver():
    opts = webdriver.ChromeOptions()
    opts.add_argument("--headless=new")
    return webdriver.Chrome(options=opts)

def extract(driver, url):
    driver.get(url)
    # wait for elements, read only the fields you need
    return []

def run(urls):
    driver = make_driver()
    out = []
    try:
        for url in urls:
            out.extend(extract(driver, url))
            logging.info("%s -> %d rows", url, len(out))
            time.sleep(PAUSE_SECONDS)
    finally:
        driver.quit()
    return out

DASHBOARD VALIDATION PLAN

Dashboard        : ______________  Owner: __________
Runs after       : pipeline task ______________
Service account  : read-only, stored in secrets manager

Check                  Selector              Compare with
---------------------  --------------------  --------------------
Page loads in < 30s    [data-test=main]      time limit
Total revenue          [data-test=total]     SQL: SUM(amount)
Last updated label     [data-test=updated]   max(load_ts)
Default date filter    [data-test=period]    "Last 7 days"

On failure: alert channel ______________