faucet-source-graphql

Config-driven GraphQL API source for the faucet-stream ecosystem. Executes a GraphQL query against any endpoint, follows Relay-style cursor pagination, extracts records with a JSONPath expression, and streams them 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 you want to pull a paginated GraphQL collection (users, repositories, orders, issues…) out of a SaaS API or your own service. Pages are emitted as they arrive, so peak memory stays bounded no matter how many pages the query walks, and transient HTTP failures are retried automatically.

Feature highlights

• Native streaming — overrides Source::stream_pages: every upstream GraphQL response is emitted as one StreamPage and written to the sink immediately, so a million-row collection never buffers client-side.
• Relay cursor pagination — follows pageInfo { hasNextPage, endCursor }, injecting the endCursor back into the query's after: variable on each request. Stops cleanly when hasNextPage is false, the cursor is absent, the same cursor repeats (loop guard), or max_pages is reached. If has_next_page_path can't be resolved to a boolean on a page, the signal is treated as "unknown" and pagination defers to cursor presence (and warns once) rather than silently stopping — so a missing has-next field never drops the remaining pages.
• Offset pagination — for query languages that page with LIMIT … OFFSET … (ShopifyQL and similar). Injects an integer offset into a GraphQL variable (starting at 0, advancing by page_size after each page) and terminates on a short page — fewer than page_size records — rather than a pageInfo boolean. Set stop_when_short: false to keep paginating until a fully empty page instead. See Offset pagination.
• Variable injection — static variables from config, plus per-request cursor / page-size variables, plus parent-record context values (${parent.path} matrix fan-out) merged into the GraphQL variables map at runtime.
• JSONPath record extraction — records_path plucks the record array out of any response shape ($.data.users.edges[*].node). When unset, the whole data object is returned as a single record.
• Pluggable authentication — inline Bearer or custom-header auth, or a { ref: <name> } pointer to a shared auth: provider so many sources share one token with single-flight refresh.
• Automatic retries — transient HTTP failures (5xx, connection resets) are retried up to 3 times with exponential backoff + jitter (500 ms base).
• Client built once — the reqwest client is constructed in new() and reused for every request.

Installation

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

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

source-graphql is not in the CLI default build — enable it explicitly (or use the full / source aggregate features). Via the umbrella crate: cargo add faucet-stream --features source-graphql.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: graphql
    config:
      endpoint: https://api.example.com/graphql
      query: |
        query Users($after: String, $first: Int!) {
          users(first: $first, after: $after) {
            edges { node { id name email createdAt } }
            pageInfo { endCursor hasNextPage }
          }
        }
      auth:
        type: bearer
        config:
          token: ${env:API_TOKEN}
      records_path: $.data.users.edges[*].node
      pagination:
        has_next_page_path: $.data.users.pageInfo.hasNextPage
        cursor_path: $.data.users.pageInfo.endCursor
        cursor_variable: after
        page_size_variable: first
      batch_size: 100
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

faucet run pipeline.yaml

Configuration reference

Core

Pagination

pagination accepts one of two shapes, distinguished by the presence of a type: field. Omit the block entirely for a single-page query. max_pages caps either style.

Cursor pagination (Relay, the default shape — no type:)

Offset pagination (type: Offset)

Batching

Authentication

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

Mutual TLS (client certificates)

