Skip to content

Chapter 21: API Client

1. Calling External APIs

Every real application talks to something external. A payment processor. A weather service. A shipping carrier. A CRM. You write the HTTP call, parse the response, handle timeouts, retry on failure, add auth headers.

Tina4's Api class makes outbound HTTP requests from your server-side code. It returns a consistent response format, handles auth in one line, and supports SSL and timeout configuration, all with no extra libraries.


2. Basic Requests

python
from tina4_python.api import Api

api = Api()

# GET
result = api.get("https://api.example.com/products")

# POST with JSON body
result = api.post("https://api.example.com/orders", {
    "product_id": 42,
    "quantity": 3
})

# PUT
result = api.put("https://api.example.com/orders/101", {
    "status": "shipped"
})

# PATCH
result = api.patch("https://api.example.com/orders/101", {
    "tracking_number": "1Z999AA10123456784"
})

# DELETE
result = api.delete("https://api.example.com/orders/101")

Every method returns the same structure:

python
{
    "http_code": 200,
    "body": { ... },     # Parsed JSON, or raw string if not JSON
    "headers": { ... },  # Response headers
    "error": None        # None on success, error string on failure
}

3. Reading the Response

python
from tina4_python.api import Api

api = Api()
result = api.get("https://jsonplaceholder.typicode.com/posts/1")

if result["error"]:
    print(f"Request failed: {result['error']}")
else:
    print(f"Status: {result['http_code']}")
    print(f"Title: {result['body']['title']}")

Check result["error"] first. If it is None, the request succeeded. The HTTP status code is in result["http_code"]. The parsed response is in result["body"].

python
# In a route handler
from tina4_python.core.router import get
from tina4_python.api import Api

@get("/api/posts/{post_id}")
async def proxy_post(request, response):
    api = Api()
    post_id = request.params["post_id"]

    result = api.get(f"https://jsonplaceholder.typicode.com/posts/{post_id}")

    if result["error"]:
        return response({"error": result["error"]}, 502)

    if result["http_code"] == 404:
        return response({"error": "Post not found"}, 404)

    return response(result["body"])

4. Authentication Headers

Bearer Token

python
from tina4_python.api import Api

api = Api(bearer_token="eyJhbGciOiJSUzI1NiJ9...")

result = api.get("https://api.example.com/me")

The Authorization: Bearer <token> header is sent with every request made by this Api instance.

Basic Authentication

python
api = Api(username="api_user", password="secret123")

result = api.get("https://api.example.com/data")

Sends Authorization: Basic <base64(username:password)> with every request.

Custom Headers

python
api = Api(headers={
    "X-API-Key": "my-api-key-here",
    "X-Client-Version": "1.0.0",
    "Accept": "application/json"
})

result = api.get("https://api.example.com/data")

Custom headers are merged with any authentication headers and sent with every request.

Mixing Auth and Custom Headers

python
api = Api(
    bearer_token="eyJhbGciOiJSUzI1NiJ9...",
    headers={"X-Request-Source": "tina4-app"}
)

5. SSL Verification and Timeouts

python
# Disable SSL verification (dev only, never in production)
api = Api(verify_ssl=False)

# Set a 10-second timeout
api = Api(timeout=10)

# Both
api = Api(verify_ssl=False, timeout=5)

The default timeout is 30 seconds. The default SSL behaviour is to verify certificates (verify_ssl=True).

Never disable SSL verification in production. If an external API has a self-signed certificate, obtain the CA bundle and pass it explicitly, or ask the provider for their CA bundle.


6. Sending Query Parameters

Pass query parameters as a dictionary to the params argument:

python
api = Api()

result = api.get("https://api.example.com/products", params={
    "category": "Electronics",
    "page": 1,
    "limit": 20,
    "in_stock": True
})
# Requests: GET /products?category=Electronics&page=1&limit=20&in_stock=True

7. Real-World Patterns

Payment Gateway

python
import os
from tina4_python.api import Api

class PaymentGateway:
    def __init__(self):
        self.api = Api(
            bearer_token=os.environ["PAYMENT_API_KEY"],
            timeout=15
        )
        self.base = "https://api.payment-provider.com/v1"

    def charge(self, amount_cents, currency, card_token):
        result = self.api.post(f"{self.base}/charges", {
            "amount": amount_cents,
            "currency": currency,
            "source": card_token
        })

        if result["error"]:
            return {"success": False, "error": result["error"]}

        if result["http_code"] not in (200, 201):
            return {
                "success": False,
                "error": result["body"].get("message", "Payment declined")
            }

        return {
            "success": True,
            "charge_id": result["body"]["id"],
            "status": result["body"]["status"]
        }

    def refund(self, charge_id, amount_cents=None):
        body = {"charge": charge_id}
        if amount_cents:
            body["amount"] = amount_cents

        result = self.api.post(f"{self.base}/refunds", body)
        return result["body"]

