APIs, Databases & External Boundaries

What you will learn

1. Why calls to other systems can fail, and what to do.
2. Two ways to read pages: offset and cursor.
3. How to follow a cursor without trusting the API.
4. How to retry, and how to obey a rate limit.
5. How to read and write a database safely.
6. How to make a re-run safe.

An API is a way for one program to ask another program for data. You send a request. The other program sends back a response.

Most APIs answer in JSON. JSON is text in a fixed shape, with curly braces and square brackets. Python reads it with json.loads and gives you dicts and lists.

This page cannot call a real API. So we use a tiny fake one. It returns the shop orders from the earlier chapters, two orders per page, as JSON text.

import json

ORDERS = [
    {"order_id": 101, "customer_id": 1, "status": "paid", "amount": 40.0},
    {"order_id": 102, "customer_id": 1, "status": "paid", "amount": 25.0},
    {"order_id": 103, "customer_id": 2, "status": "refunded", "amount": 60.0},
    {"order_id": 104, "customer_id": 3, "status": "paid", "amount": 15.5},
    {"order_id": 105, "customer_id": 3, "status": "pending", "amount": None},
    {"order_id": 106, "customer_id": 4, "status": "paid", "amount": 80.0},
]

def fake_api(cursor=None, limit=2):
    last = 0 if cursor is None else int(cursor)
    page = [o for o in ORDERS if o["order_id"] > last][:limit]
    has_more = bool(page) and page[-1] != ORDERS[-1]
    next_cursor = str(page[-1]["order_id"]) if has_more else None
    return json.dumps({"orders": page, "next_cursor": next_cursor})

page[-1] is the last order on this page. If it is not the last order of all, there are more pages, so has_more is True. An empty page has no last order, so we check bool(page) first.

Call the function with limit=1 and look at the raw response. A real API sends text like this over the network:

text = fake_api(limit=1)
print(text)
print(type(text).__name__)

Now parse it. json.loads turns the text into a dict. A JSON null would become None, the same as order 105 in our table.

data = json.loads(fake_api())

print(type(data).__name__)
print(data["next_cursor"])
for order in data["orders"]:
    print(order["order_id"], order["amount"])

Code inside your program is fast and in your control. A call to another system is not. Three things can go wrong:

1. The call is slow, or never answers.
2. The call fails. The other system may be down for a moment.
3. The other system refuses you, because you ask too often.

Every response has a status code. It is a number that tells you what happened. The body is the data part of the response.

A code starting with 2 means it worked. A 4 means your request has a problem. A 5 means the server failed.

Two 4xx codes are exceptions: 408 and 429 are worth a retry.

A temporary error is also called transient. You met the idea in the validation chapter.

A 429 or a 503 is transient, but a 404 is not. This function sorts the codes:

def is_retryable(status):
    return status in (408, 429) or 500 <= status < 600

print(is_retryable(503), is_retryable(429), is_retryable(404), is_retryable(200))

Always set a timeout. A timeout is the longest time you wait for an answer. Without one, a stuck server can freeze your job for a very long time.

With requests, you write it like this. We do not run it here, because it needs the network:

import requests

response = requests.get("https://api.example.com/orders", timeout=10)
response.raise_for_status()
data = response.json()

A slow call ends with an error. Real requests raises requests.exceptions.Timeout. Our later examples use the built-in TimeoutError, so they run without the library.

An API does not send a million orders in one response. It sends a page, and you ask for the next page. There are two common ways to ask.

With offset, you say how many rows to skip: "skip 2, give me 2". Here a plain list stands in for an API that lists the newest order first. Between your two calls, a new order, 107, arrives at the front:

orders = [106, 105, 104, 103, 102, 101]

def offset_api(offset, limit=2):
    return orders[offset:offset + limit]

print(offset_api(0))
orders.insert(0, 107)
print(offset_api(2))

If a row had been deleted instead, a row would be skipped. A cursor avoids both problems.

The API gives you a bookmark: the last id you got. You send it back to get the next page.

Here is the same story with a cursor. The next page is the orders older than the bookmark:

orders = [106, 105, 104, 103, 102, 101]

def cursor_api(cursor=None, limit=2):
    older = [n for n in orders if cursor is None or n < cursor]
    return older[:limit]

print(cursor_api())
orders.insert(0, 107)
print(cursor_api(cursor=105))

Our fake_api uses a cursor too, but it lists the oldest order first. This loop follows its cursor to the end:

cursor = None
while True:
    data = json.loads(fake_api(cursor))
    ids = [o["order_id"] for o in data["orders"]]
    print("page:", ids, "next:", data["next_cursor"])
    cursor = data["next_cursor"]
    if cursor is None:
        break

The loop in the last section trusts the API. A broken API can send the same cursor again and again. Then your loop never ends.

So add two guards. First, stop when a cursor repeats. Second, stop after a maximum number of pages:

