faucet-source-xml

A config-driven XML / SOAP API source for the faucet-stream ecosystem. It fetches an XML (or SOAP) HTTP endpoint, converts the response to JSON, pulls the repeating record element out by a dot-separated element path, and streams the records page-by-page into any faucet-stream sink — a file, a database, a warehouse, a queue — with one declarative config and no glue code.

Reach for it when the upstream system only speaks XML or SOAP: legacy enterprise web services, RSS/Atom-style feeds, government data portals, or any HTTP endpoint that returns application/xml. The parser is event-driven (streaming quick-xml), so peak memory stays bounded by batch_size even for large documents.

Feature highlights

• Automatic XML → JSON conversion — every response is parsed into a serde_json::Value tree; attributes, text nodes, and repeated elements all map to predictable JSON shapes.
• Element-path record extraction — records_element_path walks a dot-separated path (e.g. Envelope.Body.GetUsersResponse.Users.User) to the repeating element and emits one record per match. A single element collapses to one record; a repeated element fans out to many.
• First-class SOAP — a soap: block assembles the envelope, injects the version-correct headers (SOAPAction for 1.1, Content-Type action param for 1.2), resolves records_element_path relative to Envelope.Body, and surfaces SOAP <Fault> responses as errors. Or drop down to the raw-envelope path (method: POST + body) any time.
• Two pagination styles — page-number and offset/limit, each with a built-in loop guard so a misbehaving endpoint can't spin forever; max_pages is a hard cap across both.
• Pluggable authentication — none, bearer, basic, or fully custom headers — inline or via a shared provider from the CLI auth: catalog (OAuth2 token reuse across matrix rows).
• Retry with backoff — transient HTTP failures (5xx / connection resets) are retried up to 3 attempts with exponential backoff (base 500 ms) via the shared faucet_core::execute_with_retry.
• Bounded-memory streaming — Source::stream_pages accumulates matched subtrees and yields a StreamPage every batch_size records; batch_size: 0 drains the whole document into one page.
• Client built once — the reqwest client is constructed in new() and reused for every request and every page.

Installation

# As a library:
cargo add faucet-source-xml

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-xml

Or pull it in through the umbrella crate:

cargo add faucet-stream --features source-xml

source-xml is not in the CLI default build — enable it explicitly via the feature flag above.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://api.example.com
      path: /users.xml
      method: GET
      records_element_path: Response.Users.User
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

faucet run pipeline.yaml

Configuration reference

Core

Authentication

Pagination

Batching

headers exists on the Rust config struct for programmatic use but is #[serde(skip)] — it is not settable from YAML/JSON. Use Custom auth (or query_params) to attach request headers from config.

Authentication

auth accepts either an inline { type, config } block or a { ref: <name> } pointer to a shared provider declared in the CLI's top-level auth: catalog. Inline variants:

# Bearer token (read from the environment)
auth:
  type: bearer
  config:
    token: ${env:FEED_TOKEN}

# HTTP Basic
auth:
  type: basic
  config:
    username: admin
    password: ${env:XML_PASSWORD}

# Custom headers — e.g. SOAPAction + an API key
auth:
  type: custom
  config:
    headers:
      SOAPAction: "http://example.com/orders/GetOrders"
      X-API-Key: ${env:API_KEY}

# Shared provider from the top-level auth: catalog (OAuth2 token reused across rows)
auth: { ref: my_idp }

Examples

REST XML API with page-number pagination

version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://api.example.com
      path: /api/products.xml
      records_element_path: Products.Product
      pagination:
        type: PageNumber
        param_name: page
        start_page: 1
        page_size: 50
        page_size_param: per_page
      max_pages: 20
  sink:
    type: jsonl
    config:
      path: ./products.jsonl

SOAP service with Basic auth (first-class soap: block)

version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://soap.example.com
      path: /ws
      method: POST
      auth:
        type: basic
        config:
          username: admin
          password: ${env:SOAP_PASSWORD}
      soap:
        version: "1.1"                # "1.1" (default) or "1.2"
        action: http://example.com/orders/GetOrders
        body_inner: '<GetOrders xmlns="http://example.com/orders"/>'
      # Resolved relative to Envelope.Body by default:
      records_element_path: GetOrdersResponse.Orders.Order
  sink:
    type: postgres
    config:
      connection_url: ${env:DATABASE_URL}
      table: orders

Offset-paginated feed → MongoDB (Bearer auth)

version: 1
name: xml_to_mongodb
pipeline:
  source:
    type: xml
    config:
      base_url: https://feeds.example.com
      path: /catalog
      method: GET
      auth:
        type: bearer
        config:
          token: ${env:FEED_TOKEN}
      query_params:
        format: xml
      records_element_path: catalog.products.product
      pagination:
        type: Offset
        offset_param: offset
        limit_param: limit
        limit: 250
  sink:
    type: mongodb
    config:
      connection_uri: mongodb://localhost:27017
      database: warehouse
      collection: catalog_items
      batch_size: 1000

Runnable copies of the last two shapes live in cli/examples/xml_to_mongodb.yaml and cli/examples/xml_to_s3.yaml.

SOAP

