Ingesting API-Sourced Files
Filedge does not fetch from SaaS APIs directly. It ingests Files.
For API sources such as Stripe, Salesforce, HubSpot, Jira, or GitHub, use an upstream Fetcher to land complete NDJSON files in a Watched Directory. Then run Filedge against that directory. This keeps API-sourced data on the same ingestion path as file drops: Content Hash deduplication, PENDING -> COMMITTED audit state, strict validation, row-level provenance, and filedge status visibility.
See ADR-0006.
The boundary:
The Fetcher owns API behavior. Filedge starts when complete files appear in the Watched Directory.
Fetchers
A Fetcher can be any tool or job that writes complete files:
- a dlt pipeline configured to write NDJSON files
- a vendor export job
- a custom script
- Airbyte or Meltano configured to write files
- an existing internal ingestion job
dlt is a useful option because it already handles many SaaS APIs, but it is not a Filedge dependency and it is not the loader of record. The loader of record is still filedge run.
The Fetcher must guarantee:
- only complete files are promoted into the Watched Directory
- failed or partial fetches remain in staging or are deleted
- filenames are unique enough for operators to understand where they came from
- file contents are stable once visible to Filedge
The Reference Fetcher (filedge-fetch)
Filedge ships a runnable example of a correct Fetcher: filedge-fetch. It is an external companion to filedge run, not part of the core ingestion path and never a loader of record (see ADR-0018). It exists to show — and enforce — the contract above end to end.
Given a Sources Config, filedge-fetch:
- reads the last incremental cursor for the API Source;
- pages through the API (rate-limit aware) for records newer than that cursor;
- writes them as one complete NDJSON File in a staging area;
- emits an OpenLineage-shaped Source Manifest sidecar for that File;
- promotes the sidecar then the data File into the Watched Directory under a Fetch Lock — so a File is never visible without its provenance, and two concurrent fetches cannot race partial files;
- advances the cursor only after a successful promotion, so a crash retries the same window rather than skipping data.
# Pull one API Source into complete NDJSON Files + manifests in ./landing/
filedge-fetch --config examples/sources.yaml --source github-commits
# Preview the window and target file without fetching
filedge-fetch --config examples/sources.yaml --source github-commits --dry-run
# Then ingest exactly like any file drop
filedge run --dir ./landing --config pipeline.yaml --audit-db-url $FILEDGE_AUDIT_DB_URL
Run the two as independent scheduled jobs — the same two-job pattern recommended for SFTP sync in ADR-0005.
The reference targets the public GitHub REST API because it needs no credentials and exercises both pagination and a since-style cursor. The API-specific code sits behind an API Source adapter seam, so adapting it to a fintech API (Stripe, Plaid) is a new adapter plus Sources Config parsing, not a rewrite of the staging, promotion, manifest, or cursor logic. See How to add an API Source adapter for the extension pattern.
EDGAR is also supported by the Reference Fetcher through the SEC companyConcept
endpoint. It needs no API key, but SEC policy requires a descriptive
User-Agent contact string. EDGAR returns a whole concept document, so the
Fetcher applies the incremental cursor client-side by keeping facts whose
filed date is newer than the stored high-water mark.
For runnable client-facing walkthroughs, start with the Stripe-style API to audited DuckDB demo, then compare the no-credential EDGAR API to audited SQLite demo.
Sources Config
A Sources Config (sources.yaml) is a Fetcher-only file, separate from pipeline.yaml. It declares the endpoint, the incremental cursor, an optional environment-variable credential lookup, and the staging/state/landing paths. No secret is ever written into it.
version: 1
sources:
- name: github-commits
type: github
url: https://api.github.com/repos/tongqqiu/filedge/commits
staging_dir: ./staging
watched_directory: ./landing
state_dir: ./state
cursor:
param: since # query param carrying the high-water mark
field: commit.committer.date # dotted path the next cursor is read from
query:
sha: main
# credential_env: GITHUB_TOKEN # optional; bearer token from this env var
page_size: 100
gzip: false
- name: edgar-apple-revenues
type: edgar
cik: 320193
taxonomy: us-gaap # optional; defaults to us-gaap
concept: Revenues
unit: USD
user_agent: "Your Company your-email@example.com"
staging_dir: ./staging
watched_directory: ./landing
state_dir: ./state
cursor:
field: filed # client-side high-water mark in each fact
gzip: false
The same example ships at examples/sources.yaml.
EDGAR companyConcept workflow
Run the EDGAR source the same way as the GitHub source:
filedge-fetch --config examples/sources.yaml --source edgar-apple-revenues
filedge run --dir ./landing --config pipeline.yaml --audit-db-url $FILEDGE_AUDIT_DB_URL
For cik: 320193, taxonomy: us-gaap, and concept: Revenues, the loader
builds:
The Fetcher extracts one NDJSON row per fact from units.USD, emits a Source
Manifest with source_type: edgar and the covered filed date range, promotes
the File under the Fetch Lock, then advances the cursor only after promotion.
The same pattern can support SEC frames later: a different URL builder and
record path, with the staging, manifest, promotion, and cursor rules unchanged.
A matching pipeline.yaml sketch for the emitted facts:
format: ndjson
dest_table: edgar_company_facts
connector:
type: sqlite
url: sqlite:///filedge.db
columns:
- source: filed
dest: filed
type: string
required: true
- source: fy
dest: fiscal_year
type: integer
required: false
- source: fp
dest: fiscal_period
type: string
required: false
- source: form
dest: form
type: string
required: false
- source: val
dest: value
type: integer
required: true
Example: Stripe Events -> S3 -> BigQuery
1. Fetch to files
The Reference Fetcher has a first-party stripe source. Declare it in a Sources
Config — only the resource and the env var holding the secret key are required:
version: 1
sources:
- name: stripe-charges
type: stripe
resource: charges
credential_env: STRIPE_API_KEY
staging_dir: ./staging
watched_directory: ./landing
state_dir: ./state
# api_base: http://localhost:12111 # point at stripe-mock for a no-account run
Set the secret key in the named environment variable (never in the file) and fetch:
It pages the Stripe list API (starting_after / has_more), writes one complete
NDJSON File of the data records with a Source Manifest, promotes it under the
Fetch Lock, and advances the incremental created cursor only after promotion —
so a re-run fetches only newer charges (created[gt]). To try it without a Stripe
account, run stripe-mock and set
api_base to it.
External fetchers (dlt, a scheduled vendor export, an internal platform job) remain valid alternatives — anything that lands a complete NDJSON File works. See how to add an API Source.
2. Ingest
filedge run \
--dir s3://my-bucket/landing/stripe/ \
--config pipeline.yaml \
--audit-db-url $FILEDGE_AUDIT_DB_URL
3. Configure the file schema
pipeline.yaml describes the files that the Fetcher lands:
format: ndjson
connector:
type: bigquery
project: my-gcp-project
dataset: raw
dest_table: stripe_events
write_mode: append
columns:
- source: id
dest: id
type: string
required: true
- source: type
dest: type
type: string
required: true
- source: created
dest: created
type: timestamp
required: true
- source: data
dest: data
type: string
required: false
Example: Salesforce -> local files -> PostgreSQL
An internal job or external Fetcher lands complete files:
Then Filedge ingests those files:
filedge run \
--dir /data/landing/salesforce/ \
--config pipeline.yaml \
--audit-db-url $FILEDGE_AUDIT_DB_URL
Scheduling
Schedule the Fetcher before filedge run. Different API sources can run on different cadences. Filedge does not need to know which tool produced the files.
Cloud Scheduler / EventBridge:
Every 15 min:
├── run-stripe-fetcher --output s3://.../landing/stripe/
Every hour:
├── run-salesforce-fetcher --output s3://.../landing/salesforce/
Every hour + 10 min:
├── filedge run --dir s3://.../landing/stripe/ --config pipeline.yaml ...
└── filedge run --dir s3://.../landing/salesforce/ --config pipeline.yaml ...
Airflow:
fetch_stripe = BashOperator(
task_id="fetch_stripe",
bash_command="run-stripe-fetcher --output s3://.../landing/stripe/",
)
ingest_stripe = BashOperator(
task_id="ingest_stripe",
bash_command="filedge run --dir s3://.../landing/stripe/ --config pipeline.yaml ...",
)
fetch_stripe >> ingest_stripe
Audit visibility
Because API-sourced data passes through the same pipeline:
filedge status --audit-db-url $FILEDGE_AUDIT_DB_URL
PENDING: 0
PROCESSING: 0
COMMITTED: 142
FAILED: 1
Recent failures:
stripe_20260522T140000_0001.ndjson: schema mismatch on column 'amount_decimal'
Every destination row carries _source_file_hash linking it back to the exact fetched file. An auditor asking "what Stripe data landed on 2026-05-22 and which rows did it produce?" can answer from Filedge's Audit DB and destination provenance columns without knowing which Fetcher produced the file.
Responsibility boundary
| Concern | Owner |
|---|---|
| API authentication | Fetcher |
| Pagination | Fetcher |
| Rate limiting | Fetcher |
| Incremental cursor (last fetched ID/timestamp) | Fetcher |
| Concurrent fetch prevention | Fetcher or scheduler |
| Partial-fetch atomicity (staging -> Watched Directory) | Fetcher |
| File-level deduplication (content hash) | filedge run |
| PENDING → COMMITTED state machine | filedge run |
Row-level provenance (_source_file_hash) |
filedge run |
| Retry on ingestion failure | filedge run |
Operator visibility (filedge status) |
filedge run |
Source Manifests (optional)
The Fetcher can write a *.manifest.json sidecar next to each NDJSON file to record the API cursor range, run ID, and producer identity:
landing/stripe/
stripe_events_20260522T140000_0001.ndjson
stripe_events_20260522T140000_0001.ndjson.manifest.json
Filedge reads the sidecar at registration and stores the metadata on the File's Audit Record. Then operators can drill in with filedge lineage or filter filedge status --json failures by API source.
{
"eventType": "COMPLETE",
"eventTime": "2026-05-22T14:00:30Z",
"producer": "https://github.com/dlt-hub/dlt",
"run": {
"runId": "dlt-stripe-2026-05-22T14:00",
"facets": {"_filedgeManifest": {"manifest_version": "1", "record_count": 1500}}
},
"job": {"namespace": "api", "name": "stripe.charges"},
"inputs": [{
"name": "https://api.stripe.com/v1/charges",
"facets": {"_sourceRange": {
"cursor_start": "ch_3OXa...",
"cursor_end": "ch_3OYz...",
"endpoint": "/v1/charges"
}}
}]
}
See Source Manifests for the full schema, policy modes (disabled / optional / required), and validation error categories. Direct file drops without manifests continue to work unchanged under the default optional policy.
Related
- API Source adapters — extend the Fetcher to a new API
- Source manifests — upstream lineage for fetched Files
- Queue sources — the same pattern for message brokers
- EDGAR tutorial — see the Fetcher end to end