Authenticate requests¶

Static credentials go in the config. Credentials that expire, such as an OAuth token or a session cookie, need code that renews them, and that code is a middleware: a class beside the config that databrew calls around each request.

Static headers¶

When an API key or a fixed header is enough, set it in [fetch]:

[fetch]
headers = { "Authorization" = "Bearer abc123", "Accept" = "application/json" }

Avoid committing secrets with the config; a middleware can read them from the environment instead, as below.

A token that expires¶

This middleware fetches a token before the first request, sends it with every request, and fetches a new one when the API answers 401 Unauthorized:

site_auth.pysite.toml

import asyncio
import os

import httpx

from databrew import Middleware, MiddlewareContext


class RefreshToken(Middleware):
    """Send a bearer token with every request, and fetch a new one on 401."""

    def __init__(self, token_url: str) -> None:
        self.token_url = token_url
        self.token: str | None = None
        self._lock = asyncio.Lock()

    async def _refresh(self, stale: str | None) -> None:
        """Replace the token, unless another request replaced `stale` meanwhile."""
        async with self._lock:
            if self.token != stale:
                return
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    self.token_url, data={"client_secret": os.environ["SITE_SECRET"]}
                )
                response.raise_for_status()
            self.token = response.json()["access_token"]

    async def pre_fetch(self, ctx: MiddlewareContext) -> MiddlewareContext:
        if self.token is None:
            await self._refresh(None)
        ctx.headers["Authorization"] = f"Bearer {self.token}"
        return ctx

    async def on_error(self, ctx: MiddlewareContext) -> MiddlewareContext:
        if ctx.error is None or ctx.error.http_status != 401:
            return ctx
        sent = ctx.headers.get("Authorization", "").removeprefix("Bearer ")
        try:
            await self._refresh(sent)
        except httpx.HTTPError as error:
            ctx.stop_reason = f"token refresh failed: {error}"
            return ctx
        ctx.error = None  # retry this URL with the new token
        return ctx

name = "site"
start_urls = ["https://api.example.com/v1/listings?page=1"]

[listing]
type = "json"

[listing.items]
path = "data.items"
key = "id"

[listing.links]
pagination = ["links.next"]

[[middleware]]
use = "site_auth:RefreshToken"
token_url = "https://api.example.com/oauth/token"

The [[middleware]] table names the class and passes token_url to its constructor, as [[middleware]] describes.

The hooks do the work:

pre_fetch fetches the first token and sets the Authorization header on ctx.headers.
on_error sees the 401 on ctx.error. Clearing ctx.error retries the URL within its max_retries, and the retry's pre_fetch sends the new token.
Requests run concurrently, so _refresh holds a lock: the first request fetches the token while the others wait, and a burst of 401 answers fetches one new token rather than one each. A request whose token another request already replaced only retries.
When the token cannot be renewed, every further request would fail. Setting ctx.stop_reason ends the run with the blocked stop as soon as the requests in flight finish, instead of waiting for the stop rules. The reason appears in the report, and the queued URLs wait for the next run.

Headers need an HTTP backend

The browser backend refuses per-request headers, so with it every request fails.

The middleware reference lists every hook and the order they run in.