Google shut down the QPX Express API — the only official programmatic access to real-time flight pricing — in April 2018. They replaced it with nothing. No successor API. No paid commercial tier. No developer programme. Just a web interface and a policy that prevents automated access.

The airfare data is all there. Google Flights aggregates live inventory from hundreds of airlines and provides one of the most accurate, current fare data sources available. But there is no official way for a developer to access it programmatically — which means every flight price alert system, every travel intelligence tool, and every corporate travel management dashboard that needs live fare data is built on scraping Google Flights rather than calling an official endpoint.

ScrapeBadger's Google Flights API handles everything between your query and Google's data: anti-bot bypass, session management, and structured JSON output. You pass an origin, destination, and date. You get back airlines, prices, schedules, layovers, carbon emissions, a price_insights block with Google's own price level assessment, and a booking token that deep-links directly to the Google Flights booking flow.

This guide builds a complete flight price alert system: watchlist configuration, multi-route monitoring, price history tracking, target price detection, and Slack alerts — plus a calendar scanner that checks multiple dates around your target window to find the cheapest travel days.

What the API Returns (Including the Null Fields)

Before writing a line of code, understand what the Google Flights API actually returns versus what varies or is sometimes absent. The ScrapeBadger response on a real LAX→JFK search showed a price spread of $534–$1,462 across results on the same day. The price field is always populated. Several other fields are not.

Consistently present:

• airline — carrier name

• price — total fare in the requested currency

• total_duration_minutes — end-to-end journey time

• carbon_emissions_grams — CO₂ estimate for the flight

• booking_token — deep-link to Google Flights booking page

Sometimes null (handle gracefully):

• departure_time and arrival_time — present on most legs, null on some

• flight_number — not always returned

• aircraft — plane model, often absent

• legroom — seat pitch, present when Google has the data

• price_insights.lowest_price, price_insights.price_level, price_insights.typical_price_range — available on popular routes, null on thinner routes or specific date combinations

The price_insights block when populated is particularly useful. price_level is Google's own classification: "low", "typical", or "high" — their assessment of whether the current fares are below, at, or above historical norms for the route. typical_price_range gives the normal fare range as [min, max]. These two fields together answer the question "is this a good deal?" without requiring you to build your own price history baseline.

Setup

bash

pip install httpx sqlalchemy aiohttp python-dotenv

env

SCRAPEBADGER_API_KEY=your_key_here
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/xxx/yyy/zzz

Step 1: Data Models

python

# models.py
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime


@dataclass
class FlightLeg:
    """A single flight segment within a journey."""
    departure_airport: str
    arrival_airport: str
    duration_minutes: int
    airline: str
    departure_time: Optional[str] = None   # Sometimes null
    arrival_time: Optional[str] = None     # Sometimes null
    flight_number: Optional[str] = None   # Sometimes null
    aircraft: Optional[str] = None        # Sometimes null
    legroom: Optional[str] = None         # Sometimes null
    travel_class: Optional[str] = None    # Sometimes null
    extensions: list = field(default_factory=list)  # Amenities: Wi-Fi, power, etc.


@dataclass
class Layover:
    airport: str
    duration_minutes: int


@dataclass
class FlightOption:
    """A complete flight option (possibly multi-leg) returned by the API."""
    price: float
    currency: str
    total_duration_minutes: int
    legs: list[FlightLeg]
    layovers: list[Layover]
    airline: str                           # Primary carrier
    carbon_emissions_grams: Optional[int]
    carbon_diff_vs_typical: Optional[int]  # Negative = greener than average
    booking_token: Optional[str]
    is_nonstop: bool = False

    @property
    def stops(self) -> int:
        return len(self.layovers)

    @property
    def hours(self) -> float:
        return round(self.total_duration_minutes / 60, 1)

    @property
    def co2_kg(self) -> Optional[float]:
        if self.carbon_emissions_grams:
            return round(self.carbon_emissions_grams / 1000, 1)
        return None

    @property
    def departure_display(self) -> str:
        if self.legs and self.legs[0].departure_time:
            return self.legs[0].departure_time
        return "N/A"

    @property
    def arrival_display(self) -> str:
        if self.legs and self.legs[-1].arrival_time:
            return self.legs[-1].arrival_time
        return "N/A"


