File APIs Database and SaaS Ingestion Patterns

Purpose: choosing the right ingestion shape

File, API, database, and SaaS ingestion patterns are the first mile of a data engineering system. You will learn to choose an extraction method that preserves correctness without straining the source. A nightly CSV dropped into object storage, a paginated REST endpoint, a transactional database table, and a managed SaaS export all deliver records, but they expose very different signals about time, ordering, identity, and deletion.

In a batch pipeline, ingestion is the controlled movement from an external system into a durable landing area where downstream transforms can replay, audit, and recover. Good ingestion records the request, response, batch name, and rerun path.

How the mechanisms work internally

A file-based source usually separates production from consumption. A vendor, application, or upstream job writes a file, often to SFTP, cloud object storage, or a shared volume. The ingestion job detects a complete file, validates its name and schema, copies it into a raw zone, and records a manifest row. The key problem is completeness. A reader must not process a file while it is still being written, so robust feeds use atomic rename, ready marker files, immutable object keys, or manifests that list the exact payload objects.

An HTTP API source exposes a request-response contract. The ingestion job builds requests with authentication, filters, page cursors, and rate-limit handling. The API may paginate by page number, opaque cursor, offset, or time window. Cursor pagination is safer for changing datasets because the server owns the continuation token. Offset pagination can skip or duplicate rows during inserts. Time-window pagination needs overlap when the source updates old records.

Database ingestion normally uses one of three mechanisms. A full snapshot reads the entire table and replaces or versions the downstream copy. A high-watermark extract selects rows whose update timestamp or monotonically increasing id is greater than the previous checkpoint. Change data capture, or CDC, reads a database log or change stream and captures inserts, updates, and deletes in commit order. Snapshots are simple but expensive, watermarks are efficient but may miss hard deletes and late timestamp corrections, and CDC is precise but operationally more complex.

SaaS ingestion sits between API and file ingestion. Many SaaS platforms have REST APIs, bulk export jobs, webhook notifications, and prebuilt connectors. Internally, a connector must still solve the same problems: authenticate, enumerate entities, page through records, checkpoint progress, normalize nested JSON, and decide how to represent deleted or merged business objects. The SaaS-specific challenge is that fields and limits often reflect product behavior, not database design. For example, an account merge may preserve one customer id, retire another, and emit only an audit event.

API and configuration anatomy

Every ingestion configuration should make the source, extraction window, destination, and checkpoint explicit: what to read, what to request, where to land raw data, and what has been committed.

{
  "source_type": "api",
  "name": "billing_invoices",
  "endpoint": "https://api.example.test/invoices",
  "incremental_field": "updated_at",
  "checkpoint": "2026-09-01T00:00:00Z",
  "lookback_minutes": 15,
  "page_size": 200,
  "raw_destination": "raw.billing_invoices"
}

This configuration says the extractor will request invoice records updated after the stored checkpoint, but with a 15 minute lookback. If the last committed checkpoint is midnight, the next run asks from 23:45 the previous day. The overlap creates duplicates so late updates are not lost. The downstream load must therefore be idempotent, commonly by merging on invoice id and keeping the highest updated_at value.

Example 1: landing immutable files with a manifest

Suppose a payment processor drops daily settlement files into object storage. A fragile job might list the prefix and load every CSV it sees. A more reliable pattern requires a manifest file that names the complete payload. The manifest is the source of truth for the batch, while each payload file remains immutable.

from dataclasses import dataclass

@dataclass(frozen=True)
class ManifestEntry:
    batch_id: str
    path: str
    rows: int

def validate_manifest(entries):
    if not entries:
        raise ValueError("manifest is empty")
    batch_ids = {entry.batch_id for entry in entries}
    if len(batch_ids) != 1:
        raise ValueError("manifest mixes batches")
    for entry in entries:
        if not entry.path.endswith(".csv"):
            raise ValueError(f"not a csv payload: {entry.path}")
        if entry.rows < 0:
            raise ValueError(f"negative row count: {entry.path}")
    return sum(entry.rows for entry in entries)

entries = [
    ManifestEntry("settlement_2026_09_06", "s3://raw/payments/2026-09-06/part-000.csv", 1200),
    ManifestEntry("settlement_2026_09_06", "s3://raw/payments/2026-09-06/part-001.csv", 1187),
]
assert validate_manifest(entries) == 2387

The deterministic output is a successful assertion. If one entry used a different batch id or a non-CSV extension, validation would fail before loading. This is the core file pattern: prove completeness before parsing, then record the manifest and files durably.

Example 2: paginating an API with an overlap checkpoint

For an API that returns records sorted by updated_at, the extractor should separate the requested start time from the committed checkpoint. The start time includes a lookback. The committed checkpoint should advance only after the raw data and run metadata have been written successfully.

from datetime import datetime, timedelta, timezone

