Python Requests for API Calls with Sessions, Timeouts and Error Handling

The curl lesson showed what an API exchange looks like by hand. This one moves the same work into Python with the requests library. By the end you’ll have a small client function that sets a timeout on every call, reuses one connection, and turns every way a call can fail into one clear error your code can act on.

You need Python 3.10 or newer, which is the minimum for the current requests release (2.34.2). Everything runs against a test server on your own machine, so no account or API key is involved.

Set up a project folder with a virtual environment and pin the version:

python3 -m venv .venv
source .venv/bin/activate
pip install requests==2.34.2

On Windows the activate script is .venv\Scripts\activate. Pinning matters more than it looks. A client that works today and breaks after an unplanned upgrade is a miserable thing to debug.

A server that misbehaves on purpose

Real APIs fail in several distinct ways, and you want to see each one before you meet it in production. Save this as server.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import time

ITEMS = {"1": {"id": 1, "name": "tea"}, "2": {"id": 2, "name": "coffee"}}

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/whoami":
            self.send_json(200, {"client_port": self.client_address[1],
                                 "user_agent": self.headers.get("User-Agent"),
                                 "api_key": self.headers.get("X-Api-Key")})
        elif self.path == "/slow":
            time.sleep(3)
            self.send_json(200, {"done": True})
        elif self.path == "/maintenance":
            body = b"<html><body>Back soon</body></html>"
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        elif self.path.startswith("/items/"):
            item = ITEMS.get(self.path.rsplit("/", 1)[1])
            if item:
                self.send_json(200, item)
            else:
                self.send_json(404, {"error": "item not found"})
        else:
            self.send_json(404, {"error": "no such route"})

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        try:
            data = json.loads(self.rfile.read(length))
        except ValueError:
            self.send_json(400, {"error": "body must be JSON"})
            return
        if not data.get("name"):
            self.send_json(422, {"error": "name is required"})
            return
        self.send_json(201, {"id": 3, "name": data["name"]})

    def log_message(self, *args):
        pass

ThreadingHTTPServer(("127.0.0.1", 18767), Handler).serve_forever()

Start it with python server.py and leave that terminal open. It has a normal item route, a missing item, an endpoint that takes three seconds, and a “maintenance page” that answers 200 with HTML where your code expects JSON. That last one catches more people than any other.

First calls, and why a 404 doesn’t raise

In a second terminal (with the virtual environment active), run this:

import requests

r = requests.get("http://127.0.0.1:18767/items/1", timeout=5)
print(r.status_code)
print(r.headers["Content-Type"])
print(r.json())

r = requests.get("http://127.0.0.1:18767/items/99", timeout=5)
print(r.status_code, r.ok)
print(r.json())

Output:

200
application/json
{'id': 1, 'name': 'tea'}
404 False
{'error': 'item not found'}

Look at the second call. The server said 404 and requests handed you a response object without complaint. That’s by design: a 404 is a perfectly valid HTTP answer, so the library leaves the decision to you. Beginners often skip the check, call r.json(), and carry an error body through their program as if it were data.

r.raise_for_status() turns any 4xx or 5xx into an exception:

requests.exceptions.HTTPError: 404 Client Error: Not Found for url: http://127.0.0.1:18767/items/99

Call it right after every request unless you have a reason to handle a status yourself.

Sending JSON is just as short. Pass a dict to json= and requests serializes it and sets Content-Type: application/json for you:

import requests

with requests.Session() as s:
    r = s.post("http://127.0.0.1:18767/items", json={"name": "cocoa"}, timeout=5)
    print(r.status_code, r.json())
    print(r.request.headers["Content-Type"])

    r = s.post("http://127.0.0.1:18767/items", json={}, timeout=5)
    print(r.status_code, r.json())

201 {'id': 3, 'name': 'cocoa'}
application/json
422 {'error': 'name is required'}

Use json=, not data=, for JSON APIs. data= with a dict sends form encoding, and plenty of APIs reply with a confusing 400 when that happens.

Timeouts are off unless you set them

Requests has no default timeout. Leave timeout out and a server that accepts the connection and never answers will hang your script for as long as it likes. Every call in this lesson passes one, and so should yours.

