Part 2 provides practical code patterns, schemas and flow designs to implement rate-limited, consent-aware, auditable data acquisition pipelines in enterprise AI and analytics environments.
Part 1 introduced the architectural and governance foundations for compliant data acquisition, including consent lineage, provenance services and rate-limited acquisition orchestrators. This second part focuses on implementation-level patterns and code snippets that teams can adapt to build contract-aware, enterprise-grade data harvesting pipelines.
The examples below assume a modern cloud-native stack, but the patterns are portable to most languages and platforms as long as they preserve consent, contracts, provenance and rate limiting as first-class concerns.
A core building block is a shared acquisition client that enforces rate limits, honours robots.txt and uses ToS-compliant access paths. Instead of allowing each team to write their own HTTP loop, centralising this logic reduces compliance risk and ensures consistent handling of limits and errors.
Key behaviours
Example: Python-style compliant acquisition client
import time
import logging
from typing import Dict, Any, Iterator
import requests
logger = logging.getLogger("compliant_acq")
class RateLimiter:
"""
Local, single-process sliding-window rate limiter.
Production note:
This in-memory implementation is suitable only for a single worker or
local development. In a distributed deployment with multiple container
workers, replace it with a shared Redis-backed Token Bucket or Leaky
Bucket algorithm so that all acquisition workers enforce the same
source-specific quota.
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.request_timestamps: list[float] = []
def acquire(self)->None:
now = time.time()
self.request_timestamps = [
for timestamp in self.request_timestamps
if now - timestamp < self.window_seconds
]
if len(self.request_timestamps) >= self.max_requests:
sleep_for = self.window_seconds - (now - self.request_timestamps[0])
time.sleep(max(sleep_for, 0))
self.request_timestamps.append(time.time())
class CompliantClient:
def __init__(
self,
base_url: str,
rate_limiter: RateLimiter,
legal_basis: Dict[str, Any],
provenance_logger
):
self.base_url = base_url
self.rate_limiter = rate_limiter
self.legal_basis = legal_basis # consent_id / licence_id
self.provenance_logger = provenance_logger
self.session = requests.Session()
def get_paginated(
self,
path: str,
params: Dict[str, Any] = None
) -> Iterator[Dict[str, Any]]:
page = 1
while True:
self.rate_limiter.acquire()
url = f"{self.base_url}{path}"
response = self.session.get(
url,
params={**(params or {}), "page": page},
headers={"X-Legal-Basis": self.legal_basis["type"]}
)
self._handle_rate_limit_response(response)
data = response.json()
self._log_provenance("INGEST_PAGE", url, params, data)
if not data.get("items"):
break
yield data
page += 1
def _handle_rate_limit_response(self, response):
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", "5"))
time.sleep(retry_after)
response.raise_for_status()
def _log_provenance(
self,
event_type: str,
url: str,
params: Dict[str, Any],
payload: Dict[str, Any]
):
self.provenance_logger.log_event(
event_type=event_type,
url=url,
params=params,
payload_hash=self._hash_payload(payload),
legal_basis=self.legal_basis,
) import hashlib
import json
def _hash_payload(self, payload: Dict[str, Any]) -> str:
"""
Generate a deterministic SHA-256 digest for an acquisition payload.
JSON keys are sorted and separators are fixed so that logically
identical payloads produce the same digest across Python runtimes,
worker processes and environments.
"""
canonical_payload = json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
return hashlib.sha256(canonical_payload).hexdigest()
The local RateLimiter is intentionally simple and should be treated as a reference implementation. In production, multiple acquisition workers, Kubernetes pods or serverless functions must share a common quota state. A Redis-backed Token Bucket or Leaky Bucket limiter should be used, keyed by source system, API endpoint, licence ID or service identity. This prevents each container from independently consuming the full allowance and accidentally exceeding the source’s published rate limit.
Payload hashing must also be deterministic. Using sort_keys=True and fixed separators produces a canonical JSON representation, ensuring that the same logical payload generates the same SHA-256 digest regardless of Python version, serialisation order or runtime environment. This is essential for reliable provenance chains, duplicate detection and tamper-evident audit logs.
This example illustrates how rate limiting, legal basis propagation and provenance logging can be baked into a reusable client rather than treated as afterthoughts. In production, you would extend this to parse specific rate limit headers, integrate with robots.txt parsing and honour API-specific ToS constraints.
Provenance logging should be implemented as middleware around acquisition and transformation steps so that every material event is recorded consistently. Many data platforms already support event hooks (e.g. ingestion completed, job finished), which can be wired to a provenance service.
Example: Provenance logging interface
from abc import ABC, abstractmethod
from typing import Any, Dict
from datetime import datetime, timezone
import uuid
class ProvenanceLogger(ABC):
"""Interface for emitting immutable, auditable provenance events."""
@abstractmethod
def log_event(
self,
event_type: str,
dataset_id: str,
metadata: Dict[str, Any]
) -> None:
...
class JsonProvenanceLogger(ProvenanceLogger):
"""
Emits provenance events to an append-only observability sink.
The sink should be an OpenTelemetry-compatible exporter or an
append-only event stream such as Apache Kafka or AWS Kinesis. This
ensures provenance events are durable, queryable and observable across
distributed data and AI platforms.
"""
def __init__(self, sink):
# Sink can be a Kafka topic, append-only file, or database table.
self.sink = sink
def log_event(
self,
event_type: str,
dataset_id: str,
metadata: Dict[str, Any]
) -> None:
event = {
"event_id": str(uuid.uuid4()),
"event_type": event_type,
"dataset_id": dataset_id,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
**metadata,
}
self.sink.write(event)
The timestamp is generated using datetime.now(timezone.utc).isoformat(), which is timezone-aware and replaces the deprecated datetime.utcnow() pattern in Python 3.12 and later. This avoids ambiguity between naive and UTC timestamps and ensures consistent event ordering across services and regions.
The provenance sink should not be treated as a local log file in production. It should feed directly into an OpenTelemetry-compliant exporter or an append-only event stream such as Apache Kafka or AWS Kinesis. This aligns provenance logging with modern cloud observability standards, enabling durable retention, correlation with traces and metrics, near-real-time alerting and audit-ready replay of data lineage events.
By standardising the event schema and sink, organisations can build lineage visualisers, consent provenance reports and audit packs without modifying individual jobs. This aligns with guidance that provenance systems should capture events at the point of action, not through retroactive reconstruction.
Consent lineage records act as the evidentiary spine for privacy and AI compliance, capturing every consent state change and linking it cryptographically. A consent provenance report is then a summarised view of that lineage, generated for a specific subject or dataset for regulatory reporting or principal review.
Simplified consent lineage event schema
{
"consent_root_id": "consent_root_123",
"event_id": "evt_c_0001",
"event_type": "ISSUED",
"subject_id": "user_456",
"scope": [
"analytics",
"ai_training_conversational"
],
"notice_version": "v5.2",
"issued_at": "2026-03-01T10:00:00Z",
"issued_by": "cmp_web_portal",
"prev_event_hash": null,
"event_hash": "3baf9c..."
}
Subsequent scope reductions, renewals or revocations append new events with prev_event_hash pointing to the previous event, forming a tamper-evident chain. These logs must be exportable in standard formats so that cross-platform audits can reconstruct lineage without access to proprietary systems.
Modern systems use this flow to generate consent provenance reports summarising all consent events, current scope and withdrawal handling for each subject or dataset. This supports both regulatory reporting (e.g. for privacy regulators) and principal reviews when individuals request explanations of how their data is being used.
To make this performant at enterprise scale, teams should adopt immutable graph modelling for consent. The append-only event log remains the authoritative source of truth, preserving the raw, tamper-evident chain of consent issuance, modification, delegation and revocation events. However, querying complex consent chains across millions of users during a data subject request should not require replaying the entire event history at audit time.
Instead, consent events should be continuously projected into a graph structure or relational index. Nodes can represent data subjects, consent records, purposes, datasets, processing activities and downstream AI assets, while edges capture relationships such as granted consent for, derived from, shared with and revoked by. This projection enables low-latency traversal of consent relationships, rapid identification of affected datasets and downstream systems, and efficient generation of audit-ready consent provenance reports.
In practice, the event log provides immutability and evidential integrity, while the graph or relational projection provides operational query performance. Together, they allow organisations to answer DSRs and regulator enquiries quickly without sacrificing the integrity of the underlying consent history.
A dataset registry is a structured inventory of all datasets, including their origin, lawful basis, licences, data stewards and version history. For AI workloads, registries need to track which datasets are used for training, validation and production inference, as well as their risk profiles.
Example: Dataset registry table (conceptual)
This contains_synthetic_data field distinguishes genuine source records from synthetic, generated or augmented data. This is important for bias assessment, quality assurance, licensing review and transparency in AI training documentation.
The model_card_ids field creates a bidirectional compliance link between datasets and the models trained on them. When a consent is revoked, a licence expires or a dataset is found to be non-compliant, the registry can immediately identify affected model versions, trigger impact assessment and update the relevant model cards with data provenance and limitation information.
Together, these fields strengthen AI training governance by connecting source data, lawful basis, licensing state, synthetic-data disclosure and downstream model accountability in a single auditable registry.
Checklist-style guidance for responsible AI data acquisition emphasises that teams maintain source-of-origin documentation, contributor consent records and licensing chain-of-custody, all tied back to such registry entries. Standards such as ISO 42001 explicitly expect digital, version-controlled inventories and named data stewards for AI datasets.
Data security posture management (DSPM) and discovery tools increasingly support continuous data classification, RoPA (Records of Processing Activities) generation and DSR evidence logs enriched with provenance. When integrated with acquisition pipelines, they can automate parts of audit reporting by aggregating lineage and access logs.
Example: DSR fulfilment with integrated provenance
The DSPM or lineage service can automatically gather: data locations, consent records, provenance events and access logs across infrastructures such as S3 and CloudTrail. This enables organisations to produce comprehensive responses to subject access or deletion requests, including evidence of how and where data has flowed into AI systems.
This assurance must include continuous PII re-scanning. Even when a third-party source asserts that a dataset has been scrubbed of personal data, internal DSPM scanners should run automated checks at the lakehouse landing zone before the data is promoted into curated AI feature stores. These scans should detect direct identifiers, quasi-identifiers, free-text PII, credentials, payment data and other sensitive attributes that may have been missed by the supplier’s own redaction process.
If potential PII is detected, the ingestion workflow should quarantine the affected records or partitions, notify the data steward and privacy team, and prevent downstream propagation into feature stores, vector indexes or model training pipelines until the issue is reviewed and resolved. Every scan result, remediation decision and release approval should be recorded in the provenance and audit trail.
In this model, third-party claims of “PII-free” data are treated as inputs to risk assessment not as substitutes for independent verification. Continuous re-scanning ensures that privacy controls remain effective as schemas evolve, new records arrive and data is transformed across the AI lifecycle.
To keep compliant data acquisition operationally realistic, governance checks must be integrated into CI/CD for data pipelines rather than handled only in manual reviews. A typical pattern is to implement “governance gates” that block deployments or schedule runs when required metadata or licences are missing.
Example: policy-as-code gate (pseudo-code)
policy:
name: ensure_dataset_metadata_complete
applies_to:
- airflow_dag: ingest_partner_crm
rules:
- field: dataset_id
required: true
- field: legal_basis_type
required: true
- field: licence_id
required_if:
origin_type: [partner, vendor]
- field: consent_evidence_ref
required_if:
legal_basis_type: consent
- field: provenance_stream_id
required: true
- field: data_steward
required: true
Pipeline deployment and execution engines can check these policies before enabling new acquisition jobs. This enforces the ISO 42001 expectation that no data enters AI systems unless licensing, consent and retention parameters have been captured and validated at onboarding.
Enterprise AI teams often purchase datasets from providers who claim to supply fully provenance-traced, rights-cleared assets. Due diligence should verify that providers can supply manifest files, consent logs, licence metadata and PII audit reports upfront.
A core technical control is cryptographic manifest verification. Every third-party delivery should include a signed cryptographic manifest listing each file or data partition, its SHA-256 checksum, byte size, schema version, licence reference and delivery identifier. The vendor signs the manifest using its private key, while the enterprise verifies the signature using the vendor’s published public key or certificate.
The ingestion pipeline must validate this manifest before unzipping, extracting or ingesting any asset into the lakehouse. It should recompute SHA-256 checksums for every delivered file, compare them with the manifest, verify the manifest signature and confirm that the licence and delivery identifiers are active and expected. Any checksum mismatch, missing file, unsigned manifest, expired certificate or unrecognised delivery should cause the delivery to be quarantined rather than ingested.
Typically, compliant providers deliver:
Enterprises should insist that these artefacts are ingested into their own dataset registry and provenance systems rather than treated as static attachments in procurement systems. This avoids future disputes about whether the organisation genuinely had rights to use the data in specific AI training scenarios, and provides cryptographic evidence that the delivered assets have not been altered in transit or after packaging.
Bringing the patterns together, an end-to-end compliant acquisition flow for a new AI training dataset might look like this
This flow operationalises the idea that compliant data acquisition is a lifecycle capability involving business owners, legal teams, stewards, engineers and AI practitioners. Each step produces evidence that can be surfaced quickly when auditors, customers or regulators demand transparency about AI training data.
When treated as a one-off project, compliant data acquisition inevitably degrades as new sources, models and regulatory expectations emerge. By contrast, treating it as a platform product with APIs, shared clients, provenance services, dataset registries and self-service reporting enables organisations to scale AI confidently while meeting evolving compliance expectations.
The patterns and code examples in this two-part series provide a starting point for designing such a platform, but they should be interpreted as a baseline rather than a ceiling: organisations should evolve their controls as standards like ISO 42001, AI-specific regulations and privacy laws continue to mature.