def request_start(last_checkpoint, lookback_minutes):
    return last_checkpoint - timedelta(minutes=lookback_minutes)

def next_checkpoint(records, previous_checkpoint):
    if not records:
        return previous_checkpoint
    latest = max(record["updated_at"] for record in records)
    return max(previous_checkpoint, latest)

previous = datetime(2026, 9, 7, 10, 0, tzinfo=timezone.utc)
records = [
    {"id": "inv_1", "updated_at": datetime(2026, 9, 7, 9, 58, tzinfo=timezone.utc)},
    {"id": "inv_2", "updated_at": datetime(2026, 9, 7, 10, 4, tzinfo=timezone.utc)},
]
assert request_start(previous, 15).isoformat() == "2026-09-07T09:45:00+00:00"
assert next_checkpoint(records, previous).isoformat() == "2026-09-07T10:04:00+00:00"

The next request starts at 09:45 UTC even though the committed checkpoint was 10:00 UTC. Record inv_1 may already exist downstream, so the loader must upsert rather than append blindly. The checkpoint advances to 10:04 UTC because that is the latest observed update after a successful run.

Example 3: database watermark extraction and merge intent

A database table with updated_at and a stable primary key can often be ingested with a high-watermark query. The extractor reads rows updated since the previous checkpoint, lands them unchanged, and lets the warehouse merge repeated versions. The pattern is efficient, but it captures only rows that remain visible to the query.

SELECT
  customer_id,
  email,
  status,
  updated_at
FROM app.customers
WHERE updated_at >= TIMESTAMP '2026-09-07 09:45:00+00:00'
  AND updated_at < TIMESTAMP '2026-09-07 10:15:00+00:00'
ORDER BY updated_at, customer_id;

If customer C101 changed at 09:52 and C205 changed at 10:02, both rows are returned. The upper bound gives the batch a closed window. Without it, a long-running query may chase new updates and make it hard to know what the batch represents. A later merge keeps the latest row per customer_id. However, if C101 is physically deleted, this query returns nothing for that delete. Use soft deletes, audit tables, or CDC when delete semantics matter.

Example 4: SaaS object sync with deleted records

SaaS systems expose business objects rather than relational rows. A support platform might provide tickets, users, organizations, comments, and audit events. Decide whether each object lands as current-state records, append-only events, or both.

For a SaaS connector, a deleted ticket may appear as a tombstone record, a status flag, or an audit event. Define downstream behavior before loading: remove the current-state row, mark it deleted, or keep it but suppress it in marts. In analytics pipelines, marking the row with deleted_at is usually safer than physical deletion because historical metrics and replay remain possible.

Design choices and trade-offs

• Full refresh is easiest to reason about and works well for small dimension-like sources, but it consumes more network, storage, and database capacity as data grows.
• Incremental watermarks are lightweight and simple to schedule, but they rely on a trustworthy monotonic field and need overlap plus idempotent loads.
• CDC gives the richest representation of inserts, updates, and deletes, but it introduces log retention, schema evolution, snapshot coordination, and ordering concerns.
• Bulk SaaS exports are efficient for large objects, while row-by-row APIs may be better for low-latency syncs or entities that lack export support.
• Raw landing should preserve source shape before transformation. Normalizing too early makes replay and vendor dispute analysis harder.

Choose based on source capability and the cost of missing a change. A campaign list may tolerate a daily full refresh. Revenue invoices need overlap and reconciliation. Deletion requests require tombstones; operational dashboards may need CDC or webhook-assisted sync.

Failure modes and troubleshooting

A common file failure is partial payload processing. The symptom is an unexpectedly low row count or a parser error near the end of a file. The cause is usually reading a file before the producer finished writing it, or treating a directory listing as a completion signal. Diagnose by comparing object modification times, file sizes, and manifest entries across retries. Correct it by requiring ready markers, immutable object names, or manifests with expected byte and row counts.

A common API failure is duplicate or missing records around page boundaries. Symptoms include counts that change between reruns, repeated ids in raw data, or gaps in updated_at ranges. Causes include offset pagination against a changing dataset, insufficient lookback, or checkpoint advancement before commit. Diagnose by logging request URLs, cursor tokens, response counts, minimum and maximum updated_at values, and checkpoint writes. Correct it by preferring cursor tokens, widening the overlap, and making the destination merge idempotent.

A database ingestion failure often appears as source performance degradation. Symptoms include slow application queries, lock waits, or elevated replica lag during extraction. Causes include scanning a hot primary table, missing indexes on watermark columns, or too-large batch windows. Diagnose with query plans, database wait metrics, and extraction timings. Correct it by reading from a replica, adding an index that matches the predicate, reducing window size, or moving to CDC when repeated range scans become too expensive.

A SaaS connector failure may show up as schema drift: a new custom field appears, a field changes type, or an enum gets a new value. The cause is usually product configuration by business users. Diagnose by comparing raw JSON samples and connector schema logs between runs. Correct it by landing semi-structured raw payloads, adding tolerant parsing, and alerting on fields that are required by downstream models.