For endpoints requiring a client certificate, add a tls: block (requires the crate's mtls feature): a PEM pair (client_cert + client_key) or a PKCS#12 file (client_identity_pkcs12 + pkcs12_password), optional min_version. The certificate is presented on every request. See the faucet-source-rest README for the full field reference.

# Bearer token (resolved from the environment at load time)
auth:
  type: bearer
  config:
    token: ${env:API_TOKEN}

# Custom headers — API key + tenant
auth:
  type: custom
  config:
    headers:
      X-Api-Key: ${env:API_KEY}
      X-Tenant: acme

# Shared provider from the top-level auth: catalog
auth: { ref: my_idp }

When an auth: { ref } is resolved (or a shared provider is attached via with_auth_provider), the provider's credential takes precedence over any inline auth and is shared across every source referencing it, with single-flight token refresh.

Examples

GraphQL → PostgreSQL with cursor pagination

version: 1
name: graphql_to_postgres
pipeline:
  source:
    type: graphql
    config:
      endpoint: https://api.example.com/graphql
      query: |
        query Users($after: String, $first: Int!) {
          users(first: $first, after: $after) {
            edges { node { id name email createdAt } }
            pageInfo { endCursor hasNextPage }
          }
        }
      variables:
        first: 100
      auth:
        type: bearer
        config:
          token: ${env:API_TOKEN}
      records_path: $.data.users.edges[*].node
      pagination:
        has_next_page_path: $.data.users.pageInfo.hasNextPage
        cursor_path: $.data.users.pageInfo.endCursor
        cursor_variable: after
        page_size_variable: first
      batch_size: 100
  sink:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost/app
      table_name: users_imported
      column_mapping:
        type: auto_map
      batch_size: 1000
      max_connections: 8

GraphQL → BigQuery with custom-header auth and a page cap

version: 1
name: graphql_to_bigquery
pipeline:
  source:
    type: graphql
    config:
      endpoint: https://api.example.com/graphql
      query: |
        query Orders($after: String, $first: Int!) {
          orders(first: $first, after: $after) {
            edges { node { id total status createdAt } }
            pageInfo { endCursor hasNextPage }
          }
        }
      variables:
        first: 200
      auth:
        type: custom
        config:
          headers:
            X-Api-Key: ${env:API_KEY}
            X-Tenant: acme
      records_path: $.data.orders.edges[*].node
      pagination:
        has_next_page_path: $.data.orders.pageInfo.hasNextPage
        cursor_path: $.data.orders.pageInfo.endCursor
        cursor_variable: after
        page_size_variable: first
      max_pages: 500
      batch_size: 200
  sink:
    type: bigquery
    config:
      project_id: my-gcp-project
      dataset_id: raw
      table_id: orders
      auth:
        type: application_default
      batch_size: 1000

Single-page query (no pagination)

source:
  type: graphql
  config:
    endpoint: https://api.example.com/graphql
    query: "query($id: ID!) { user(id: $id) { name email } }"
    variables:
      id: user-123
    records_path: $.data.user
    # no `pagination:` block → one request, one page

Offset pagination (ShopifyQL)

For APIs whose query language pages with LIMIT … OFFSET …, bake the limit into the query and parameterize only the offset. Pagination stops on the first page shorter than page_size.

source:
  type: graphql
  config:
    endpoint: https://shop.example.com/admin/api/2024-10/graphql.json
    query: |
      query($q_offset: Int!) {
        shopifyqlQuery(query: "FROM orders SHOW total_sales LIMIT 250 OFFSET $q_offset") {
          ... on TableResponse { tableData { rowData } }
        }
      }
    auth:
      type: custom
      config:
        headers:
          X-Shopify-Access-Token: ${env:SHOPIFY_TOKEN}
    records_path: $.data.shopifyqlQuery.tableData.rowData[*]
    pagination:
      type: Offset
      offset_variable: q_offset
      page_size: 250
      stop_when_short: true

Streaming & batching

GraphqlStream overrides Source::stream_pages: each upstream GraphQL response becomes one StreamPage, written to the sink as it arrives rather than buffering the entire result set.

• batch_size (default 1000, max 1_000_000) maps directly to the page-size variable named by pagination.page_size_variable (first: by default). The upstream ultimately decides how many records to return per request — a server that caps first: produces smaller pages than requested.
• batch_size = 0 is the "no batching" sentinel: the page-size variable is omitted from the request so the upstream uses its own default page size, and the whole response is emitted as a single page. Use it for tiny lookup queries or when the upstream default already matches what the sink wants. If the upstream schema declares the page-size variable as non-null (first: Int!), the server returns a GraphQL validation error and the stream surfaces a FaucetError::Config — pick a non-zero batch_size in that case.

This is a one-shot query source — page bookmarks are always None, so it has no incremental-replication / resume support (each run re-walks the query from page one). For incremental loads, encode a watermark in the query and drive it from a matrix context value or a ${now.*} token.

Config loading & schema

Load from YAML/JSON or environment:

use faucet_core::config::{load_json, load_env_file};
use faucet_source_graphql::GraphqlStreamConfig;

let config: GraphqlStreamConfig = load_json("config.json")?;
let config: GraphqlStreamConfig = load_env_file(".env", "GRAPHQL")?;

Inspect the full JSON Schema with:

faucet schema source graphql

Library usage

use faucet_core::{Pipeline, Source};
use faucet_source_graphql::{GraphqlStream, GraphqlStreamConfig};
use faucet_source_graphql::config::{GraphqlAuth, GraphqlPagination};

# async fn run(my_sink: Box<dyn faucet_core::Sink>) -> Result<(), Box<dyn std::error::Error>> {
let config = GraphqlStreamConfig::new(
    "https://api.example.com/graphql",
    r#"query($first: Int!, $after: String) {
        users(first: $first, after: $after) {
            edges { node { id name email } }
            pageInfo { hasNextPage endCursor }
        }
    }"#,
)
.auth(GraphqlAuth::Bearer { token: "your-token".into() })
.records_path("$.data.users.edges[*].node")
.pagination(GraphqlPagination::default())
.with_batch_size(100)
.max_pages(20);

let source = GraphqlStream::new(config);
let pipeline = Pipeline::new(Box::new(source), my_sink);
let result = pipeline.run().await?;
println!("Transferred {} records", result.records_written);
# Ok(())
# }

To share one token across many sources, build a provider and inject it with GraphqlStream::new(config).with_auth_provider(provider).

How it works

1. new() builds the reqwest::Client once and reuses it for every request.
2. Each request merges static variables, the current cursor (into cursor_variable), the page-size value (into page_size_variable, unless batch_size = 0), and any parent-record context values into the GraphQL variables map.
3. The POST is retried up to 3 times with exponential backoff + jitter (500 ms base) on retriable HTTP failures. When driven by the CLI, a pipeline-level resilience: block replaces these built-in retry defaults with one shared policy — GraphQL honors the policy's max_attempts, base, max, jitter, and retry_on in full.
4. records_path extracts the record array via JSONPath; the page is yielded immediately.
5. Cursor pagination advances by reading hasNextPage / endCursor, stopping on a false flag, an absent or repeated cursor (loop guard), or max_pages. Offset pagination advances an integer offset variable by page_size each page, stopping on a short page (stop_when_short, default), a fully empty page, or max_pages.

Lineage dataset URI

https://<endpoint> with credentials stripped — e.g. https://api.example.com/graphql.

Feature flags

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

Troubleshooting / FAQ

See also

• Pagination guide — cursor pagination across connectors.
• Authentication guide — inline auth and the shared auth: catalog.
• Connector reference — capability matrix.
• faucet-source-rest — the REST sibling source with broader pagination styles.
• faucet-auth — shared OAuth2 / token-endpoint providers.

License

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

Observability

Metrics emitted by this source are labelled connector="graphql".