Flight Price Tracking API

import json
import os
import sqlite3
import sys
import time
from contextlib import closing
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path

import requests

API = "https://ignav.com/api"
API_KEY = os.environ["IGNAV_API_KEY"]
MAX_DAILY_REQUESTS = 20  # Includes booking lookups and every retry attempt.
DROP = Decimal("0.90")  # Alert on a drop of at least 10%.


def post(db, endpoint, payload):
    for attempt in range(3):
        day = datetime.now(timezone.utc).date().isoformat()
        # Reserve before sending, even if the response is lost. Shared DB = shared cap.
        with db:
            db.execute("BEGIN IMMEDIATE")
            db.execute("INSERT OR IGNORE INTO request_budget VALUES (?, 0)", (day,))
            used = db.execute(
                "SELECT attempts FROM request_budget WHERE day = ?", (day,)
            ).fetchone()[0]
            if used >= MAX_DAILY_REQUESTS:
                raise SystemExit("Daily request cap reached; stop until tomorrow UTC")
            db.execute(
                "UPDATE request_budget SET attempts = attempts + 1 WHERE day = ?",
                (day,),
            )
        try:
            response = requests.post(
                f"{API}/fares/{endpoint}",
                headers={"X-Api-Key": API_KEY}, json=payload,
                timeout=60, allow_redirects=False,
            )
        except (requests.ConnectionError, requests.Timeout):
            if attempt == 2:
                raise
        else:
            if response.status_code == 200:
                return response.json()
            if response.status_code in (401, 402, 403, 429):
                raise SystemExit(f"Account error HTTP {response.status_code}: {response.text}; stop polling")
            retryable = response.status_code in (500, 502, 503, 504) or (
                response.status_code == 424
                and response.json().get("error", {}).get("code")
                == "unable_to_complete_request"
            )
            if not retryable or attempt == 2:
                raise requests.HTTPError(f"HTTP {response.status_code}: {response.text}", response=response)
        time.sleep(2 ** attempt)


def check(db, watch):
    route = watch["route"]
    currency = watch["currency"]
    target = Decimal(str(watch["target_price"]))
    # Include return date and all search options, never the short-lived ignav_id.
    identity = json.dumps(route, sort_keys=True)
    endpoint = "round-trip" if route.get("return_date") else "one-way"
    result = post(db, endpoint, route)
    fares = [item for item in result["itineraries"]
             if item["price"]["status"] == "verified"
             and item["price"]["currency"] == currency
             and Decimal(str(item["price"]["amount"])).is_finite()
             and Decimal(str(item["price"]["amount"])) > 0]
    if not fares:
        print(f"No verified {currency} fares: {identity}")
        return
    best = min(fares, key=lambda item: Decimal(str(item["price"]["amount"])))
    price = Decimal(str(best["price"]["amount"]))
    with db:
        previous = db.execute(
            "SELECT price FROM watch_checks WHERE route = ? AND currency = ? "
            "ORDER BY id DESC LIMIT 1", (identity, currency),
        ).fetchone()
        db.execute(
            "INSERT INTO watch_checks (route, currency, price, checked_at) "
            "VALUES (?, ?, ?, ?)",
            (identity, currency, str(price), datetime.now(timezone.utc).isoformat()),
        )
    if price <= target or (previous and price <= Decimal(previous[0]) * DROP):
        print(f"Alert: {price:.2f} {currency} for {identity}")
        booking = post(db, "booking-links", {"ignav_id": best["ignav_id"]})
        # Print each offer's coverage: a round trip can require separate purchases.
        for option in booking["booking_options"]:
            for link in option["links"]:
                print(f"Book ({', '.join(option['legs'])}): {link['url']}")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("Usage: python tracker.py watchlist.json")
    watches = json.loads(Path(sys.argv[1]).read_text())
    with closing(sqlite3.connect("fares.db")) as db:
        db.execute("CREATE TABLE IF NOT EXISTS request_budget "
                   "(day TEXT PRIMARY KEY, attempts INTEGER NOT NULL)")
        db.execute("""CREATE TABLE IF NOT EXISTS watch_checks (
            id INTEGER PRIMARY KEY, route TEXT NOT NULL, currency TEXT NOT NULL,
            price TEXT NOT NULL, checked_at TEXT NOT NULL
        )""")
        for watch in watches:
            try:
                check(db, watch)
            except (requests.RequestException, ValueError, KeyError) as error:
                print(f"Check failed for {watch['route']}: {error}", file=sys.stderr)