@dataclass
class SearchResult:
    """Complete result for a single route + date query."""
    departure_id: str
    arrival_id: str
    outbound_date: str
    return_date: Optional[str]
    trip_type: str
    currency: str
    best_flights: list[FlightOption]

    # Price insights (Google's own assessment — may be None)
    price_level: Optional[str] = None           # "low" | "typical" | "high"
    typical_price_range: Optional[list] = None  # [min, max]
    lowest_price: Optional[float] = None

    searched_at: str = ""

    @property
    def cheapest_price(self) -> Optional[float]:
        if not self.best_flights:
            return None
        return min(f.price for f in self.best_flights)

    @property
    def cheapest_flight(self) -> Optional[FlightOption]:
        if not self.best_flights:
            return None
        return min(self.best_flights, key=lambda f: f.price)

    @property
    def is_good_deal(self) -> Optional[bool]:
        """True if Google's price_level indicates below-typical fares."""
        return self.price_level == "low" if self.price_level else None

Step 2: The ScrapeBadger Google Flights Collection Layer

python

# flights_collector.py
import httpx
import asyncio
import os
from typing import Optional
from datetime import datetime
from models import FlightLeg, Layover, FlightOption, SearchResult

API_KEY = os.environ["SCRAPEBADGER_API_KEY"]
BASE_URL = "https://scrapebadger.com/v1"
HEADERS = {"x-api-key": API_KEY}


def _parse_leg(raw: dict) -> FlightLeg:
    """Parse a single flight leg from the API response."""
    return FlightLeg(
        departure_airport=raw.get("departure_airport", ""),
        arrival_airport=raw.get("arrival_airport", ""),
        duration_minutes=int(raw.get("duration_minutes") or raw.get("duration") or 0),
        airline=raw.get("airline", ""),
        departure_time=raw.get("departure_time"),      # May be None
        arrival_time=raw.get("arrival_time"),          # May be None
        flight_number=raw.get("flight_number"),        # May be None
        aircraft=raw.get("aircraft"),                  # May be None
        legroom=raw.get("legroom"),                    # May be None
        travel_class=raw.get("travel_class"),          # May be None
        extensions=raw.get("extensions") or [],
    )


def _parse_flight_option(raw: dict) -> Optional[FlightOption]:
    """Parse a flight option (one row in best_flights)."""
    try:
        price = raw.get("price")
        if price is None:
            return None

        legs_raw = raw.get("legs") or raw.get("flights") or []
        legs = [_parse_leg(l) for l in legs_raw if isinstance(l, dict)]

        layovers_raw = raw.get("layovers") or []
        layovers = [
            Layover(
                airport=l.get("id") or l.get("name") or "",
                duration_minutes=int(l.get("duration") or l.get("duration_minutes") or 0),
            )
            for l in layovers_raw
        ]

        # Carbon emissions
        co2 = raw.get("carbon_emissions_grams")
        co2_diff = raw.get("carbon_emissions_diff_typical")

        # If carbon data is nested (some response formats)
        co2_block = raw.get("carbon_emissions") or {}
        if isinstance(co2_block, dict) and co2 is None:
            co2 = co2_block.get("this_flight")
            typical = co2_block.get("typical_for_this_route")
            if co2 is not None and typical is not None:
                co2_diff = co2 - typical

        # Primary airline from first leg or top-level field
        airline = raw.get("airline") or (legs[0].airline if legs else "Unknown")

        return FlightOption(
            price=float(price),
            currency=raw.get("currency", "USD"),
            total_duration_minutes=int(
                raw.get("total_duration_minutes") or raw.get("total_duration") or 0
            ),
            legs=legs,
            layovers=layovers,
            airline=airline,
            carbon_emissions_grams=int(co2) if co2 is not None else None,
            carbon_diff_vs_typical=int(co2_diff) if co2_diff is not None else None,
            booking_token=raw.get("booking_token"),
            is_nonstop=len(layovers) == 0,
        )
    except Exception as e:
        print(f"  Parse error on flight option: {e}")
        return None