Weather Service

python
import os
from tina4_python.api import Api

class WeatherService:
    def __init__(self):
        self.api = Api(timeout=5)
        self.api_key = os.environ["OPENWEATHER_API_KEY"]
        self.base = "https://api.openweathermap.org/data/2.5"

    def get_current(self, city):
        result = self.api.get(f"{self.base}/weather", params={
            "q": city,
            "appid": self.api_key,
            "units": "metric"
        })

        if result["error"] or result["http_code"] != 200:
            return None

        data = result["body"]
        return {
            "city": data["name"],
            "temp_c": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }

Usage in a route:

python
from tina4_python.core.router import get

weather = WeatherService()

@get("/api/weather/{city}")
async def get_weather(request, response):
    city = request.params["city"]
    data = weather.get_current(city)

    if data is None:
        return response({"error": "Weather data unavailable"}, 502)

    return response(data)

8. Retry Logic

For transient failures, add retry logic around Api calls:

python
import time
from tina4_python.api import Api
from tina4_python.debug import Log

def api_get_with_retry(url, max_retries=3, backoff=1.0, **kwargs):
    api = Api(**kwargs)
    last_error = None

    for attempt in range(1, max_retries + 1):
        result = api.get(url)

        if not result["error"] and result["http_code"] < 500:
            return result

        last_error = result["error"] or f"HTTP {result['http_code']}"
        Log.warning("API call failed, retrying", url=url, attempt=attempt, error=last_error)

        if attempt < max_retries:
            time.sleep(backoff * attempt)

    Log.error("API call failed after retries", url=url, error=last_error)
    return {"http_code": 0, "body": None, "headers": {}, "error": last_error}

9. Exercise: Weather Dashboard API

Build a route that fetches weather for multiple cities and returns a dashboard-ready response.

Requirements

  1. Create GET /api/dashboard/weather that:

    • Accepts a cities query parameter (comma-separated)
    • Calls a mock weather API for each city
    • Returns aggregated results
    • Returns 502 if any city fails
  2. Use proper error checking on every API call

  3. Add a 5-second timeout and Bearer token auth

Test with:

bash
curl "http://localhost:7146/api/dashboard/weather?cities=London,Berlin,Tokyo"

10. Solution

Create src/routes/weather_dashboard.py:

python
from tina4_python.core.router import get
from tina4_python.api import Api
from tina4_python.debug import Log

# Simulate a weather API with mock data
MOCK_WEATHER = {
    "london": {"city": "London", "temp_c": 12.3, "description": "Cloudy", "humidity": 78},
    "berlin": {"city": "Berlin", "temp_c": 8.1, "description": "Clear", "humidity": 55},
    "tokyo": {"city": "Tokyo", "temp_c": 18.7, "description": "Partly cloudy", "humidity": 65},
    "paris": {"city": "Paris", "temp_c": 14.0, "description": "Rainy", "humidity": 82},
}

def fetch_weather(city_name):
    # In production this would be a real API call:
    # api = Api(bearer_token=os.environ["WEATHER_API_KEY"], timeout=5)
    # return api.get(f"https://api.weather.com/v1/current?city={city_name}")

    key = city_name.lower().strip()
    data = MOCK_WEATHER.get(key)
    if data is None:
        return {"http_code": 404, "body": None, "headers": {}, "error": f"City not found: {city_name}"}
    return {"http_code": 200, "body": data, "headers": {}, "error": None}


@get("/api/dashboard/weather")
async def weather_dashboard(request, response):
    cities_param = request.params.get("cities", "")

    if not cities_param:
        return response({"error": "Provide at least one city via ?cities=City1,City2"}, 400)

    cities = [c.strip() for c in cities_param.split(",") if c.strip()]
    results = []
    errors = []

    for city in cities:
        Log.debug("Fetching weather", city=city)
        result = fetch_weather(city)

        if result["error"]:
            Log.warning("Weather fetch failed", city=city, error=result["error"])
            errors.append({"city": city, "error": result["error"]})
        else:
            results.append(result["body"])

    if errors:
        return response({
            "error": "One or more cities could not be fetched",
            "failed": errors,
            "succeeded": results
        }, 502)

    return response({
        "cities": results,
        "count": len(results)
    })
bash
curl "http://localhost:7146/api/dashboard/weather?cities=London,Berlin,Tokyo"
json
{
  "cities": [
    {"city": "London", "temp_c": 12.3, "description": "Cloudy", "humidity": 78},
    {"city": "Berlin", "temp_c": 8.1, "description": "Clear", "humidity": 55},
    {"city": "Tokyo", "temp_c": 18.7, "description": "Partly cloudy", "humidity": 65}
  ],
  "count": 3
}

11. Gotchas

1. Not checking result["error"]

Problem: Code accesses result["body"]["id"] but the request timed out, so body is None.