def follow_cursor(api, max_pages=10):
    seen = set()
    cursor = None
    for _ in range(max_pages):
        data = json.loads(api(cursor))
        yield from data["orders"]
        cursor = data["next_cursor"]
        if cursor is None:
            return
        if cursor in seen:
            raise RuntimeError(f"repeated cursor {cursor}")
        seen.add(cursor)
    raise RuntimeError(f"more than {max_pages} pages")

orders = follow_cursor(fake_api)
print([o["order_id"] for o in orders])

• for _ in range(max_pages) repeats at most max_pages times. The name _ means we do not use the number.
• yield from data["orders"] gives each order of the page, one by one. It is a short form of a for loop with yield.
• On the good API, the cursor becomes None on page 3, and return ends the generator.

Now a broken API. It always says the next cursor is 102:

def stuck_api(cursor=None):
    return json.dumps({"orders": ORDERS[:2], "next_cursor": "102"})

try:
    for order in follow_cursor(stuck_api):
        print("got", order["order_id"])
except RuntimeError as error:
    print("stopped:", error)

A second broken API always sends a new cursor and never ends. The page cap stops it. Here the cap is 3:

def endless_api(cursor=None):
    n = 0 if cursor is None else int(cursor)
    return json.dumps({"orders": ORDERS[:1], "next_cursor": str(n + 1)})

try:
    rows = list(follow_cursor(endless_api, max_pages=3))
except RuntimeError as error:
    print("stopped:", error)

You met backoff in the validation chapter: wait longer after each failure, up to a maximum. That chapter also mentions jitter, a small random extra wait.

Here we add what is new for an API. The status code decides if we retry.

A timeout counts as a failure. And the server may tell us how long to wait.

import time

def call_with_retry(call, max_tries=4, sleep=time.sleep):
    for attempt in range(1, max_tries + 1):
        retry_after = None
        try:
            response = call()
        except TimeoutError:
            status = 408
        else:
            status = response["status"]
            if status == 200:
                return response["body"]
            retry_after = response.get("retry_after")
        if not is_retryable(status) or attempt == max_tries:
            raise RuntimeError(f"gave up: status {status}, try {attempt}")
        wait = min(30, retry_after or 2 ** (attempt - 1))
        sleep(wait)

Read the new parts one at a time:

• attempt counts the tries: 1, 2, 3, 4. range(1, max_tries + 1) makes these numbers.
• The else after try runs only when the try block did not fail.
• The except catches only TimeoutError, so a bug in your code is not retried. A timeout counts as status 408.
• x or y gives x if x has a value. If x is None, it gives y. So the wait is retry_after, or the backoff if there is none.
• 2 ** n means 2 to the power n. 2 ** 0 is 1, 2 ** 1 is 2, and 2 ** 2 is 4.
• min(30, ...) cuts the wait to at most 30 seconds. A longer wait would only freeze the job.

To test it, we need calls that fail in a planned way. The helper planned gives one answer from a list each time you call it. If the answer is an error, it raises the error:

def planned(answers):
    answers = iter(answers)
    def call():
        answer = next(answers)
        if isinstance(answer, Exception):
            raise answer
        return answer
    return call

ok = {"status": 200, "body": "orders"}
waits = []
call = planned([TimeoutError(), {"status": 503}, ok])
print(call_with_retry(call, sleep=waits.append))
print(waits)

A 404 is not transient, so the function must not retry it. And if the server never recovers, the function must stop:

waits = []
try:
    call_with_retry(planned([{"status": 404}, ok]), sleep=waits.append)
except RuntimeError as error:
    print(error, waits)

waits = []
down = {"status": 503}
try:
    call_with_retry(planned([down, down, down, down]), sleep=waits.append)
except RuntimeError as error:
    print(error, waits)

A rate limit is a cap on how many requests you may send in a period. If you go over it, the API answers 429 (too many requests).

A good API also tells you how long to wait. It sends a Retry-After header, often a number of seconds.

A header is extra information that comes with the response. We store it as retry_after.

Our call_with_retry already uses it: wait = min(30, retry_after or 2 ** (attempt - 1)). Here the API says wait 5 seconds:

waits = []
too_many = {"status": 429, "retry_after": 5}
print(call_with_retry(planned([too_many, ok]), sleep=waits.append))
print(waits)

Better still, do not hit the limit. Add a short pause between pages, for example time.sleep(0.5). A short pause is cheap.

A database is another system you call. Here we use sqlite3, a small database that comes with Python. The name :memory: means it lives in memory and vanishes at the end.

A connection is your open line to the database. A table holds rows. Each column has a type: INTEGER is a whole number, TEXT is text, and REAL is a decimal number.

order_id INTEGER PRIMARY KEY means each order_id may appear only once. executemany runs one statement for many rows. Each ? is a placeholder, and the next section explains it:

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute(
    "CREATE TABLE orders ("
    "order_id INTEGER PRIMARY KEY, "
    "customer_id INTEGER, status TEXT, amount REAL)"
)
order_tuples = [
    (o["order_id"], o["customer_id"], o["status"], o["amount"])
    for o in ORDERS
]
conn.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", order_tuples)

print(order_tuples[0])
print(conn.execute("SELECT COUNT(*) FROM orders").fetchone())