async def search_flights(
    client: httpx.AsyncClient,
    departure_id: str,
    arrival_id: str,
    outbound_date: str,
    return_date: str = None,
    trip_type: str = "one_way",
    adults: int = 1,
    travel_class: str = "economy",
    stops: str = "any",             # "any" | "nonstop" | "one_stop"
    currency: str = "USD",
    gl: str = "us",
    hl: str = "en",
) -> Optional[SearchResult]:
    """
    Search Google Flights via ScrapeBadger.
    Returns None on failed or empty response (not charged).
    
    Cost: 7 credits per successful call.
    Response time: ~3 seconds.
    """
    params = {
        "departure_id": departure_id.upper(),
        "arrival_id": arrival_id.upper(),
        "outbound_date": outbound_date,
        "trip_type": trip_type,
        "adults": adults,
        "travel_class": travel_class,
        "stops": stops,
        "currency": currency,
        "gl": gl,
        "hl": hl,
    }
    if return_date and trip_type == "round_trip":
        params["return_date"] = return_date

    try:
        response = await client.get(
            f"{BASE_URL}/google/flights/search",
            params=params,
            timeout=30.0,
        )
        response.raise_for_status()
        data = response.json()

        # Parse flight options
        best_raw = data.get("best_flights") or []
        best_flights = [
            f for f in (_parse_flight_option(r) for r in best_raw)
            if f is not None
        ]

        if not best_flights:
            print(f"  No results: {departure_id}→{arrival_id} {outbound_date}")
            return None

        # Parse price insights (may be null)
        insights = data.get("price_insights") or {}
        price_level = insights.get("price_level")   # "low" | "typical" | "high" | None
        typical_range = insights.get("typical_price_range")
        lowest = insights.get("lowest_price")

        result = SearchResult(
            departure_id=departure_id.upper(),
            arrival_id=arrival_id.upper(),
            outbound_date=outbound_date,
            return_date=return_date,
            trip_type=trip_type,
            currency=currency,
            best_flights=best_flights,
            price_level=price_level,
            typical_price_range=typical_range,
            lowest_price=float(lowest) if lowest else None,
            searched_at=datetime.utcnow().isoformat(),
        )

        cheapest = result.cheapest_price
        deal_tag = f" 🟢 [{price_level.upper()}]" if price_level else ""
        print(
            f"  {departure_id}→{arrival_id} {outbound_date}: "
            f"{len(best_flights)} options, cheapest {currency} {cheapest:.0f}{deal_tag}"
        )
        return result

    except httpx.HTTPStatusError as e:
        print(f"  HTTP {e.response.status_code} on {departure_id}→{arrival_id}: {e}")
        return None
    except Exception as e:
        print(f"  Error searching {departure_id}→{arrival_id}: {e}")
        return None

Step 3: Price History Database

python

# database.py
from sqlalchemy import (
    create_engine, Column, Integer, Float, String,
    Boolean, DateTime, Text, Index,
)
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from datetime import datetime, timedelta
from typing import Optional

class Base(DeclarativeBase):
    pass


class PriceObservation(Base):
    """One price observation per route per check cycle."""
    __tablename__ = "price_observations"

    id = Column(Integer, primary_key=True)
    route = Column(String, nullable=False, index=True)  # "JFK-LHR"
    outbound_date = Column(String, nullable=False)
    return_date = Column(String, nullable=True)
    trip_type = Column(String, default="one_way")
    currency = Column(String, default="USD")

    # Price data
    cheapest_price = Column(Float, nullable=False)
    cheapest_airline = Column(String)
    cheapest_duration_min = Column(Integer)
    cheapest_stops = Column(Integer)
    cheapest_is_nonstop = Column(Boolean, default=False)
    booking_token = Column(Text, nullable=True)

    # Price insights
    price_level = Column(String, nullable=True)         # "low" | "typical" | "high"
    typical_range_min = Column(Float, nullable=True)
    typical_range_max = Column(Float, nullable=True)

    # Carbon
    cheapest_co2_kg = Column(Float, nullable=True)

    observed_at = Column(DateTime, default=datetime.utcnow, index=True)

    __table_args__ = (
        Index("ix_route_date_time", "route", "outbound_date", "observed_at"),
    )