The optional soap: block is sugar over the XML-over-HTTP request/response path — it is not a WSDL client. When present, the source:

1. assembles the request envelope from version (the namespace), any declared namespaces (prefix → URI on the envelope element), and body_inner (placed inside <soap:Body>);
2. injects the version-correct headers, regardless of the auth variant (so real bearer / basic auth stays free):
   • 1.1 → SOAPAction: "<action>" header and Content-Type: text/xml; charset=utf-8;
   • 1.2 → Content-Type: application/soap+xml; charset=utf-8; action="<action>" and no SOAPAction header;
3. resolves records_element_path relative to Envelope.Body when path_relative_to_body (default) — you write GetUsersResponse.Users.User;
4. surfaces a SOAP <Fault> as FaucetError::Source when fault_as_error (default); set it false to emit zero records (logged once) instead.

Load-time validation. Setting both the top-level body and soap.body_inner is rejected as ambiguous, and a soap: block with method: GET is rejected (SOAP is a POST protocol). Both surface as FaucetError::Config before any request is made.

Manual alternative — a raw SOAP envelope

The soap: block is optional sugar. You can always hand-write the whole envelope as a raw body and target the fully-qualified path (namespace prefixes preserved). This is the pre-soap: behavior and is unchanged:

version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://soap.example.com
      path: /ws
      method: POST
      auth:
        type: custom
        config:
          headers:
            SOAPAction: '"http://example.com/orders/GetOrders"'
      body: |
        <?xml version="1.0"?>
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
          <soapenv:Body>
            <GetOrders xmlns="http://example.com/orders"/>
          </soapenv:Body>
        </soapenv:Envelope>
      records_element_path: soapenv:Envelope.soapenv:Body.GetOrdersResponse.Orders.Order
  sink:
    type: jsonl
    config:
      path: ./orders.jsonl

Pagination

max_pages caps the total number of pages for both styles. With no pagination block, exactly one request is made.

Streaming & batching

The source overrides Source::stream_pages. The XML response is parsed with an event-driven quick-xml reader: only the subtree matching records_element_path is materialized, accumulated into a buffer, and yielded as a StreamPage whenever the buffer reaches batch_size. Memory is bounded at one page regardless of document size. With batch_size: 0, the whole document is drained and emitted in a single page — handy for small lookup payloads or for sinks (SQL COPY, BigQuery load jobs) that prefer one large request to many small ones.

This is a one-shot fetch source: each run re-requests the endpoint and has no incremental bookmark / resume support. For incremental loads, encode a watermark in query_params or the request body (e.g. a since=${now.date} param) and drive it from a matrix context or ${now.*} token.

Config loading & schema introspection

Load from YAML/JSON files, environment variables, or a .env file via the helpers in faucet_core::config:

use faucet_core::config::{load_json, load_env_file};
use faucet_source_xml::XmlStreamConfig;

let config: XmlStreamConfig = load_json("config.json")?;
let config: XmlStreamConfig = load_env_file(".env", "XML")?;

Inspect the full JSON Schema with:

faucet schema source xml

Library usage

use faucet_core::Source;
use faucet_source_xml::{XmlStream, XmlStreamConfig, XmlAuth, XmlPagination};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let config = XmlStreamConfig::new("https://feeds.example.com", "/catalog")
    .auth(XmlAuth::Bearer { token: std::env::var("FEED_TOKEN")? })
    .query_param("format", "xml")
    .records_element_path("catalog.products.product")
    .pagination(XmlPagination::Offset {
        offset_param: "offset".into(),
        limit_param: "limit".into(),
        limit: 250,
    })
    .with_batch_size(500);

let records = XmlStream::new(config).fetch_all().await?;
println!("fetched {} records", records.len());
# Ok(())
# }

To wire it into a pipeline, hand the XmlStream to faucet_core::Pipeline (batch) or faucet_core::run_stream (streaming) alongside any Sink. For shared-token reuse across many sources, build a faucet_core::AuthProvider once and inject it with XmlStream::with_auth_provider(provider).

How it works

1. new() builds the reqwest client once and stores it on the struct.
2. Each page issues one request (base_url + path with query_params + the pagination param), retried up to 3 times with exponential backoff (base 500 ms) on transient failures via faucet_core::execute_with_retry. Non-cloneable request bodies are not retried (surfaced as FaucetError::Source). When driven by the CLI, a pipeline-level resilience: block replaces these built-in retry defaults with one shared policy — XML honors the policy's max_attempts, base, max, jitter, and retry_on in full.
3. The response is converted XML → JSON and records_element_path is walked to find the repeating element.
4. Records are buffered and yielded as StreamPages of batch_size; pagination advances until the stop condition or max_pages.

Lineage dataset URI

<base_url><path> with any embedded credentials stripped — e.g. https://soap.example.com/svc.

Feature flags

This crate has no optional features of its own. Enable it in the CLI / umbrella crate via the source-xml feature.

Troubleshooting / FAQ

See also

• Connector reference — capability matrix.
• Authentication cookbook — inline vs shared auth: providers.
• Pagination cookbook.
• faucet-source-rest — the JSON/REST sibling source.
• faucet-source-graphql — GraphQL source.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.