Fix: Always check result["error"] before accessing result["body"]. Treat any non-None error as a failure.

2. Using verify_ssl=False in production

Problem: A man-in-the-middle intercepts requests to a payment gateway because SSL verification is disabled.

Fix: Only use verify_ssl=False against local services or during development. Never disable SSL for external APIs.

3. No timeout set for slow external APIs

Problem: An external API hangs for 90 seconds. Your route handler blocks for 90 seconds. All workers are eventually held waiting.

Fix: Always set timeout to a sensible value (5-15 seconds for most external APIs). Return a 502 or 504 to the caller if the external API does not respond in time.

4. Hardcoding API keys

Problem: api = Api(bearer_token="sk-live-abc123...") exposes the key in source control.

Fix: Read credentials from environment variables: bearer_token=os.environ["PAYMENT_API_KEY"]. Never hardcode secrets.


12. Uploading Files (New in 3.13.69)

upload() posts a multipart/form-data body: a file plus optional text fields. You supply the file two ways, so your code never has to stage a temp file first.

python
import os
from tina4_python.api import Api

api = Api("https://api.example.com", bearer_token=os.environ["API_TOKEN"])

# A file on disk. filename defaults to the basename.
result = api.upload("/avatars", file_path="/tmp/me.png")

# In-memory bytes. Pass a filename so the server sees a real name.
raw = build_thumbnail()          # bytes
result = api.upload(
    "/avatars",
    file_bytes=raw,
    filename="me.png",
    extra_fields={"user_id": "42"},   # extra text parts
)

The full signature:

python
api.upload(path="", file_path=None, field_name="file",
           extra_fields=None, headers=None,
           file_bytes=None, filename=None) -> dict

field_name is the form field the file rides under (default "file"). extra_fields become additional text parts. headers merge extra per-call headers onto the request. The part's Content-Type is guessed from the filename, falling back to application/octet-stream.

upload() returns the same result dict every verb returns. A missing file, or no source at all, returns a clean error dict and never raises:

python
result = api.upload("/avatars", file_path="/tmp/gone.png")
# {"http_code": None, "body": None, "headers": {}, "error": "file not found: /tmp/gone.png"}

13. Streaming Downloads (New in 3.13.69)

download() streams a GET body straight to disk, 64KB at a time. A multi-gigabyte export never lands in memory at once.

python
result = api.download("/reports/2026.csv", dest_path="/tmp/2026.csv")

if result["error"] is None:
    print("saved to", result["path"])   # /tmp/2026.csv

The signature:

python
api.download(path="", dest_path=None, params=None) -> dict

The result has no body key. The body went to disk. It carries http_code, headers, error, and path. On success path is your dest_path; on any error (no dest, an HTTP error status, or a transport failure) path is None and no file is written.

python
{"http_code": 200, "headers": {...}, "error": None, "path": "/tmp/2026.csv"}

download() runs through the same opener as every verb, so redirect following, the cross-origin auth strip, and the SSL setting all apply.


14. Testing Your Code: the transport Seam (New in 3.13.69)

The constructor accepts a transport callable that fully replaces the network call. Point it at your own function and the code that calls an Api runs in a unit test with no live server.

python
def fake_transport(method, url, headers, body, timeout):
    return {"http_code": 200, "body": {"ok": True}, "headers": {}, "error": None}

api = Api("https://api.example.com", transport=fake_transport)
result = api.get("/health")     # returns the canned dict, opens no socket

The callable signature is transport(method, url, headers, body, timeout), and it returns the standard result dict.

This seam is for your tests, not Tina4's. The framework's own suite never injects a fake transport: it follows the no-mock rule and drives the real network against a real local server. Reach for transport to test the code that calls an Api, never to stand in for Api itself.


Pass cookies=True and the client keeps a per-instance, in-memory cookie jar. It reads Set-Cookie on each response and replays the accumulated Cookie header on the next request, so a session carries across a login and the calls that follow.

python
api = Api("https://api.example.com", cookies=True)

api.post("/login", {"user": "alice", "pass": "secret"})   # server sets a session cookie
api.get("/account")                                         # the cookie is sent automatically

The jar is off by default. It keeps only the leading name=value of each cookie, it is never written to disk, and it lives and dies with the instance.


16. Redirects and Cross-Origin Safety (New in 3.13.69)

The client follows redirects. On a redirect that crosses to a different origin (a different scheme, host, or port), it strips the Authorization header and the cookie-jar Cookie header before following.

That strip is a security boundary. Without it, a call to https://api.example.com/login that redirects to https://evil.example/ would hand your bearer token and session cookie to a host you never authenticated against. Same-origin redirects keep both headers, so an ordinary login-then-redirect flow is untouched.

You get this on every verb, on upload(), and on download(). There is nothing to switch on.

Sponsored with 🩵 by Code InfinityCode Infinity