class AlertLog(Base):
    """Every alert that has been sent — prevents duplicate alerts."""
    __tablename__ = "alert_log"

    id = Column(Integer, primary_key=True)
    route = Column(String, nullable=False)
    outbound_date = Column(String)
    alert_type = Column(String)  # "target_reached" | "price_drop" | "good_deal"
    price = Column(Float)
    prev_price = Column(Float, nullable=True)
    price_level = Column(String, nullable=True)
    airline = Column(String, nullable=True)
    alerted_at = Column(DateTime, default=datetime.utcnow)


engine = create_engine("sqlite:///flight_alerts.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)


class PriceDB:

    def save(self, route: str, result) -> None:
        cheapest = result.cheapest_flight
        if not cheapest:
            return
        insights = result
        typical = result.typical_price_range or [None, None]
        with Session() as session:
            obs = PriceObservation(
                route=route,
                outbound_date=result.outbound_date,
                return_date=result.return_date,
                trip_type=result.trip_type,
                currency=result.currency,
                cheapest_price=cheapest.price,
                cheapest_airline=cheapest.airline,
                cheapest_duration_min=cheapest.total_duration_minutes,
                cheapest_stops=cheapest.stops,
                cheapest_is_nonstop=cheapest.is_nonstop,
                booking_token=cheapest.booking_token,
                price_level=result.price_level,
                typical_range_min=typical[0] if typical else None,
                typical_range_max=typical[1] if len(typical) > 1 else None,
                cheapest_co2_kg=cheapest.co2_kg,
                observed_at=datetime.utcnow(),
            )
            session.add(obs)
            session.commit()

    def get_last_price(self, route: str, outbound_date: str) -> Optional[float]:
        with Session() as session:
            row = (
                session.query(PriceObservation.cheapest_price)
                .filter(
                    PriceObservation.route == route,
                    PriceObservation.outbound_date == outbound_date,
                )
                .order_by(PriceObservation.observed_at.desc())
                .first()
            )
            return row[0] if row else None

    def already_alerted(
        self, route: str, outbound_date: str, alert_type: str, within_hours: int = 24
    ) -> bool:
        """Prevent duplicate alerts for the same event within a time window."""
        cutoff = datetime.utcnow() - timedelta(hours=within_hours)
        with Session() as session:
            exists = (
                session.query(AlertLog)
                .filter(
                    AlertLog.route == route,
                    AlertLog.outbound_date == outbound_date,
                    AlertLog.alert_type == alert_type,
                    AlertLog.alerted_at >= cutoff,
                )
                .first()
            )
            return exists is not None

    def log_alert(
        self,
        route: str,
        outbound_date: str,
        alert_type: str,
        price: float,
        prev_price: float = None,
        price_level: str = None,
        airline: str = None,
    ) -> None:
        with Session() as session:
            session.add(AlertLog(
                route=route,
                outbound_date=outbound_date,
                alert_type=alert_type,
                price=price,
                prev_price=prev_price,
                price_level=price_level,
                airline=airline,
            ))
            session.commit()

    def get_price_history(
        self, route: str, outbound_date: str, days: int = 30
    ) -> list[dict]:
        cutoff = datetime.utcnow() - timedelta(days=days)
        with Session() as session:
            rows = (
                session.query(
                    PriceObservation.observed_at,
                    PriceObservation.cheapest_price,
                    PriceObservation.cheapest_airline,
                    PriceObservation.price_level,
                )
                .filter(
                    PriceObservation.route == route,
                    PriceObservation.outbound_date == outbound_date,
                    PriceObservation.observed_at >= cutoff,
                )
                .order_by(PriceObservation.observed_at.asc())
                .all()
            )
            return [
                {
                    "at": r.observed_at.isoformat(),
                    "price": r.cheapest_price,
                    "airline": r.cheapest_airline,
                    "level": r.price_level,
                }
                for r in rows
            ]

Step 4: Alert Detection and Delivery

python

# alerts.py
import aiohttp
import os
from database import PriceDB

db = PriceDB()


async def send_slack(text: str) -> None:
    webhook = os.environ.get("SLACK_WEBHOOK_URL")
    if not webhook:
        print(f"[ALERT] {text}")
        return
    async with aiohttp.ClientSession() as session:
        try:
            await session.post(
                webhook,
                json={"text": text},
                timeout=aiohttp.ClientTimeout(total=10),
            )
        except Exception as e:
            print(f"Slack error: {e}")
            print(f"[ALERT] {text}")


async def check_and_alert(
    result,
    route: str,
    alert_config: dict,
) -> bool:
    """
    Evaluate whether a search result warrants an alert.

    alert_config keys:
      target_price:        Alert when cheapest falls below this absolute value
      drop_threshold_pct:  Alert when price drops >= this % vs last observation
      alert_on_good_deal:  Alert when Google's price_level = "low"
    """
    if not result or not result.cheapest_flight:
        return False

    cheapest = result.cheapest_flight
    current_price = cheapest.price
    outbound_date = result.outbound_date
    alerted = False

    last_price = db.get_last_price(route, outbound_date)

    # ── Alert 1: Target price reached ────────────────────────────────────────
    target = alert_config.get("target_price")
    if target and current_price <= target:
        if not db.already_alerted(route, outbound_date, "target_reached"):
            co2_note = f" | {cheapest.co2_kg}kg CO₂" if cheapest.co2_kg else ""
            booking_note = f"\n🔗 {cheapest.booking_token}" if cheapest.booking_token else ""
            await send_slack(
                f"🎯 *Target price reached!*\n"
                f"Route: *{route}* | Date: {outbound_date}\n"
                f"Price: *{result.currency} {current_price:.0f}* (target was {result.currency} {target:.0f})\n"
                f"Airline: {cheapest.airline} | {cheapest.hours}h | {cheapest.stops} stop(s)"
                f"{co2_note}{booking_note}"
            )
            db.log_alert(
                route, outbound_date, "target_reached",
                current_price, last_price, result.price_level, cheapest.airline
            )
            alerted = True

    # ── Alert 2: Price drop vs last observation ───────────────────────────────
    drop_threshold = alert_config.get("drop_threshold_pct", 8.0)
    if last_price and last_price > 0:
        change_pct = (current_price - last_price) / last_price * 100
        if change_pct <= -drop_threshold:
            if not db.already_alerted(route, outbound_date, "price_drop"):
                await send_slack(
                    f"📉 *Price drop detected*\n"
                    f"Route: *{route}* | Date: {outbound_date}\n"
                    f"Was: {result.currency} {last_price:.0f} → Now: {result.currency} {current_price:.0f} "
                    f"({change_pct:.1f}%)\n"
                    f"Airline: {cheapest.airline} | {cheapest.hours}h"
                )
                db.log_alert(
                    route, outbound_date, "price_drop",
                    current_price, last_price, result.price_level, cheapest.airline
                )
                alerted = True

    # ── Alert 3: Google's own "low" price assessment ─────────────────────────
    if alert_config.get("alert_on_good_deal") and result.price_level == "low":
        if not db.already_alerted(route, outbound_date, "good_deal"):
            range_note = ""
            if result.typical_price_range:
                lo, hi = result.typical_price_range
                range_note = f"\nTypical range: {result.currency} {lo:.0f}–{hi:.0f}"
            await send_slack(
                f"💚 *Google says: below-typical fares*\n"
                f"Route: *{route}* | Date: {outbound_date}\n"
                f"Current best: {result.currency} {current_price:.0f} | {cheapest.airline}"
                f"{range_note}"
            )
            db.log_alert(
                route, outbound_date, "good_deal",
                current_price, last_price, "low", cheapest.airline
            )
            alerted = True

    return alerted

Step 5: Calendar Scanner — Finding the Cheapest Travel Window

The single most useful feature beyond basic price tracking is scanning a range of dates around your target travel period to find which specific days are cheapest. Airfare variance across a 7–14 day window is often 20–50%. The calendar scan costs one API call per date checked — at 7 credits each, scanning 14 dates costs 98 credits.

python

# calendar_scanner.py
import asyncio
import httpx
import os
from datetime import date, timedelta
from flights_collector import search_flights

HEADERS = {"x-api-key": os.environ["SCRAPEBADGER_API_KEY"]}


async def scan_date_window(
    departure_id: str,
    arrival_id: str,
    center_date: str,          # YYYY-MM-DD — your target travel date
    days_before: int = 3,
    days_after: int = 4,
    trip_type: str = "one_way",
    return_offset_days: int = 7,  # For round trips: return N days after outbound
    currency: str = "USD",
    max_concurrent: int = 3,
) -> list[dict]:
    """
    Scan a window of dates around a target to find the cheapest travel days.

    For a 3+4 window around 2026-08-15:
    Checks: Aug 12, 13, 14, 15, 16, 17, 18, 19 — 8 API calls (56 credits)

    Returns dates sorted by price ascending.
    """
    center = date.fromisoformat(center_date)
    dates_to_check = [
        (center + timedelta(days=i)).isoformat()
        for i in range(-days_before, days_after + 1)
    ]

    semaphore = asyncio.Semaphore(max_concurrent)
    results = []

    async with httpx.AsyncClient(headers=HEADERS) as client:

        async def check_date(d: str) -> dict:
            async with semaphore:
                await asyncio.sleep(0.3)
                ret = None
                if trip_type == "round_trip":
                    ret = (date.fromisoformat(d) + timedelta(days=return_offset_days)).isoformat()
                result = await search_flights(
                    client,
                    departure_id=departure_id,
                    arrival_id=arrival_id,
                    outbound_date=d,
                    return_date=ret,
                    trip_type=trip_type,
                    currency=currency,
                )
                if result and result.cheapest_price:
                    cheapest = result.cheapest_flight
                    return {
                        "date": d,
                        "price": result.cheapest_price,
                        "currency": currency,
                        "airline": cheapest.airline if cheapest else "",
                        "duration_h": cheapest.hours if cheapest else 0,
                        "stops": cheapest.stops if cheapest else 0,
                        "price_level": result.price_level or "unknown",
                        "co2_kg": cheapest.co2_kg if cheapest else None,
                    }
                return {"date": d, "price": None, "error": "no results"}

        scan_results = await asyncio.gather(*[check_date(d) for d in dates_to_check])

    results = [r for r in scan_results if r.get("price") is not None]
    results.sort(key=lambda r: r["price"])

    # Print summary
    print(f"\n📅 Calendar scan: {departure_id}→{arrival_id} | {days_before+days_after+1} dates")
    print(f"{'Date':<14} {'Price':>8} {'Airline':<20} {'Duration':>8} {'Stops':>6} {'Level':<10}")
    print("-" * 70)
    for r in results:
        level_emoji = {"low": "🟢", "typical": "🟡", "high": "🔴"}.get(r["price_level"], "⚪")
        print(
            f"{r['date']:<14} "
            f"{r['currency']} {r['price']:>5.0f}   "
            f"{r['airline']:<20} "
            f"{r['duration_h']:>5.1f}h   "
            f"{r['stops']:>5}   "
            f"{level_emoji} {r['price_level']}"
        )

    if results:
        best = results[0]
        worst = results[-1]
        savings = worst["price"] - best["price"]
        print(f"\nBest date: {best['date']} at {best['currency']} {best['price']:.0f}")
        print(f"Worst date: {worst['date']} at {worst['currency']} {worst['price']:.0f}")
        print(f"Potential saving by choosing the best date: {best['currency']} {savings:.0f}")

    return results

Step 6: The Complete Pipeline

python

# main.py
import asyncio
import httpx
import os
from flights_collector import search_flights
from alerts import check_and_alert
from database import PriceDB
from calendar_scanner import scan_date_window

HEADERS = {"x-api-key": os.environ["SCRAPEBADGER_API_KEY"]}
db = PriceDB()


# ── Watchlist ─────────────────────────────────────────────────────────────────
WATCHLIST = [
    {
        "route": "LHR-JFK",
        "departure_id": "LHR",
        "arrival_id": "JFK",
        "outbound_date": "2026-09-12",
        "trip_type": "one_way",
        "currency": "GBP",
        "alert_config": {
            "target_price": 320,            # Alert when under £320
            "drop_threshold_pct": 8.0,      # Alert on 8%+ price drop
            "alert_on_good_deal": True,     # Alert when Google says "low"
        },
    },
    {
        "route": "FCO-BCN",
        "departure_id": "FCO",
        "arrival_id": "BCN",
        "outbound_date": "2026-08-20",
        "return_date": "2026-08-27",
        "trip_type": "round_trip",
        "currency": "EUR",
        "alert_config": {
            "target_price": 180,
            "drop_threshold_pct": 10.0,
            "alert_on_good_deal": True,
        },
    },
    {
        "route": "SYD-SIN",
        "departure_id": "SYD",
        "arrival_id": "SIN",
        "outbound_date": "2026-10-05",
        "trip_type": "one_way",
        "currency": "AUD",
        "alert_config": {
            "target_price": 450,
            "drop_threshold_pct": 12.0,
            "alert_on_good_deal": False,
        },
    },
]


async def run_monitoring_cycle() -> None:
    """Run one monitoring cycle across the full watchlist."""
    print(f"\n{'='*55}")
    print(f"Flight price monitoring — {len(WATCHLIST)} routes")
    print(f"{'='*55}")

    alerts_sent = 0

    async with httpx.AsyncClient(headers=HEADERS) as client:
        for watch in WATCHLIST:
            result = await search_flights(
                client,
                departure_id=watch["departure_id"],
                arrival_id=watch["arrival_id"],
                outbound_date=watch["outbound_date"],
                return_date=watch.get("return_date"),
                trip_type=watch.get("trip_type", "one_way"),
                currency=watch.get("currency", "USD"),
            )

            if result:
                db.save(watch["route"], result)
                alerted = await check_and_alert(
                    result, watch["route"], watch["alert_config"]
                )
                if alerted:
                    alerts_sent += 1

            await asyncio.sleep(1.0)

    print(f"\nCycle complete. {alerts_sent} alerts sent.")


async def main():
    import sys
    command = sys.argv[1] if len(sys.argv) > 1 else "monitor"

    if command == "monitor":
        # Continuous monitoring every 4 hours
        print("Starting flight price monitor...")
        while True:
            await run_monitoring_cycle()
            print("\nNext check in 4 hours...")
            await asyncio.sleep(4 * 60 * 60)

    elif command == "scan":
        # Calendar scan: cheapest dates for a route
        # Usage: python main.py scan LHR JFK 2026-09-15
        dep = sys.argv[2] if len(sys.argv) > 2 else "LHR"
        arr = sys.argv[3] if len(sys.argv) > 3 else "JFK"
        center = sys.argv[4] if len(sys.argv) > 4 else "2026-09-15"
        await scan_date_window(
            departure_id=dep,
            arrival_id=arr,
            center_date=center,
            days_before=3,
            days_after=4,
        )

    elif command == "once":
        # Single monitoring run without the loop
        await run_monitoring_cycle()


if __name__ == "__main__":
    asyncio.run(main())

Running the calendar scanner:

bash

python main.py scan LHR JFK 2026-09-15

  LHR→JFK 2026-09-12: 6 options, cheapest GBP 398 🟡 [TYPICAL]
  LHR→JFK 2026-09-13: 5 options, cheapest GBP 312 🟢 [LOW]
  LHR→JFK 2026-09-14: 7 options, cheapest GBP 344 🟡 [TYPICAL]
  LHR→JFK 2026-09-15: 6 options, cheapest GBP 371 🟡 [TYPICAL]
  LHR→JFK 2026-09-16: 4 options, cheapest GBP 429 🔴 [HIGH]
  LHR→JFK 2026-09-17: 6 options, cheapest GBP 361 🟡 [TYPICAL]
  LHR→JFK 2026-09-18: 5 options, cheapest GBP 388 🟡 [TYPICAL]
  LHR→JFK 2026-09-19: 6 options, cheapest GBP 402 🟡 [TYPICAL]

📅 Calendar scan: LHR→JFK | 8 dates
Date           Price   Airline              Duration  Stops  Level     
----------------------------------------------------------------------
2026-09-13    GBP 312  British Airways       7.1h       0   🟢 low
2026-09-14    GBP 344  Virgin Atlantic       7.2h       0   🟡 typical
2026-09-17    GBP 361  American              7.3h       1   🟡 typical
...

Best date: 2026-09-13 at GBP 312
Worst date: 2026-09-16 at GBP 429
Potential saving by choosing the best date: GBP 117

What the price_level Field Is Actually Worth

Google's price_level assessment is the most overlooked field in the response. It requires no baseline-building on your part — Google has already computed historical norms for the route and date, and is telling you where the current fare sits relative to those norms.

For a personal price alert system, alert_on_good_deal: True combined with a reasonable target price creates a two-signal alert that is more reliable than either signal alone. A fare that is simultaneously below your target price AND classified by Google as "low" is a well-validated opportunity. A fare below your target but classified as "typical" may reflect a target that was set too high.

For a corporate travel management application, the price_level combined with typical_price_range creates a policy compliance signal: was the fare booked within the typical range for that route and date? The answer requires no internal benchmark database — it comes directly from the API response.

The price_history field within price_insights — when it is populated, which varies by route — provides a historical price time series from Google's own data, separate from the local database you build with this pipeline. On popular routes where it appears, it supplements your observation database during the early days of monitoring before you have accumulated enough observations of your own.

The full endpoint documentation is at docs.scrapebadger.com. Free trial at scrapebadger.com — 1,000 credits, no credit card.

FAQ

Why is there no official Google Flights API?

Google shut down the QPX Express API — the only official programmatic access to flight pricing — in April 2018. No replacement has been released. Google's stated reason was that the API was primarily used by businesses that already had direct GDS (Global Distribution System) relationships with airlines. The web interface at Google Flights provides the same data to users but has no documented, stable API for developer access. The ScrapeBadger Google Flights API provides the same data through infrastructure that handles Google's anti-bot protection.

How often should I check prices for a specific route?

Airfare pricing on most routes updates multiple times per day. The practical monitoring cadence depends on how close the departure date is. For flights 60+ days out, checking every 4–6 hours is sufficient — fare classes change during business hours and rarely shift significantly overnight. For flights within 2 weeks of departure, more frequent checks (every 1–2 hours) make sense because last-minute yield management pricing can produce significant drops or spikes in short windows.

What does departure_time: null mean in the response?

Some flight result cards in Google Flights do not include granular departure and arrival times in the structured response data — typically for itineraries where Google is showing a range of options rather than a specific flight selection. The price is always present and accurate. The departure_time null case is handled in the data model with a graceful "N/A" fallback. For alert purposes, the price is the critical field; departure time is contextual detail.

Can I monitor multi-city itineraries?

The ScrapeBadger Google Flights endpoint currently supports one-way and round-trip searches. Multi-city search (trip_type: "multi_city") is a planned addition. For now, multi-city monitoring can be approximated by monitoring the individual legs separately and summing the cheapest options — useful for approximate budget planning though not for booking a single multi-city ticket.

How much does continuous monitoring cost at scale?

At 7 credits per call and a 4-hour monitoring cycle, one route costs 42 credits per day. A watchlist of 10 routes costs 420 credits per day — approximately 12,600 credits per month. ScrapeBadger's $49/month plan includes enough credits for continuous monitoring of roughly 4 routes per day; the $99/month plan covers 10+ routes. Calendar scanning adds credits at 7 per date checked — a 14-day window scan costs 98 credits, worth running once when setting up monitoring for a new route.