This part designs a complete API-driven ingestion layer with connectors, queues, lakehouse integration, governance hooks, and Python patterns for resilient, compliant pipelines powering downstream analytics and AI workloads.
Part 1 argued that sustainable data acquisition depends on official APIs, licensed data, and BYOD rather than brittle scraping, especially under modern privacy and AI regulations. Now, we’ll translate that strategic stance into concrete architecture suitable for enterprise-scale stacks (Azure, AWS, GCP, hybrid/on-prem).
Our goal: a universal ingestion layer that treats APIs, streaming, and batch sources uniformly, centralizes observability and governance, and emits AI-ready data products rather than ad hoc CSVs.
An API-driven data pipeline typically decomposes into several specialized components:
A modular architecture, as emphasized in universal ingestion frameworks, enables plug-and-play source connectors and sink adapters while preserving a common control plane.
This design enforces:
Below is a simplified Python ingestion pattern suitable for Azure Functions, AWS Lambda, or containerized workers. It assumes an official API with pagination and rate limits.
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone
import requests
from tenacity import (
retry,
retry_if_exception_type,
stop_after_attempt,
wait_base,
)
Rate-limit handling
class RateLimitError(Exception):
def __init__(self, retry_after=None):
self.retry_after = retry_after
super().__init__("API rate limit exceeded")
class WaitRetryAfterOrExponential(wait_base):
def __call__(self, retry_state):
exception = retry_state.outcome.exception()
if isinstance(exception, RateLimitError):
if exception.retry_after is not None:
return exception.retry_after
attempt = retry_state.attempt_number
return min(2 ** attempt, 30)
Retry-After parsing
def parse_retry_after(value):
if not value:
return None
try:
return max(0, int(value))
except ValueError:
try:
retry_at = parsedate_to_datetime(value)
now = datetime.now(timezone.utc)
return max(0, int((retry_at - now).total_seconds()))
except (TypeError, ValueError, OverflowError):
return None
API request with retries
@retry(
retry=retry_if_exception_type(
(RateLimitError, requests.RequestException)
),
wait=WaitRetryAfterOrExponential(),
stop=stop_after_attempt(5),
reraise=True,
)
def call_api(endpoint, params=None):
response = requests.get(
f"{API_BASE}/{endpoint}",
headers=build_headers(),
params=params,
timeout=30,
)
if response.status_code == 429:
retry_after = parse_retry_after(
response.headers.get("Retry-After")
)
raise RateLimitError(retry_after=retry_after)
response.raise_for_status()
return response.json()
Key practices illustrated here are aligned with 2026 best practices for API ingestion: authentication handling, exponential backoff with jitter, respecting rate limits, pagination, and incremental sync using timestamps or tokens.
API payloads must be governed by formal data contracts (for example, Open Data Contract patterns or tooling like Great Expectations). Validate every incoming payload at the ingestion boundary against a formal schema (JSON Schema, Avro, or Protobuf). Conformant payloads are written into the bronze/raw lakehouse; schema violations and semantic quality failures are atomically routed to a dead-letter queue (DLQ) with full context (payload, feed id, ingestion job id, checksum, timestamp) for automated alerting and manual triage.
Downstream processing enforces schema evolution rules (backward/forward compatibility), runs deterministic and probabilistic enrichment (entity resolution, ontology mapping), and promotes validated records to curated zones only after quality gates pass. This enforces contract-first ingestion, reduces silent data drift, and makes provenance and auditability tractable.
To achieve sustainable ingestion, governance must not be an afterthought:
Retention, deletion, and usage restrictions should remain linked to the originating API contract, license, and applicable regulations. This ensures that access decisions are policy-driven, sensitive values are protected at query time, and downstream AI pipelines cannot consume data beyond its contractual or regulatory entitlement.
Implementing these governance hooks at design time is what differentiates “yet another ingestion pipeline” from a sustainable data acquisition platform.
In Part 3, we will extend this architecture to licensed datasets and BYOD ingestion, and show how to model data products and operating agreements around them.
