airflow-provider-shopsavvy

April 7, 2026 · View on GitHub

Apache Airflow provider for ShopSavvy — gives your DAGs first-class operators, hooks, and sensors for scheduled price data ETL, data warehouse population, and price drop alerting.

What's included

Class	Description
`ShopSavvyHook`	Connection management and HTTP client for the ShopSavvy Data API
`ShopSavvySearchOperator`	Search products; push results to XCom or write to a file / S3
`ShopSavvyOffersOperator`	Fetch current offers for a product; output as JSON or CSV
`ShopSavvyPriceHistoryOperator`	Pull price history for a date range; output as JSON or CSV
`ShopSavvyPriceSensor`	Polls until a product's price drops to or below a target

Installation

pip install airflow-provider-shopsavvy

Optional extras:

# S3 output support
pip install "airflow-provider-shopsavvy[amazon]"

# Slack alert support (used in the price-monitor example DAG)
pip install "airflow-provider-shopsavvy[slack]"

Requires Apache Airflow 2.4+ and Python 3.8+.

Configure the Airflow connection

Open the Airflow UI and go to Admin → Connections → + Add Connection.
Fill in:

Field Value
Connection Id shopsavvy_default (or any name you prefer)
Connection Type shopsavvy
Host https://data.shopsavvy.com/v1 (leave blank to use the default)
Password Your ShopSavvy API key
Save.

Field	Value
Connection Id	`shopsavvy_default` (or any name you prefer)
Connection Type	`shopsavvy`
Host	`https://data.shopsavvy.com/v1` (leave blank to use the default)
Password	Your ShopSavvy API key

Environment variable fallback

If no Airflow connection is configured, the hook reads SHOPSAVVY_API_KEY from the environment:

export SHOPSAVVY_API_KEY="ss_live_your_key_here"

Quick start

Search for products

from shopsavvy_provider.operators.search import ShopSavvySearchOperator

search = ShopSavvySearchOperator(
    task_id="search_headphones",
    query="sony wh-1000xm5",
    limit=10,
)

Results are pushed to XCom under the key search_results and returned from execute().

Fetch current offers

from shopsavvy_provider.operators.offers import ShopSavvyOffersOperator

offers = ShopSavvyOffersOperator(
    task_id="get_offers",
    identifier="B0CHBNXN1G",       # ASIN, barcode, URL, model number, or name
    output_path="/data/offers.json",
    output_format="json",
)

Pull price history

from shopsavvy_provider.operators.price_history import ShopSavvyPriceHistoryOperator

history = ShopSavvyPriceHistoryOperator(
    task_id="get_price_history",
    identifier="B0CHBNXN1G",
    start_date="{{ macros.ds_add(ds, -30) }}",  # Airflow template macros work here
    end_date="{{ ds }}",
    output_path="/data/history.csv",
    output_format="csv",
)

Wait for a price drop

from shopsavvy_provider.sensors.price import ShopSavvyPriceSensor

wait = ShopSavvyPriceSensor(
    task_id="wait_for_price_drop",
    identifier="B0CHBNXN1G",
    target_price=249.99,
    retailer="amazon.com",     # optional — watches all retailers if omitted
    poke_interval=3600,        # check every hour
    timeout=60 * 60 * 24 * 7, # give up after 7 days
    mode="reschedule",
)

When the sensor fires, it pushes matching offers to XCom under matching_offers.

Write results to S3

Any operator that accepts output_path also accepts S3 URIs:

ShopSavvyOffersOperator(
    task_id="offers_to_s3",
    identifier="B0CHBNXN1G",
    output_path="s3://my-bucket/shopsavvy/{{ ds }}/offers.json",
    aws_conn_id="aws_default",   # your AWS connection
)

Requires pip install "airflow-provider-shopsavvy[amazon]".

Example DAGs

Two ready-to-use example DAGs are included in the package:

DAG	File	Description
`shopsavvy_daily_price_etl`	`example_dags/daily_price_etl.py`	Runs daily, fetches current offers and 30-day price history for a list of tracked products, writes JSON/CSV
`shopsavvy_product_price_monitor`	`example_dags/product_price_monitor.py`	Hourly sensor-driven pipeline that fires a Slack alert and saves a snapshot when a target price is met

Copy them to your DAGs folder or point Airflow's dags_folder at the package's example_dags/ directory to try them out:

# airflow.cfg
[core]
dags_folder = /path/to/shopsavvy_provider/example_dags

Use your own connection ID

All components accept a conn_id parameter so you can manage multiple API keys:

ShopSavvySearchOperator(
    task_id="search",
    query="laptops",
    conn_id="shopsavvy_prod",   # use a non-default connection
)

API

`ShopSavvyHook(conn_id="shopsavvy_default", timeout=30)`

Method	Parameters	Description
`get_conn()`	—	Returns a configured `requests.Session`
`search_products(query, limit, offset)`	`query` required	Search for products
`get_current_offers(identifier, retailer, format)`	`identifier` required	Current offers for a product
`get_price_history(identifier, start_date, end_date, retailer, format)`	`identifier`, `start_date`, `end_date` required	Price history over a date range

`ShopSavvySearchOperator`

Parameter	Type	Default	Description
`query`	`str`	required	Search term
`limit`	`int`	`20`	Max results (max 100)
`offset`	`int`	`0`	Pagination offset
`output_path`	`str`	`None`	Local or S3 path for JSON output
`aws_conn_id`	`str`	`aws_default`	AWS connection for S3 writes
`push_to_xcom`	`bool`	`True`	Push results to XCom
`conn_id`	`str`	`shopsavvy_default`	ShopSavvy connection ID

`ShopSavvyOffersOperator`

Parameter	Type	Default	Description
`identifier`	`str`	required	Barcode, ASIN, URL, model number, or product name
`retailer`	`str`	`None`	Filter to one retailer
`output_path`	`str`	`None`	Local or S3 path
`output_format`	`str`	`json`	`json` or `csv`
`aws_conn_id`	`str`	`aws_default`	AWS connection for S3 writes
`push_to_xcom`	`bool`	`True`	Push results to XCom
`conn_id`	`str`	`shopsavvy_default`	ShopSavvy connection ID

`ShopSavvyPriceHistoryOperator`

Parameter	Type	Default	Description
`identifier`	`str`	required	Barcode, ASIN, URL, model number, or product name
`start_date`	`str`	required	`YYYY-MM-DD`; supports Airflow template macros
`end_date`	`str`	required	`YYYY-MM-DD`; supports Airflow template macros
`retailer`	`str`	`None`	Filter to one retailer
`output_path`	`str`	`None`	Local or S3 path
`output_format`	`str`	`json`	`json` or `csv`
`aws_conn_id`	`str`	`aws_default`	AWS connection for S3 writes
`push_to_xcom`	`bool`	`True`	Push results to XCom
`conn_id`	`str`	`shopsavvy_default`	ShopSavvy connection ID

`ShopSavvyPriceSensor`

Parameter	Type	Default	Description
`identifier`	`str`	required	Barcode, ASIN, URL, model number, or product name
`target_price`	`float`	required	Fires when any offer price is ≤ this value
`retailer`	`str`	`None`	Limit check to one retailer
`conn_id`	`str`	`shopsavvy_default`	ShopSavvy connection ID
`poke_interval`	`int`	Airflow default	Seconds between pokes
`timeout`	`int`	Airflow default	Max wait in seconds
`mode`	`str`	`poke`	`poke` or `reschedule`

Running the tests

pip install "airflow-provider-shopsavvy[dev]"
pytest tests/ -v

Or use the included script:

bash test.sh

License

MIT — see LICENSE.