Python Developer Career in the Travel Industry

# FastAPI — a booking endpoint that reflects real
# travel backend patterns.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx

app = FastAPI()

class BookingRequest(BaseModel):
    flight_id: str
    passenger: dict
    payment_method: str

@app.post("/api/bookings")
async def create_booking(req: BookingRequest):
    # 1. Re-validate availability with the supplier
    #    (availability can change between search and book)
    async with httpx.AsyncClient(timeout=10) as client:
        check = await client.get(
            f"https://supplier.example.com/flights/{req.flight_id}/availability"
        )
    if check.status_code != 200 or not check.json()["available"]:
        raise HTTPException(409, "Flight no longer available")

    # 2. Create the booking with the supplier
    async with httpx.AsyncClient(timeout=20) as client:
        resp = await client.post(
            "https://supplier.example.com/bookings",
            json={
                "flight_id": req.flight_id,
                "passenger": req.passenger,
            },
        )
    if resp.status_code != 201:
        raise HTTPException(502, "Supplier booking failed")

    supplier_booking = resp.json()

    # 3. Persist internally, charge, notify — usually
    #    done via a background worker, not in this request
    return {
        "booking_id": supplier_booking["id"],
        "status": "confirmed",
        "pnr": supplier_booking["pnr"],
    }

# What this proves in an interview:
# - You understand that availability must be rechecked at booking time
# - You handle supplier failures explicitly (not just "assume success")
# - You keep the request fast — heavy work goes to a worker
# - You know why travel booking is a distributed transaction,
#   not a single DB insert

# This is the exact shape of real travel backends.

# Async — query multiple suppliers in parallel.
# Travel search is all about parallel I/O.

import asyncio
import httpx

SUPPLIERS = [
    "https://supplier-a.example.com/flights",
    "https://supplier-b.example.com/flights",
    "https://supplier-c.example.com/flights",
]

async def fetch_supplier(client, url, params):
    try:
        resp = await client.get(url, params=params, timeout=5)
        if resp.status_code == 200:
            return resp.json()["flights"]
        return []
    except httpx.TimeoutException:
        # A slow supplier should not break the whole search
        return []
    except Exception:
        return []

async def search_all_suppliers(origin, destination, date):
    params = {
        "origin": origin,
        "destination": destination,
        "date": date,
    }
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            *[fetch_supplier(client, url, params) for url in SUPPLIERS]
        )
    # Merge + dedupe + sort by price
    all_flights = [f for batch in results for f in batch]
    return sorted(all_flights, key=lambda f: f["price"])

# What this proves in an interview:
# - You understand async I/O for parallel supplier queries
# - You handle supplier failures gracefully (one bad supplier
#   should not break the whole search)
# - You know why travel search is not a loop of sequential HTTP calls
# - You can reason about timeouts and fallbacks

# This is the exact pattern behind metasearch engines
# like Skyscanner, Kayak, and Google Flights.

# Celery — background task for confirmation email.
# Travel systems use background workers heavily because
# booking flows involve many steps that must not block
# the response.

from celery import Celery
import smtplib

app = Celery("travel", broker="redis://localhost:6379/0")

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_confirmation_email(self, booking_id, email, itinerary):
    try:
        with smtplib.SMTP("smtp.example.com", 587) as server:
            server.starttls()
            server.login("user", "pass")
            server.sendmail(
                "bookings@company.com",
                email,
                render_itinerary(itinerary),
            )
    except Exception as exc:
        # Retry with backoff — email servers fail sometimes
        raise self.retry(exc=exc)

def render_itinerary(itinerary):
    # Plain text for now — production would use Jinja2
    return f"Your booking {itinerary['pnr']} is confirmed."

# What this proves in an interview:
# - You know booking flows must not block on email/SMS
# - You know background tasks need retries (emails fail)
# - You understand why Celery + Redis is the default pattern
# - You can explain the difference between sync and async
#   parts of a booking

# In travel, the user must see a confirmation within 1-2
# seconds. Everything else (email, invoice, loyalty,
# analytics) is offloaded to workers.

# Getting this wrong is the #1 cause of slow travel apps.