Fetch Amazon prices on a schedule with PriceFetch API, store them in SQLite, and query price history. Complete Python script included.
A Python script that checks Amazon product prices on a schedule and stores every data point in SQLite. You'll be able to query price history, find the lowest price a product has ever been, and spot pricing patterns.
This is the backend for any price tracking app. It runs as a cron job — no server framework needed. The data accumulates over time and you can query it however you want.
Try it yourself — 500 free API credits, no credit card required.
Start FreeSQLite is the right choice here — no server to manage, the database is just a file, and it handles the read/write patterns of a price tracker perfectly. We create a simple schema: one table for products you're tracking, one for price snapshots.
import sqlite3
from datetime import datetime
def init_db(db_path: str = "prices.db") -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT UNIQUE NOT NULL,
name TEXT,
created_at TEXT DEFAULT (datetime('now'))
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER NOT NULL,
price REAL NOT NULL,
currency TEXT NOT NULL,
in_stock INTEGER NOT NULL,
checked_at TEXT DEFAULT (datetime('now')),
FOREIGN KEY (product_id) REFERENCES products(id)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_price_history_product
ON price_history(product_id, checked_at)
""")
conn.commit()
return connThe core loop: iterate through tracked products, fetch the current price from PriceFetch API, and insert a row into the history table. We use synchronous `httpx` here since cron jobs run sequentially and simplicity matters more than speed.
Each successful fetch costs one credit. If you're tracking 10 products every hour, that's 240 credits per day — well within the free tier for testing.
import httpx
import os
API_KEY = os.environ["PRICEFETCH_API_KEY"]
API_URL = "https://api.pricefetch.dev/v1/price"
def fetch_and_store(conn: sqlite3.Connection, product_id: int, url: str) -> None:
"""Fetch current price and store in history."""
response = httpx.get(
API_URL,
params={"url": url},
headers={"X-API-Key": API_KEY},
timeout=15.0,
)
data = response.json()
if not data["success"]:
print(f"Error for {url}: {data['error']['message']}")
return
price_data = data["data"]
conn.execute(
"INSERT INTO price_history (product_id, price, currency, in_stock) VALUES (?, ?, ?, ?)",
(product_id, price_data["price"], price_data["currency"], int(price_data["in_stock"])),
)
conn.commit()
print(f" {url}: {price_data['currency']} {price_data['price']}")
def check_all_prices(db_path: str = "prices.db") -> None:
"""Check prices for all tracked products."""
conn = init_db(db_path)
products = conn.execute("SELECT id, url FROM products").fetchall()
print(f"Checking {len(products)} products at {datetime.now().isoformat()}")
for product_id, url in products:
fetch_and_store(conn, product_id, url)
conn.close()With data accumulating, you can answer questions like "what's the lowest price this product has ever been?" or "how has the price changed over the last 30 days?" These are simple SQL queries against the price_history table.
def get_price_history(conn: sqlite3.Connection, url: str, days: int = 30) -> list:
"""Get price history for a product over the last N days."""
return conn.execute("""
SELECT ph.price, ph.currency, ph.in_stock, ph.checked_at
FROM price_history ph
JOIN products p ON p.id = ph.product_id
WHERE p.url = ?
AND ph.checked_at >= datetime('now', ?)
ORDER BY ph.checked_at DESC
""", (url, f"-{days} days")).fetchall()
def get_lowest_price(conn: sqlite3.Connection, url: str) -> tuple | None:
"""Get the all-time lowest price for a product."""
return conn.execute("""
SELECT MIN(ph.price), ph.currency, ph.checked_at
FROM price_history ph
JOIN products p ON p.id = ph.product_id
WHERE p.url = ? AND ph.in_stock = 1
""", (url,)).fetchone()
def add_product(conn: sqlite3.Connection, url: str, name: str = "") -> int:
"""Add a product to track."""
conn.execute(
"INSERT OR IGNORE INTO products (url, name) VALUES (?, ?)",
(url, name),
)
conn.commit()
row = conn.execute("SELECT id FROM products WHERE url = ?", (url,)).fetchone()
return row[0]Set up a cron job to run the price checker on a schedule. Every 6 hours is a good starting point — frequent enough to catch most price changes, conservative enough on credits.
Make sure your cron environment has the PRICEFETCH_API_KEY variable set. The easiest way is to put it in a `.env` file and source it in the cron command.
# Add to crontab (crontab -e)
# Check prices every 6 hours
0 */6 * * * cd /path/to/project && source .env && python tracker.py
# Or every hour for more granular tracking
0 * * * * cd /path/to/project && source .env && python tracker.pyHere's the full tracker.py that ties everything together. Run it manually to test, then set up cron for automation. Add products by calling add_product() with Amazon URLs.
This script is deliberately simple — no frameworks, no complex dependencies. It does one thing well: track prices over time. You can build dashboards, alerts, and reports on top of the SQLite data later.
#!/usr/bin/env python3
"""Amazon price tracker using PriceFetch API."""
import sys
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "add":
# Add a new product: python tracker.py add <url> [name]
conn = init_db()
url = sys.argv[2]
name = sys.argv[3] if len(sys.argv) > 3 else ""
pid = add_product(conn, url, name)
print(f"Added product {pid}: {url}")
conn.close()
elif len(sys.argv) > 1 and sys.argv[1] == "history":
# Show history: python tracker.py history <url>
conn = init_db()
url = sys.argv[2]
history = get_price_history(conn, url)
for price, currency, in_stock, checked_at in history:
stock = "in stock" if in_stock else "out of stock"
print(f" {checked_at}: {currency} {price:.2f} ({stock})")
lowest = get_lowest_price(conn, url)
if lowest and lowest[0]:
print(f"\n All-time low: {lowest[1]} {lowest[0]:.2f} on {lowest[2]}")
conn.close()
else:
# Default: check all prices
check_all_prices()Sign up in 30 seconds. No credit card required. One credit per successful API call.