The value can be a single number or a pair. A pair sets the connect timeout (how long to wait for the TCP connection) and the read timeout (how long to wait between bytes from the server) separately:

import requests

try:
    r = requests.get("http://127.0.0.1:18767/slow", timeout=(3.05, 1))
except requests.Timeout as exc:
    print("timed out:", type(exc).__name__)

timed out: ReadTimeout

It gave up after about one second instead of waiting the full three. The requests docs suggest a connect timeout slightly above a multiple of 3, since 3 seconds is the default TCP retransmission window, hence the odd-looking 3.05. Note that the read timeout is not a cap on the whole call. A server that sends one byte every half second will never trip a 1 second read timeout, so if you need a hard deadline you have to enforce it yourself.

Sessions reuse the connection and your headers

Each bare requests.get() opens a fresh connection. A Session keeps a pool and reuses connections to the same host, and it carries headers you set once. The /whoami route reports which client port the request came from, which makes the reuse visible:

import requests

for _ in range(2):
    r = requests.get("http://127.0.0.1:18767/whoami", timeout=5)
    print("no session:", r.json()["client_port"])

with requests.Session() as s:
    s.headers.update({"User-Agent": "apicourse-demo/1.0", "X-Api-Key": "test-key"})
    for _ in range(2):
        r = s.get("http://127.0.0.1:18767/whoami", timeout=5)
        print("session:", r.json())

no session: 56592
no session: 56596
session: {'client_port': 56608, 'user_agent': 'apicourse-demo/1.0', 'api_key': 'test-key'}
session: {'client_port': 56608, 'user_agent': 'apicourse-demo/1.0', 'api_key': 'test-key'}

Your port numbers will differ. What matters is that the two bare calls used two ports and the session used one. Against a remote HTTPS API that saves a TCP and TLS handshake on every call after the first, which adds up fast in a loop.

One catch: a session does not hold a timeout. You still pass timeout= on each call. And keep the real key in an environment variable, never in the source file.

Putting it together

Here’s the client function. It wraps every failure in one APIError type so the code calling it has a single thing to catch:

import requests

BASE = "http://127.0.0.1:18767"
TIMEOUT = (3.05, 2)


class APIError(Exception):
    pass


def get_json(session, path):
    url = BASE + path
    try:
        r = session.get(url, timeout=TIMEOUT)
        r.raise_for_status()
        return r.json()
    except requests.Timeout:
        raise APIError(f"{url} took too long") from None
    except requests.ConnectionError:
        raise APIError(f"could not reach {url}") from None
    except requests.HTTPError as exc:
        status = exc.response.status_code
        try:
            detail = exc.response.json().get("error", "")
        except requests.JSONDecodeError:
            detail = exc.response.text[:200]
        raise APIError(f"{status} from {url}: {detail}") from None
    except requests.JSONDecodeError:
        raise APIError(f"{url} did not return JSON") from None


if __name__ == "__main__":
    with requests.Session() as s:
        for path in ["/items/1", "/items/99", "/maintenance", "/slow"]:
            try:
                print(get_json(s, path))
            except APIError as exc:
                print("failed:", exc)

{'id': 1, 'name': 'tea'}
failed: 404 from http://127.0.0.1:18767/items/99: item not found
failed: http://127.0.0.1:18767/maintenance did not return JSON
failed: http://127.0.0.1:18767/slow took too long

Stop the server and run it again and every call ends in failed: could not reach .... Four kinds of failure, four readable messages.

The order of the except blocks is deliberate. ConnectTimeout inherits from both ConnectionError and Timeout, so if ConnectionError came first a connect timeout would be reported as “could not reach”. The HTML maintenance page came back as 200, passed raise_for_status(), and only failed at r.json(), which is why that check sits at the end. The error body is read for its error field because this API puts its message there; read the docs of the API you’re calling to see where it puts its own (the lesson on API error handling covers the common shapes).

What to try next

Change TIMEOUT to (3.05, 5) and watch /slow succeed. Add a post_json function using the same pattern and feed it the empty body that returned 422.

Resist the urge to wrap this in a retry loop just yet. Retrying a GET is usually safe, while retrying a POST can create the same record twice; the lesson on idempotency explains why, and retries with backoff are a lesson of their own.

Related