Building a production-grade Python integration with a flight data API. Session management, error handling, retry logic, rate limits, caching, async patterns and testing — with complete code examples for AirLabs and the Python requests and httpx libraries.
Building an aviation integration in Python is usually a two-part problem. The first part is knowing which endpoint answers which question — how to identify a flight from an IATA code, a tail number, an ADS-B hex or a route, and which AirLabs endpoint to call at each phase of the flight lifecycle. The second part is making the integration robust — handling errors, retries, rate limits, caching and concurrency in a way that survives production traffic.
Our How to Track a Flight — Developer Guide covers the first part in detail: the five ways to identify a flight, the endpoint decision tree, the lifecycle table and the edge cases around codeshares, diversions and timezones. This guide is about the second part — the Python-specific production patterns that turn a prototype into an integration you can rely on. It assumes you have already decided which endpoints your application will call and focuses on how to call them well.
Every code example uses the AirLabs API as the endpoint and is complete enough to run once you have an API key. The patterns translate directly between HTTP client libraries — the examples show requests for synchronous code and httpx for asynchronous code, but the same concepts apply to any client. Code examples target Python 3.8 and higher; the async examples require Python 3.7 or higher for asyncio.run.
"A working Python integration is not the code that makes the first API call succeed. It is the code that keeps succeeding when the network flickers, when the rate limit is hit, when the upstream returns an unexpected shape, when the traffic spike arrives at 3am. Those are the four moments the production patterns exist for."
Before the production concerns, it helps to see how little code is actually required to make the API work at all. The absolute minimum is three lines against requests:
import requests
response = requests.get(
"https://airlabs.co/api/v9/flight",
params={"flight_iata": "BA117", "api_key": "YOUR_API_KEY"},
)
print(response.json()["response"])This works for a prototype. It is also brittle in every way that matters — no session reuse, no error handling, no retry logic, no rate-limit awareness, hard-coded credentials in source. Each of the sections below replaces one of those problems.
The first upgrade from the three-line prototype is a shared requests.Session. A session reuses underlying TCP connections across calls, which materially improves performance when the integration makes more than a handful of requests. Attaching the API key to the session's default parameters also removes it from every individual call:
import os
import requests
API_KEY = os.environ["AIRLABS_API_KEY"]
BASE_URL = "https://airlabs.co/api/v9"
session = requests.Session()
session.params.update({"api_key": API_KEY})
# Every subsequent call reuses the connection and auto-includes the key
response = session.get(f"{BASE_URL}/flight", params={"flight_iata": "BA117"})For high-throughput services, tune the session's connection pool through a HTTPAdapter:
from requests.adapters import HTTPAdapter
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=100)
session.mount("https://", adapter)This lets the session hold up to 100 concurrent connections to the AirLabs host, which matters for services processing many flights in parallel.
Every successful AirLabs response wraps its payload under a response key; errors go under an error key. Building a small helper that respects this convention keeps the calling code clean:
def call_airlabs(endpoint: str, **params):
"""Call an AirLabs endpoint. Return the response payload, or raise."""
resp = session.get(f"{BASE_URL}/{endpoint}", params=params)
resp.raise_for_status()
body = resp.json()
if "error" in body:
err = body["error"]
raise AirLabsError(
code=err.get("code"),
message=err.get("message", "Unknown error"),
raw=err,
)
return body.get("response")
class AirLabsError(Exception):
def __init__(self, code, message, raw=None):
super().__init__(f"AirLabs error {code}: {message}")
self.code = code
self.message = message
self.raw = rawNow every caller can either use the returned payload or catch AirLabsError and inspect it. This is the shape of the wrapper the rest of the examples assume.
Not every error should be retried. A rate-limit response, a 5xx from the upstream or a network timeout are transient — the same request tried a few seconds later will probably succeed. An invalid API key, a bad parameter or a 4xx-that-is-not-429 is permanent — retrying will not help. Classifying errors correctly is what stops retry loops from making problems worse:
from requests.exceptions import ConnectionError, Timeout
RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
RATE_LIMIT_MESSAGE_MARKERS = {"minute_limit_exceeded", "rate_limit_exceeded"}
def is_transient(exc: Exception) -> bool:
if isinstance(exc, (ConnectionError, Timeout)):
return True
if isinstance(exc, AirLabsError):
if any(m in str(exc.message) for m in RATE_LIMIT_MESSAGE_MARKERS):
return True
if isinstance(exc, requests.HTTPError):
if exc.response is not None and exc.response.status_code in RETRYABLE_STATUS_CODES:
return True
return FalseEvery retry loop in the integration should consult is_transient() rather than catching bare Exception — the latter masks bugs.
The classic pattern for retrying transient errors is exponential backoff with jitter — each retry waits longer than the last, with a random component to spread out concurrent retries and avoid thundering-herd effects:
import time
import random
def call_with_retries(endpoint: str, retries: int = 4, **params):
"""Call an endpoint with exponential backoff on transient errors."""
last_exc = None
for attempt in range(retries):
try:
return call_airlabs(endpoint, **params)
except Exception as exc:
last_exc = exc
if not is_transient(exc):
raise
if attempt == retries - 1:
break
# Exponential backoff with jitter: 1s, 2s, 4s, 8s ± 25%
wait = (2 ** attempt) * (0.75 + 0.5 * random.random())
time.sleep(wait)
raise last_excFour retries is a reasonable default for most integrations. Fewer if the calling code has its own retry logic higher up; more only if the acceptable maximum latency is very long.
The single most common transient error in a production integration is the rate limit. AirLabs enforces per-minute request budgets according to the subscription plan, and exceeding the budget returns an error whose message contains a rate-limit marker. Rather than retrying blindly, wait until the next minute boundary:
def wait_for_next_minute():
"""Sleep until the start of the next minute."""
now = time.time()
time.sleep(60 - (now % 60) + 0.1)
def call_with_rate_limit_handling(endpoint: str, **params):
for attempt in range(5):
try:
return call_airlabs(endpoint, **params)
except AirLabsError as exc:
if any(m in str(exc.message) for m in RATE_LIMIT_MESSAGE_MARKERS):
wait_for_next_minute()
continue
raise
raise RuntimeError(f"Rate-limited after 5 attempts on {endpoint}")The long-term answer to persistent rate limiting is not more aggressive retries but a caching layer that reduces the request rate, or a plan upgrade for higher throughput. But this pattern prevents graceful degradation from becoming abrupt failure during traffic spikes.
Reference data — airport codes, airline names, aircraft types, city and country data — changes on weekly-to-monthly timescales, not per request. Caching it is the single largest reduction you can make in your API request volume. For most Python applications, functools.lru_cache is enough:
from functools import lru_cache
@lru_cache(maxsize=5000)
def airport_info(iata: str) -> dict:
result = call_with_retries("airports", iata_code=iata)
return result[0] if result else {}
@lru_cache(maxsize=2000)
def airline_info(iata: str) -> dict:
result = call_with_retries("airlines", iata_code=iata)
return result[0] if result else {}For applications that survive across process restarts, use a persistent cache — Redis, SQLite or a file-backed store. A day-long TTL for reference data is a safe starting point; refresh in the background rather than on demand.
Every AirLabs endpoint accepts a _fields parameter that limits the response to the specific fields you consume. On high-frequency endpoints and large aggregations, this is a meaningful performance win — smaller responses, faster JSON parsing, less memory pressure:
def departures_lean(airport_iata: str) -> list[dict]:
"""Return only the fields needed for a summary display."""
return call_with_retries(
"schedules",
dep_iata=airport_iata,
_fields="flight_iata,arr_iata,dep_estimated,dep_gate,status,delayed",
)Applied across an integration processing thousands of flights, field selection can reduce total payload size by an order of magnitude.
For applications making many concurrent requests — a dashboard querying status for hundreds of flights, a data pipeline processing many airports — an async client scales better than a thread pool. The httpx library provides an async-capable HTTP client with an API similar to requests:
import asyncio
import httpx
async def get_flight_status(client: httpx.AsyncClient, flight_iata: str) -> dict:
resp = await client.get(
"flight",
params={"flight_iata": flight_iata},
)
resp.raise_for_status()
body = resp.json()
if "error" in body:
raise AirLabsError(
code=body["error"].get("code"),
message=body["error"].get("message"),
)
return body.get("response") or {}
async def get_many_flight_statuses(flight_iatas: list[str]) -> list[dict]:
async with httpx.AsyncClient(
base_url=BASE_URL,
params={"api_key": API_KEY},
limits=httpx.Limits(max_connections=50),
timeout=10.0,
) as client:
tasks = [get_flight_status(client, f) for f in flight_iatas]
return await asyncio.gather(*tasks, return_exceptions=True)
# Usage
flights = asyncio.run(get_many_flight_statuses(["BA117", "AA100", "LH400"]))asyncio.gather with return_exceptions=True prevents a single failed request from cancelling the rest. For any batch of more than a dozen flights, this async pattern is materially faster than sequential synchronous calls.
An integration that is not testable is an integration that will break silently. For unit tests, unittest.mock and fixture responses cover most cases without hitting the real API:
from unittest.mock import patch, MagicMock
# Assumes AirLabsClient is defined (see the Complete Production Client section below)
SAMPLE_FLIGHT = {
"flight_iata": "BA117",
"dep_iata": "LHR",
"arr_iata": "JFK",
"dep_time": "2026-07-22 10:00",
"status": "active",
"delayed": 47,
}
def test_flight_status_lookup():
client = AirLabsClient(api_key="test_key")
with patch.object(client.session, "get") as mock_get:
mock_response = MagicMock()
mock_response.json.return_value = {"response": SAMPLE_FLIGHT}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
result = client.flight("BA117")
assert result["flight_iata"] == "BA117"
assert result["delayed"] == 47For integration tests against the real API, use a distinct test API key and cap the test rate to stay within the free plan's limits. Recording and replaying real responses with vcr.py gives fast, deterministic tests that reflect real API behaviour.
There is a small community-maintained Python package on PyPI, iata_codes, that provides a client for IATA code lookups against the AirLabs API. Its scope is narrow — IATA airport and city code queries — and it does not cover the real-time flights, schedules, delays, alert or fleet endpoints. It is not maintained by AirLabs directly, and its most recent activity predates the current API version.
Most production integrations use the requests-based patterns in this guide directly against the API. The reason is that the API surface is small enough — a dozen endpoints, all following the same request/response convention — that a thin wrapper you own is usually easier to maintain than a general-purpose SDK you depend on. The examples in this guide are effectively that thin wrapper, written to be copied and adapted.
Combining the patterns above into a single class produces a working starting point that most integrations can build on:
import os
import time
import random
import requests
from functools import lru_cache
from typing import Optional
from requests.exceptions import ConnectionError, Timeout
class AirLabsError(Exception):
def __init__(self, code, message, raw=None):
super().__init__(f"AirLabs error {code}: {message}")
self.code = code
self.message = message
self.raw = raw
class AirLabsClient:
BASE_URL = "https://airlabs.co/api/v9"
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
RATE_LIMIT_MARKERS = {"minute_limit_exceeded", "rate_limit_exceeded"}
def __init__(self, api_key: Optional[str] = None, timeout: float = 10.0):
self.api_key = api_key or os.environ["AIRLABS_API_KEY"]
self.timeout = timeout
self.session = requests.Session()
self.session.params.update({"api_key": self.api_key})
def _is_transient(self, exc: Exception) -> bool:
if isinstance(exc, (ConnectionError, Timeout)):
return True
if isinstance(exc, requests.HTTPError):
if exc.response is not None and exc.response.status_code in self.RETRYABLE_STATUS:
return True
if isinstance(exc, AirLabsError):
return any(m in str(exc.message) for m in self.RATE_LIMIT_MARKERS)
return False
def call(self, endpoint: str, retries: int = 4, **params):
last_exc = None
for attempt in range(retries):
try:
resp = self.session.get(
f"{self.BASE_URL}/{endpoint}",
params=params,
timeout=self.timeout,
)
resp.raise_for_status()
body = resp.json()
if "error" in body:
err = body["error"]
raise AirLabsError(err.get("code"), err.get("message"))
return body.get("response")
except Exception as exc:
last_exc = exc
if not self._is_transient(exc) or attempt == retries - 1:
raise
wait = (2 ** attempt) * (0.75 + 0.5 * random.random())
time.sleep(wait)
raise last_exc
@lru_cache(maxsize=5000)
def airport(self, iata: str) -> dict:
result = self.call("airports", iata_code=iata)
return result[0] if result else {}
def flight(self, flight_iata: str) -> dict:
result = self.call("flight", flight_iata=flight_iata)
return result if isinstance(result, dict) else {}
def departures(self, iata: str, fields: Optional[str] = None) -> list:
params = {"dep_iata": iata}
if fields:
params["_fields"] = fields
return self.call("schedules", **params) or []
# Usage example
if __name__ == "__main__":
client = AirLabsClient()
# Reference lookup (cached)
lhr = client.airport("LHR")
print(f"Airport: {lhr.get('name')}")
# Flight status
flight = client.flight("BA117")
if flight:
print(f"Flight {flight.get('flight_iata')}: {flight.get('status')}")
# Airport departures with field selection
deps = client.departures("JFK", fields="flight_iata,arr_iata,delayed")
print(f"JFK departures: {len(deps)} flights")At around 80 lines including the error class and a small usage example, this is a complete client — configuration, session management, error classification, retries with backoff, rate-limit handling, caching. Real applications extend it further with structured logging, metrics, per-endpoint timeout tuning and integration with the application's own async or task-queue framework.
requests is infinite. In production, always pass an explicit timeout parameter — 10 seconds is a reasonable default for most flight data queries.httpx with asyncio will use fewer resources than threads or processes.If you are integrating aviation data into a Python application, the AirLabs API and the patterns in this guide give you a starting point that scales from prototype to production. Start with the free flight API plan, adapt the client above to your application's own logging and configuration conventions, and let the API handle the aviation data while your code handles the product logic on top.
Our Developer API allows you to create a custom experience for your users and increase the value of your product:
_fields for lean, targeted responsesYou can try it right now without any obligation! Get a free flight API plan and see for yourself that we have exactly the data you need!
If you need more information, don't hesitate to contact us. We are always happy to chat with our customers and are sure to find a customized solution for each request.
Explore AirLabs, or create an account instantly and start using API.
Get FREE API Key