Security, performance, and reliability implications

Ingestion code touches credentials and broad read permissions. Store API tokens, database passwords, and SaaS refresh tokens in a secret manager. Scope credentials to read-only access and the minimum objects. Validate file paths so a malformed manifest cannot read outside the expected prefix.

Performance comes from bounded work. Use partitioned file prefixes, API page sizes that respect rate limits, database predicates that use indexes, and raw destinations that can absorb retries without corrupting state. Reliability comes from replayability: raw batches should be immutable, run metadata should include source parameters and checksums where possible, and checkpoint changes should happen after durable writes. These details make an ingestion batch recoverable instead of unprovable.

Hands-on lab: build a local incremental ingest simulation

This lab simulates an API or SaaS source with updated_at fields and a checkpoint. Prerequisites are Python 3 and a shell. The script requests an overlapped window, writes raw JSONL, advances a checkpoint after writing, and verifies safe reruns.

from datetime import datetime, timedelta, timezone
import json
from pathlib import Path

RAW_DIR = Path("raw_lab")
CHECKPOINT = Path("checkpoint.txt")

def parse_ts(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

def load_checkpoint():
    if CHECKPOINT.exists():
        return parse_ts(CHECKPOINT.read_text().strip())
    return datetime(2026, 9, 7, 10, 0, tzinfo=timezone.utc)

def source_records():
    return [
        {"id": "a", "updated_at": "2026-09-07T09:55:00Z", "amount": 10},
        {"id": "b", "updated_at": "2026-09-07T10:03:00Z", "amount": 20},
        {"id": "c", "updated_at": "2026-09-07T10:07:00Z", "amount": 30},
    ]

def extract(records, start):
    return [record for record in records if parse_ts(record["updated_at"]) >= start]

def run():
    RAW_DIR.mkdir(exist_ok=True)
    checkpoint = load_checkpoint()
    start = checkpoint - timedelta(minutes=10)
    batch = extract(source_records(), start)
    raw_path = RAW_DIR / f"batch_{checkpoint.strftime('%Y%m%dT%H%M%S')}.jsonl"
    with raw_path.open("w", encoding="utf-8") as fh:
        for record in batch:
            fh.write(json.dumps(record, sort_keys=True) + "\n")
    next_cp = max([checkpoint] + [parse_ts(record["updated_at"]) for record in batch])
    CHECKPOINT.write_text(next_cp.isoformat(), encoding="utf-8")
    print(f"wrote {len(batch)} records to {raw_path}")
    print(f"checkpoint {next_cp.isoformat()}")

if __name__ == "__main__":
    run()

1. Save the script as a temporary local file named ingest_lab.py in a scratch directory.
2. Run python3 ingest_lab.py. The first run starts from a default checkpoint of 2026-09-07T10:00:00+00:00 and requests a 10 minute overlap.
3. Inspect raw_lab. The JSONL batch should include records a, b, and c because record a falls inside the overlap.
4. Run python3 ingest_lab.py again. The checkpoint should already be 2026-09-07T10:07:00+00:00, so the overlapped request starts at 09:57 and includes b and c again.
5. Verification: repeated records are expected at ingestion time. A downstream merge keyed by id would keep one current row per id. Cleanup: remove ingest_lab.py, checkpoint.txt, and the raw_lab directory.

The first run reports three records and a checkpoint ending in 10:07:00+00:00. The second writes two records because the overlap no longer reaches record a. Checkpointing protects against missed late data; idempotent loading prevents duplicate facts.

Assessment exercises

1. A vendor delivers hourly files to object storage but sometimes uploads large files slowly. Design a completion signal and explain how your ingestion job avoids reading partial data.
2. An API supports both offset pagination and cursor pagination. The underlying records can change while you scan. Which pagination mode should you choose, and what metadata should you log for replay?
3. A database table has updated_at but no soft delete column. The analytics team needs accurate deletion counts. Explain why a watermark extract is insufficient and propose an alternative.
4. A SaaS object has nested custom fields that business users can add without engineering review. How should the raw landing table and parser handle this without breaking every sync?
5. Given a 15 minute API lookback and a destination merge keyed by source id, explain why duplicates in raw data are acceptable but duplicate facts in a reporting table are not.

Summary

File ingestion is built around complete immutable batches. API ingestion is built around request windows, pagination, rate limits, and checkpoints. Database ingestion is built around snapshots, watermarks, or CDC depending on how precisely changes and deletes must be represented. SaaS ingestion adds product-shaped objects, bulk exports, custom fields, and business events. Across all four, land raw data with run metadata, advance checkpoints only after commit, and make downstream loads idempotent. That keeps batch pipelines explainable when a source is late, mutable, rate-limited, or changed outside the data team.