Declarative HTTP Transport

StreamWright includes a unified, declarative HTTP engine for REST API sources and connectors. Instead of writing custom Python networking loops, pagination handlers, or retry logic, you express endpoints, headers, authentication, query parameters, paginators, and record extraction directly in YAML.

When a connector specifies transport = "http", StreamWright executes all network requests through its core HttpClient and Authenticator pipelines.

---

How It Works and How It's Wired

A declarative HTTP source connects three layers:

1. source.yaml sets source-wide defaults: Declares the auth: provider and type, plus the http: block (default base_url, shared headers, default paginator, records envelope, and retry policy).
2. streams/*.yaml declare requests: Each request defines its HTTP endpoint (path, method), query params, or JSON payload (json). Requests inherit or override the source-level defaults.
3. Connector package or generic REST API: A lightweight connector package (e.g., streamwright-openai-ads or the generic streamwright-restapi) specifies transport="http" in its ConnectorSpec. No networking or vendor SDK code lives in the connector.
4. Execution by SourceRunner:
   • Evaluates Jinja expressions ({{ config.* }}, {{ secrets.* }}, {{ partition.* }}, {{ window.* }}).
   • Signs requests via Authenticator.
   • Fetches pages iteratively using the configured Paginator.
   • Extracts record arrays from the response using records.path.
   • Loads extracted records into DuckDB for SQL transformations in transform.steps.

---

Source Schema (source.yaml)

The http: block in source.yaml configures defaults shared by all streams in the source:

source.yaml

kind: source
name: my_api_demo

spec:
  config:
    account_ids: { type: list, items: string }
  secrets:
    api_token: { type: string }

auth:
  provider: my_api
  type: bearer
  token: "{{ secrets.api_token }}"

http:
  base_url: https://api.example.com
  headers:
    X-API-Version: "v1"
  params_encoding: plain           # "plain" (default) or "dotted"
  paginator:
    type: cursor
    token_path: pagination.next_token
    param: cursor
    has_more_path: pagination.has_more
  records:
    path: data                     # Extracts records from {"data": [...]}
  retry:
    codes: [429, 500, 502, 503, 504]
    max_attempts: 5
    backoff: exponential
    max_delay: 2m

Schema Reference: http in source.yaml

Property	Type	Default	Description
base_url	string	""	Root URL prepended to relative stream request paths (e.g. https://api.example.com). Supports Jinja expressions.
headers	object	{}	Key-value pairs added to every outgoing HTTP request. Templating supported.
params_encoding	string	"plain"	How query parameter objects and lists are encoded: "plain" or "dotted" (flattens nested dicts to parent.child=val and lists to items[0]=val).
paginator	object	optional	Default pagination strategy inherited by all stream requests (see Paginators).
records	object	optional	Default record extraction config (e.g., { path: "data" }).
retry	object	optional	Default retry policy for handling rate limits (429) and transient server errors.

---

Authentication Schemes

StreamWright supports multiple declarative authentication types under auth::

Bearer Token

auth:
  provider: my_network
  type: bearer
  token: "{{ secrets.access_token }}"

Sets Authorization: Bearer <token> on every request.

OAuth2 Refresh Token Grant

For APIs using OAuth 2.0 (RFC 6749) with short-lived access tokens, StreamWright provides native, automated token lifecycle management using the Refresh Token grant (type: oauth2_refresh_token).

auth:
  provider: my_network                 # or omitted for generic HTTP sources
  type: oauth2_refresh_token
  token_url: https://auth.example.com/oauth2/token
  client_id: "{{ secrets.client_id }}"
  client_secret: "{{ secrets.client_secret }}"
  refresh_token: "{{ secrets.refresh_token }}"
  scopes:                              # optional: list of requested OAuth scopes
    - read:campaigns
    - read:reports

Schema Reference

Property	Type	Required	Description
type	string	Yes	Must be "oauth2_refresh_token".
token_url	string	Yes	The authorization server's token endpoint URL.
client_id	string	Yes	The registered OAuth2 client identifier.
client_secret	string	Yes	The client secret (referenced from {{ secrets.* }}).
refresh_token	string	Yes	The long-lived refresh token (referenced from {{ secrets.* }}).
scopes	list of strings	No	Optional list of OAuth 2.0 scopes sent in the token request as a space-separated scope parameter.

Lifecycle & Execution Mechanics

StreamWright handles the entire token lifecycle transparently inside Authenticator:

1. Initial Token Exchange: On the first HTTP request that requires authentication, StreamWright issues a POST request to token_url with form-encoded payload:

   POST /oauth2/token HTTP/1.1
   Host: auth.example.com
   Content-Type: application/x-www-form-urlencoded
   
   grant_type=refresh_token&client_id=...&client_secret=...&refresh_token=...&scope=read%3Acampaigns+read%3Areports

2. In-Memory Caching & Preemptive Refresh:

   • The returned access_token is cached in memory for the duration of the run.
   • StreamWright inspects the response's expires_in field (defaulting to 3,600 seconds / 1 hour if omitted).
   • To prevent mid-request token expiration during long batch runs, StreamWright preemptively requests a fresh token 60 seconds before the active token expires.

3. Reactive 401 Recovery: If an API endpoint fails with HTTP 401 Unauthorized (e.g. unexpected revocation or clock skew):

   • StreamWright immediately drops the cached access token.
   • It performs an on-demand refresh call to token_url.
   • If the refresh succeeds, the original HTTP request is retried once seamlessly without failing the pipeline.

4. Request Header Injection: Every outgoing stream request automatically carries the active access token:

   Authorization: Bearer <access_token>

5. Zero-Leak Secret Redaction: All tokens returned by the authorization server—including the access_token, any rotated refresh_token, and optional id_token—are automatically registered with StreamWright's engine redactor. They are masked as *** in all execution logs, HTTP summary lines, and error traces.

API Key Header or Query Param

auth:
  provider: my_network
  type: api_key
  name: X-API-Key
  value: "{{ secrets.api_key }}"
  in: header                       # "header" (default) or "query"

Basic Authentication

auth:
  provider: my_network
  type: basic
  username: "{{ secrets.username }}"
  password: "{{ secrets.password }}"

---

Stream Schema (streams/*.yaml)

Individual stream requests define their HTTP calls under requests[].http:

streams/campaigns.yaml

partitions:
  - { name: account_id, values: "{{ config.account_ids }}" }

requests:
  - name: raw_campaigns
    http:
      path: /v1/campaigns
      method: GET
      params:
        account_id: "{{ partition.account_id }}"
        limit: 100

transform:
  mode: page
  steps:
    - name: campaigns
      select: |
        SELECT partition->>'account_id' AS account_id,
               record->>'id'            AS campaign_id,
               record->>'name'          AS campaign_name,
               record->>'status'        AS status
        FROM raw_campaigns

export:
  campaigns:
    step: campaigns
    primary_key: [account_id, campaign_id]

Schema Reference: requests[].http

Property	Type	Default	Description
path	string	required	Endpoint path relative to base_url (or a full URL). Supports Jinja expressions.
method	string	"GET"	HTTP method: GET, POST, PUT, DELETE, PATCH.
params	object	{}	Query parameters. Values can be Jinja templates or nested structures when params_encoding: dotted.
json	any	optional	JSON payload serialized into the request body. Header Content-Type: application/json is added automatically.
data	any	optional	Raw string or form body.
headers	object	{}	Request-specific headers. Merged with source-level headers.
paginator	object	optional	Stream-specific paginator. Overrides the source-level paginator.
records	object	optional	Stream-specific records extraction path (e.g. { path: "result.rows" }).
retry	object	optional	Stream-specific retry override.

---

Paginators

StreamWright includes built-in paginators that automatically follow multiple pages of data until exhaustion.

Paginators support in: query (default) or in: body (for POST search queries where pagination controls are embedded in the JSON body).

1. offset

Paginates using numeric offset / skip values:

paginator:
  type: offset
  offset_param: start              # or offset, startIndex
  limit_param: count               # or limit, pageSize
  page_size: 100
  total_path: paging.total         # optional: stops when offset >= total
  in: query                        # "query" (default) or "body"

2. cursor

Paginates using an opaque continuation token:

paginator:
  type: cursor
  token_path: last_id              # response path to the next token
  param: after                     # request parameter name
  has_more_path: has_more          # optional boolean field indicating if more pages exist
  in: query                        # "query" (default) or "body"

3. page_number

Paginates using an incrementing page counter:

paginator:
  type: page_number
  page_param: page                 # parameter name (e.g., page)
  size_param: page_size            # optional
  page_size: 50
  start: 1                         # initial page index (default: 1)
  total_pages_path: total_pages    # optional: stops when page > total_pages
  in: query                        # "query" (default) or "body"

4. link_header

Follows RFC 8288 Link: <url>; rel="next" response headers automatically:

paginator:
  type: link_header

5. none

Disables pagination for endpoints that return complete datasets or single objects:

paginator:
  type: none

Origin Isolation & Security

When following next-page URLs (Link headers or extracted next_url paths), StreamWright enforces origin isolation. Requests will only follow URLs matching the same scheme, host, and port as the originating request. Links pointing to a different origin or attempting to downgrade from HTTPS to HTTP are blocked to protect authentication credentials.

---

Parameter Encoding (plain vs dotted)

APIs such as LinkedIn Rest.li or complex filtering endpoints expect nested objects and arrays in query parameters.

By setting params_encoding: dotted (in source.yaml or on a request):

params_encoding: dotted
params:
  filter:
    status: ACTIVE
  accounts:
    - act_1
    - act_2

StreamWright serializes the parameters as:

?filter.status=ACTIVE&accounts[0]=act_1&accounts[1]=act_2