Now read the paid orders. Put a ? where the value goes. Give the value as a second argument, in a tuple.

The examples reuse conn from the last section:

result = conn.execute(
    "SELECT order_id, amount FROM orders WHERE status = ?",
    ("paid",),
)
print(result.fetchall())

Never build SQL text with an f-string. Say a user types this as a status. Look at what the f-string builds:

user_input = "paid' OR '1'='1"

query = f"SELECT order_id FROM orders WHERE status = '{user_input}'"
print(query)
print(conn.execute(query).fetchall())

safe = conn.execute(
    "SELECT order_id FROM orders WHERE status = ?", (user_input,)
)
print(safe.fetchall())

A big table can have millions of rows. Do not call fetchall() on it. Use fetchmany(2) to read a few rows at a time:

result = conn.execute("SELECT order_id FROM orders ORDER BY order_id")
while True:
    chunk = result.fetchmany(2)
    if not chunk:
        break
    print(chunk)

Each call to the database is a trip. Fewer trips mean less waiting. So write a batch of rows with one executemany call.

Also wrap each batch in with conn. When the block ends, the batch is committed: saved for good. If an error happens inside, the batch is rolled back: undone.

We copy the orders into a second table, four rows at a time:

conn.execute(
    "CREATE TABLE orders_copy ("
    "order_id INTEGER PRIMARY KEY, "
    "customer_id INTEGER, status TEXT, amount REAL)"
)

source = conn.execute("SELECT * FROM orders ORDER BY order_id")
while True:
    batch = source.fetchmany(4)
    if not batch:
        break
    with conn:
        conn.executemany(
            "INSERT INTO orders_copy VALUES (?, ?, ?, ?)", batch
        )
    print("wrote", len(batch))

print(conn.execute("SELECT COUNT(*) FROM orders_copy").fetchone()[0])

Now a batch that fails halfway. The first row is new. The second row repeats order 101, so the database refuses it:

try:
    with conn:
        conn.execute("INSERT INTO orders_copy VALUES (107, 5, 'paid', 10.0)")
        conn.execute("INSERT INTO orders_copy VALUES (101, 1, 'paid', 40.0)")
except sqlite3.IntegrityError as error:
    print("error:", error)

print(conn.execute("SELECT COUNT(*) FROM orders_copy").fetchone()[0])

A connection holds resources, such as memory and a place in the database. If you forget to close it, a long job can run out of connections.

In earlier chapters, with open(...) closed a file. For sqlite3, with conn does something else.

It commits or rolls back the work. It does not close the connection.

To close it, use closing from the contextlib module. Look at what happens after the block ends:

from contextlib import closing

with closing(sqlite3.connect(":memory:")) as db:
    print(db.execute("SELECT 1").fetchone())

try:
    db.execute("SELECT 1")
except sqlite3.ProgrammingError as error:
    print("error:", error)

Use both when you write: closing for the connection, and with db for each batch. This is the usual shape of a small database job:

with closing(sqlite3.connect(":memory:")) as db:
    db.execute("CREATE TABLE visits (n INTEGER)")
    with db:
        db.executemany("INSERT INTO visits VALUES (?)", [(1,), (2,)])
    print(db.execute("SELECT COUNT(*) FROM visits").fetchone()[0])

A job fails, or you run it twice by mistake. A job is idempotent when a second run gives the same result as the first.

The records chapter used keys to remove repeats. The same idea works here.

Say the same page is fetched twice. Put each row in a dict, with order_id as the key:

first = json.loads(fake_api())["orders"]
again = json.loads(fake_api())["orders"]
rows = first + again
print(len(rows))

by_id = {}
for order in rows:
    by_id[order["order_id"]] = order
print(len(by_id), list(by_id))

A database write has the same problem. A plain INSERT of orders that already exist fails.

An upsert writes a new row, or replaces the row that has the same key. INSERT OR REPLACE is the form SQLite uses.

Other databases use other words. Run the whole copy again:

try:
    with conn:
        conn.executemany(
            "INSERT INTO orders_copy VALUES (?, ?, ?, ?)", order_tuples
        )
except sqlite3.IntegrityError as error:
    print("error:", error)

with conn:
    conn.executemany(
        "INSERT OR REPLACE INTO orders_copy VALUES (?, ?, ?, ?)", order_tuples
    )
print(conn.execute("SELECT COUNT(*) FROM orders_copy").fetchone()[0])

A job that runs often should read only what is new. After each run, save the highest order_id you stored. This saved number is the watermark.

Say the last run stopped at order 104. So saved is 104. Ask for the ids above it, then move the watermark with max:

saved = 104
new = conn.execute(
    "SELECT order_id, status FROM orders WHERE order_id > ? ORDER BY order_id",
    (saved,),
).fetchall()

if new:
    saved = max(order_id for order_id, status in new)

print(new)
print(saved)

You learned to read pages, to retry, and to read and write a database safely. Each exercise opens in the Python workspace with its own data.

Before you run each exercise, guess one thing that could go wrong. Then check your guess.