PeopleStrong Integration Engineer
Complete Interview Preparation Guide — 85+ Scenario-Based Questions
1 API to API Integration
You build a pipeline with three phases: Extract (GET from source), Transform (map fields, clean data), and Load (POST to target). The key is to never assume both APIs have the same schema. You read from the source, convert field names and data types to match the target's contract, handle pagination on the source side, and batch-insert on the target side. Always log every step for auditability.
import requests
def transfer_data(source_url, target_url, headers_src, headers_tgt):
# Extract
response = requests.get(source_url, headers=headers_src, timeout=30)
response.raise_for_status()
records = response.json()["data"]
# Transform
transformed = []
for r in records:
transformed.append({
"employee_id": r["emp_code"],
"full_name": f"{r['first_name']} {r['last_name']}",
"email": r["email"].lower().strip(),
})
# Load
resp = requests.post(target_url, json={"records": transformed}, headers=headers_tgt, timeout=60)
resp.raise_for_status()
return resp.json()
Offset-based: Use page/offset+limit params. Simple but slow for large datasets (DB scans all previous rows). Cursor-based: Server returns an opaque cursor token; you pass it in the next request. Best for real-time feeds. Keyset-based: Use the last record's ID/timestamp to fetch the next page (WHERE id > last_id). Most performant for sorted data. Always loop until the API signals no more data (empty results, no next cursor, or count < limit).
# Offset-based pagination
def fetch_all_offset(url, headers):
all_records, page = [], 1
while True:
resp = requests.get(url, params={"page": page, "limit": 100}, headers=headers)
data = resp.json()["data"]
if not data:
break
all_records.extend(data)
page += 1
return all_records
# Cursor-based pagination
def fetch_all_cursor(url, headers):
all_records, cursor = [], None
while True:
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
resp = requests.get(url, params=params, headers=headers)
body = resp.json()
all_records.extend(body["data"])
cursor = body.get("next_cursor")
if not cursor:
break
return all_records
# Keyset-based pagination
def fetch_all_keyset(url, headers):
all_records, last_id = [], 0
while True:
resp = requests.get(url, params={"after_id": last_id, "limit": 100}, headers=headers)
data = resp.json()["data"]
if not data:
break
all_records.extend(data)
last_id = data[-1]["id"]
return all_records
When you get HTTP 429 (Too Many Requests), check the Retry-After header for how long to wait. Implement exponential backoff: wait 1s, 2s, 4s, 8s, etc. with jitter (random extra delay to avoid thundering herd). Also proactively respect rate limits by throttling your requests — if the API allows 100 req/min, space your calls to ~1 per 600ms. Use a token bucket or leaky bucket algorithm for smooth rate control.
import time, random
def api_call_with_retry(url, headers, max_retries=5):
for attempt in range(max_retries):
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
return resp.json()
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
jitter = random.uniform(0, 1)
wait = retry_after + jitter
print(f"Rate limited. Waiting {wait:.1f}s (attempt {attempt+1})")
time.sleep(wait)
else:
resp.raise_for_status()
raise Exception("Max retries exceeded")
A 500 error means server-side failure. First, don't panic and don't retry infinitely. Implement: (1) Retry with exponential backoff (maybe transient). (2) Log the exact page/offset where failure happened so you can resume. (3) If retry fails 3 times, save progress (last successful page) to a checkpoint file or DB. (4) Alert the team. (5) Resume from checkpoint when source recovers. Never restart from page 1 — that wastes time and may cause duplicates.
import json
from pathlib import Path
def fetch_with_checkpoint(url, headers, checkpoint_file="checkpoint.json"):
# Resume from checkpoint if exists
checkpoint = Path(checkpoint_file)
last_page = 1
all_records = []
if checkpoint.exists():
state = json.loads(checkpoint.read_text())
last_page = state["last_successful_page"] + 1
all_records = state.get("records", [])
page = last_page
while True:
try:
resp = requests.get(url, params={"page": page, "limit": 100}, headers=headers)
resp.raise_for_status()
data = resp.json()["data"]
if not data:
break
all_records.extend(data)
# Save checkpoint after each successful page
checkpoint.write_text(json.dumps({"last_successful_page": page}))
page += 1
except requests.exceptions.HTTPError as e:
if e.response.status_code == 500:
print(f"Server error at page {page}. Saved checkpoint.")
raise
return all_records
API Key: Simplest — pass key in header or query param. OAuth2 Client Credentials: Exchange client_id + client_secret for an access token, use token in Authorization header, refresh when expired. JWT: Generate a signed token with claims (issuer, expiry, scope), send in Bearer header. Always store credentials in environment variables or a vault, never in code. Handle token expiry gracefully — refresh before it expires, not after you get a 401.
import os
# API Key auth
headers_apikey = {"X-API-Key": os.environ["SOURCE_API_KEY"]}
# OAuth2 Client Credentials
def get_oauth_token(token_url, client_id, client_secret):
resp = requests.post(token_url, data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
})
resp.raise_for_status()
token_data = resp.json()
return token_data["access_token"], token_data["expires_in"]
# Usage with token refresh
class TokenManager:
def __init__(self, token_url, client_id, client_secret):
self.token_url = token_url
self.client_id = client_id
self.client_secret = client_secret
self.token = None
self.expires_at = 0
def get_headers(self):
if time.time() >= self.expires_at - 60: # Refresh 60s before expiry
self.token, expires_in = get_oauth_token(
self.token_url, self.client_id, self.client_secret)
self.expires_at = time.time() + expires_in
return {"Authorization": f"Bearer {self.token}"}
Chunk the data into batches of the target's max size. Use Python's list slicing or a generator to split 10,000 records into 100 batches of 100. Send each batch sequentially (or in parallel with concurrency limits). Track which batches succeeded and which failed. If a batch fails, retry just that batch, not the entire dataset. This is called batch processing and is fundamental in integration work.
def chunk_list(data, size):
for i in range(0, len(data), size):
yield data[i:i + size]
def batch_upload(records, target_url, headers, batch_size=100):
results = {"success": 0, "failed": 0, "failed_batches": []}
for i, batch in enumerate(chunk_list(records, batch_size)):
try:
resp = requests.post(target_url, json={"records": batch}, headers=headers)
resp.raise_for_status()
results["success"] += len(batch)
print(f"Batch {i+1}: {len(batch)} records uploaded")
except Exception as e:
results["failed"] += len(batch)
results["failed_batches"].append({"batch": i+1, "error": str(e)})
return results
Duplicates are the #1 data quality issue in integrations. Solutions: (1) Use a natural key (employee_id, email) and check before insert. (2) Use upsert (INSERT ON CONFLICT UPDATE) at the target. (3) Maintain a dedup set of already-processed IDs. (4) Make target API idempotent — same request twice produces same result. (5) Use a hash of the record to detect if content actually changed. Always prefer idempotent design over pre-checking, as it handles race conditions.
# Dedup before sending to target
def dedup_records(records, key_field="employee_id"):
seen = set()
unique = []
for r in records:
key = r[key_field]
if key not in seen:
seen.add(key)
unique.append(r)
else:
print(f"Duplicate skipped: {key}")
return unique
# SQL upsert at target (PostgreSQL)
upsert_sql = """
INSERT INTO employees (emp_id, name, email, department)
VALUES (%s, %s, %s, %s)
ON CONFLICT (emp_id) DO UPDATE SET
name = EXCLUDED.name,
email = EXCLUDED.email,
department = EXCLUDED.department,
updated_at = NOW()
"""
Schema drift is a silent killer. Defenses: (1) Schema validation at extraction — validate every response against an expected schema (use Pydantic, JSON Schema). (2) Fail fast — if a required field is missing, stop and alert rather than inserting nulls. (3) Version detection — check API version header on each call. (4) Flexible mapping — use a config-driven field mapping so changes don't require code changes. (5) Monitoring — alert on null rates, new fields, type mismatches.
from pydantic import BaseModel, ValidationError
from typing import Optional
class EmployeeSchema(BaseModel):
emp_code: str
first_name: str
last_name: str
email: str
department: Optional[str] = None
def validate_records(records):
valid, invalid = [], []
for r in records:
try:
validated = EmployeeSchema(**r)
valid.append(validated.model_dump())
except ValidationError as e:
invalid.append({"record": r, "errors": e.errors()})
return valid, invalid
model_config = ConfigDict(extra='allow') lets new fields pass through without breaking.Exponential backoff means each retry waits longer: 1s, 2s, 4s, 8s, 16s. Add jitter (random 0-1s) to prevent synchronized retries. Set a max retries cap (usually 3-5). Only retry on transient errors (429, 500, 502, 503, 504, connection timeouts). Never retry on 400, 401, 403, 404 — those are permanent failures. Use the tenacity library in production for robust retry logic.
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=1, max=60),
retry=retry_if_exception_type(requests.exceptions.RequestException),
before_sleep=lambda retry_state: print(
f"Retry {retry_state.attempt_number}, waiting {retry_state.next_action.sleep:.1f}s"
),
)
def resilient_api_call(url, headers):
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code in (429, 500, 502, 503, 504):
resp.raise_for_status() # Will trigger retry
return resp.json()
Structured logging is essential. Log: (1) Start/end timestamps. (2) Records fetched, transformed, loaded. (3) Errors with full context (URL, status code, response body). (4) Duration per phase. Use structured logging (JSON) so logs are searchable. For monitoring: set up alerts on failure count, duration spikes, and record count anomalies. Use correlation IDs to trace a single sync run across all logs.
import logging, uuid
from datetime import datetime
logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger(__name__)
def sync_with_logging(source_url, target_url):
run_id = str(uuid.uuid4())[:8]
start = datetime.now()
logger.info(f"[{run_id}] Sync started at {start}")
try:
# Extract
records = fetch_all_offset(source_url, headers={})
logger.info(f"[{run_id}] Extracted {len(records)} records")
# Transform
transformed = [transform(r) for r in records]
logger.info(f"[{run_id}] Transformed {len(transformed)} records")
# Load
result = batch_upload(transformed, target_url, headers={})
duration = (datetime.now() - start).total_seconds()
logger.info(f"[{run_id}] Sync complete in {duration:.1f}s | "
f"Success: {result['success']}, Failed: {result['failed']}")
except Exception as e:
logger.error(f"[{run_id}] Sync FAILED: {e}")
raise
An operation is idempotent if executing it multiple times produces the same result as executing it once. In integration, this means if your sync runs twice accidentally, the data should be the same — no duplicates, no incorrect updates. Implement via: (1) Use unique keys + upsert instead of insert. (2) Send an idempotency key (unique ID per request) so the target can recognize duplicates. (3) Design target endpoints to be naturally idempotent (PUT over POST where possible). This is critical because network failures cause retries, and retries without idempotency cause data corruption.
# Idempotent request with idempotency key
import hashlib
def idempotent_post(url, record, headers):
# Generate deterministic key from record content
key = hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest()
headers["Idempotency-Key"] = key
resp = requests.post(url, json=record, headers=headers)
return resp.json()
# Target API checks: if key exists, return cached response
# If key is new, process and store the key
Never let one batch failure abort the entire sync. Strategy: (1) Process all batches independently. (2) Track success/failure per batch. (3) Failed records go to a dead letter queue (DLQ) or error table. (4) After the main run, retry the failed batches once. (5) If still failing, generate a failure report with record details and error messages. (6) Alert the support team. The successful records should NOT be rolled back — partial success is better than total failure in most integration scenarios.
def process_with_dlq(records, target_url, headers, batch_size=100):
dead_letter_queue = []
success_count = 0
for i, batch in enumerate(chunk_list(records, batch_size)):
try:
resp = requests.post(target_url, json={"records": batch}, headers=headers)
resp.raise_for_status()
success_count += len(batch)
except Exception as e:
for record in batch:
dead_letter_queue.append({
"record": record,
"error": str(e),
"batch": i + 1,
"timestamp": datetime.now().isoformat(),
})
# Retry DLQ once
if dead_letter_queue:
print(f"Retrying {len(dead_letter_queue)} failed records...")
for item in dead_letter_queue[:]:
try:
requests.post(target_url, json={"records": [item["record"]]}, headers=headers)
dead_letter_queue.remove(item)
success_count += 1
except:
pass # Still failing, keep in DLQ
return {"success": success_count, "dlq": dead_letter_queue}
Strategies to handle slow source APIs: (1) Parallel requests — fetch multiple pages concurrently using asyncio or ThreadPoolExecutor. (2) Request only what you need — use field filtering (fields=id,name,email) to reduce payload. (3) Compression — send Accept-Encoding: gzip header. (4) Caching — cache responses that don't change often (departments, locations). (5) Incremental sync — fetch only changed records using modified_since parameter. (6) Connection pooling — reuse HTTP connections with requests.Session().
from concurrent.futures import ThreadPoolExecutor, as_completed
def parallel_fetch(base_url, headers, total_pages, max_workers=5):
all_records = []
def fetch_page(page):
session = requests.Session()
session.headers.update(headers)
session.headers["Accept-Encoding"] = "gzip"
resp = session.get(base_url, params={"page": page, "limit": 100,
"fields": "id,name,email,dept"})
return resp.json()["data"]
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(fetch_page, p): p for p in range(1, total_pages + 1)}
for future in as_completed(futures):
all_records.extend(future.result())
return all_records
Timezone bugs are subtle and devastating. Rules: (1) Always store in UTC internally. Convert to local time only for display. (2) When reading from source, check if timestamps have timezone info. If not, ask which timezone they assume. (3) Use ISO 8601 format (2024-01-15T10:30:00Z) for all API communication. (4) When comparing timestamps across systems, convert both to UTC first. (5) Be aware of DST (Daylight Saving Time) — India (IST) doesn't have DST but US/Europe do.
from datetime import datetime
from zoneinfo import ZoneInfo
def normalize_to_utc(timestamp_str, source_tz="Asia/Kolkata"):
# Source sends: "2024-01-15 10:30:00" (IST, no tz info)
naive = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S")
local = naive.replace(tzinfo=ZoneInfo(source_tz))
utc = local.astimezone(ZoneInfo("UTC"))
return utc.isoformat() # "2024-01-15T05:00:00+00:00"
# Always compare in UTC
source_time = normalize_to_utc("2024-01-15 10:30:00", "Asia/Kolkata")
target_time = normalize_to_utc("2024-01-15 05:00:00", "UTC")
# These are the same moment!
Plan for target downtime: (1) Circuit breaker — after N consecutive failures, stop trying and wait. (2) Queue the data — extract from source, save to a local queue (file, Redis, DB), and process when target comes back. (3) Scheduled retry — attempt re-sync every 15 minutes. (4) Alert — notify team if target has been down > 1 hour. (5) SLA tracking — record downtime for vendor accountability. Never lose extracted data just because the target is down.
import json
from pathlib import Path
def sync_with_queue(source_url, target_url, queue_dir="./sync_queue"):
Path(queue_dir).mkdir(exist_ok=True)
# Extract from source (always works)
records = fetch_all_offset(source_url, headers={})
# Try to load to target
try:
result = batch_upload(records, target_url, headers={})
print(f"Sync success: {result['success']} records")
except requests.exceptions.ConnectionError:
# Target is down - queue the data
queue_file = Path(queue_dir) / f"pending_{datetime.now():%Y%m%d_%H%M%S}.json"
queue_file.write_text(json.dumps(records))
print(f"Target down! Queued {len(records)} records to {queue_file}")
send_alert("Target API down during scheduled sync")
def process_queue(target_url, queue_dir="./sync_queue"):
for f in sorted(Path(queue_dir).glob("pending_*.json")):
records = json.loads(f.read_text())
try:
batch_upload(records, target_url, headers={})
f.rename(f.with_suffix(".done"))
except:
break # Still down, stop retrying
A Dead Letter Queue (DLQ) captures records that fail processing after all retries are exhausted. It preserves the original data, the error reason, and metadata (timestamp, attempt count) so that support can investigate and reprocess. Implement as: (1) A database table with columns for record data, error, attempt_count, created_at, status. (2) A retry mechanism that picks up DLQ items periodically. (3) A dashboard to view and manually reprocess failed records. (4) Max retry limit per record to prevent infinite loops.
import sqlite3
def init_dlq(db_path="dlq.db"):
conn = sqlite3.connect(db_path)
conn.execute("""CREATE TABLE IF NOT EXISTS dead_letter_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
record_data TEXT NOT NULL,
error_message TEXT,
source_system TEXT,
target_system TEXT,
attempt_count INTEGER DEFAULT 1,
max_attempts INTEGER DEFAULT 3,
status TEXT DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_retry_at TIMESTAMP
)""")
conn.commit()
return conn
def add_to_dlq(conn, record, error, source, target):
conn.execute(
"INSERT INTO dead_letter_queue (record_data, error_message, source_system, target_system) VALUES (?, ?, ?, ?)",
(json.dumps(record), str(error), source, target)
)
conn.commit()
def retry_dlq(conn, target_url, headers):
rows = conn.execute(
"SELECT id, record_data, attempt_count, max_attempts FROM dead_letter_queue WHERE status='pending' AND attempt_count < max_attempts"
).fetchall()
for row_id, data, attempts, max_att in rows:
try:
record = json.loads(data)
requests.post(target_url, json=record, headers=headers).raise_for_status()
conn.execute("UPDATE dead_letter_queue SET status='resolved' WHERE id=?", (row_id,))
except Exception as e:
new_status = "failed" if attempts + 1 >= max_att else "pending"
conn.execute("UPDATE dead_letter_queue SET attempt_count=?, status=?, last_retry_at=CURRENT_TIMESTAMP WHERE id=?",
(attempts + 1, new_status, row_id))
conn.commit()
Reconciliation ensures data integrity after sync. Checks: (1) Count match — source records = target records. (2) Checksum match — hash of key fields should be identical. (3) Sample validation — randomly pick 10 records and compare field by field. (4) Aggregate validation — SUM(salary) in source should equal SUM(salary) in target. (5) Null check — no unexpected nulls in required fields. Run reconciliation after every sync and alert on mismatches.
def reconcile(source_records, target_records, key_field="employee_id"):
report = {"source_count": len(source_records), "target_count": len(target_records)}
source_keys = {r[key_field] for r in source_records}
target_keys = {r[key_field] for r in target_records}
report["missing_in_target"] = source_keys - target_keys
report["extra_in_target"] = target_keys - source_keys
report["count_match"] = len(source_keys) == len(target_keys)
# Sample field-level comparison
source_map = {r[key_field]: r for r in source_records}
target_map = {r[key_field]: r for r in target_records}
mismatches = []
for key in list(source_keys & target_keys)[:10]: # Sample 10
for field in source_map[key]:
if source_map[key].get(field) != target_map[key].get(field):
mismatches.append({"key": key, "field": field,
"source": source_map[key].get(field),
"target": target_map[key].get(field)})
report["mismatches"] = mismatches
return report
Polling: Your system periodically calls the source API to check for new data (e.g., every 5 min). Simple but wasteful — most calls return nothing. Webhook: Source system calls YOUR API when something changes (event-driven). Real-time, efficient, but you need an exposed endpoint. Use polling when: source doesn't support webhooks, you need batch processing, data changes are frequent. Use webhooks when: you need real-time updates, source supports them, you can expose an endpoint. Best practice: webhook with polling as fallback.
# Polling approach
import schedule
def poll_for_changes():
last_sync = load_last_sync_time()
records = requests.get(
"https://api.source.com/employees",
params={"modified_since": last_sync}
).json()
if records:
process_records(records)
save_last_sync_time(datetime.utcnow())
schedule.every(5).minutes.do(poll_for_changes)
# Webhook approach (FastAPI)
from fastapi import FastAPI, Request
app = FastAPI()
@app.post("/webhook/employee-update")
async def handle_webhook(request: Request):
payload = await request.json()
# Verify webhook signature
if not verify_signature(request.headers, payload):
return {"error": "Invalid signature"}, 401
process_records(payload["data"])
return {"status": "received"}
API versioning strategies: (1) URL versioning: /api/v1/employees vs /api/v2/employees. (2) Header versioning: Accept: application/vnd.api+json;version=2. (3) Query param: ?version=2. Best practices for integration: Always pin to a specific version. Subscribe to the API provider's changelog. Test against new versions in staging before switching. Use an adapter pattern — isolate API-specific code so version changes only affect one file.
# Adapter pattern for API versioning
class EmployeeAPIv1:
def get_employees(self, modified_since=None):
resp = requests.get("https://api.example.com/v1/employees",
params={"updated_after": modified_since})
data = resp.json()
return [{"id": r["emp_code"], "name": r["fullname"]} for r in data]
class EmployeeAPIv2:
def get_employees(self, modified_since=None):
resp = requests.get("https://api.example.com/v2/employees",
params={"modified_since": modified_since})
data = resp.json()["results"] # v2 wraps in "results"
return [{"id": r["employee_id"], "name": f"{r['first']} {r['last']}"} for r in data]
# Config-driven version selection
API_VERSION = os.environ.get("EMPLOYEE_API_VERSION", "v2")
api = EmployeeAPIv2() if API_VERSION == "v2" else EmployeeAPIv1()
Essential headers: (1) Authorization: Bearer token or API key. (2) Content-Type: application/json (tells server what you're sending). (3) Accept: application/json (tells server what you expect). (4) Accept-Encoding: gzip (enables compression). (5) User-Agent: identifies your client. (6) X-Request-ID: unique ID for tracing. (7) Retry-After: response header telling you when to retry. (8) ETag / If-None-Match: for caching. (9) Idempotency-Key: for safe retries on POST.
headers = {
"Authorization": "Bearer eyJhbGciOi...",
"Content-Type": "application/json",
"Accept": "application/json",
"Accept-Encoding": "gzip",
"User-Agent": "PeopleStrong-Integration/1.0",
"X-Request-ID": "sync-20240115-abc123",
"Idempotency-Key": "emp-batch-20240115-001",
}
For large files (>100MB), never load the entire response into memory. Use streaming: (1) Set stream=True in requests. (2) Iterate over response chunks. (3) Write each chunk to disk. (4) Track progress with content-length header. (5) Support range requests (Resume download if it fails mid-way). (6) Verify integrity with checksum after download.
def download_large_file(url, output_path, headers, chunk_size=8192):
with requests.get(url, headers=headers, stream=True, timeout=300) as resp:
resp.raise_for_status()
total = int(resp.headers.get("content-length", 0))
downloaded = 0
with open(output_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=chunk_size):
f.write(chunk)
downloaded += len(chunk)
if total:
pct = (downloaded / total) * 100
print(f"\rProgress: {pct:.1f}%", end="")
print(f"\nDownloaded {downloaded} bytes to {output_path}")
Range: bytes=1000000- to resume from byte 1MB. Check if the server supports range requests (Accept-Ranges: bytes in response).A circuit breaker prevents cascading failures by stopping calls to a failing service. Three states: (1) Closed (normal) — requests flow through. (2) Open (tripped) — requests fail immediately without calling the service, after N consecutive failures. (3) Half-open — after a timeout, allow one test request. If it succeeds, go back to Closed; if it fails, stay Open. This protects both your system (from hanging on timeouts) and the failing service (from being overwhelmed by retries).
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = "closed"
self.last_failure_time = 0
def call(self, func, *args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
else:
raise Exception("Circuit is OPEN - service unavailable")
try:
result = func(*args, **kwargs)
self.failure_count = 0
self.state = "closed"
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
print(f"Circuit OPENED after {self.failure_count} failures")
raise
# Usage
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
try:
result = breaker.call(requests.get, "https://api.target.com/employees")
except:
print("Using cached data or fallback")
Use asyncio (Python) or ThreadPoolExecutor for concurrent calls. Key concerns: (1) Rate limiting — use a semaphore to cap concurrent requests. (2) Connection pooling — reuse connections with aiohttp.ClientSession. (3) Error isolation — one failed request shouldn't crash others. (4) Order preservation — if order matters, use asyncio.gather() which preserves input order. (5) Resource limits — don't open 10,000 connections; use a bounded semaphore.
import asyncio
import aiohttp
async def fetch_page(session, url, page, semaphore):
async with semaphore:
async with session.get(url, params={"page": page, "limit": 100}) as resp:
data = await resp.json()
return data["records"]
async def fetch_all_concurrent(url, total_pages, max_concurrent=5):
semaphore = asyncio.Semaphore(max_concurrent)
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, url, p, semaphore) for p in range(1, total_pages + 1)]
results = await asyncio.gather(*tasks, return_exceptions=True)
all_records = []
errors = []
for i, result in enumerate(results):
if isinstance(result, Exception):
errors.append({"page": i + 1, "error": str(result)})
else:
all_records.extend(result)
return all_records, errors
# Run it
records, errors = asyncio.run(fetch_all_concurrent("https://api.source.com/employees", 100))
Throttling is limiting the rate of your outgoing requests to respect the target API's limits. Different from rate limiting (which is the server rejecting excess requests). You implement throttling proactively to avoid getting rate-limited. Methods: (1) Fixed delay between requests. (2) Token bucket — allow bursts but cap average rate. (3) Leaky bucket — smooth, constant rate. (4) Read the API's rate limit headers (X-RateLimit-Remaining) and adjust dynamically.
import time, threading
class TokenBucket:
def __init__(self, rate, capacity):
self.rate = rate # tokens per second
self.capacity = capacity # max burst
self.tokens = capacity
self.last_refill = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
def wait_and_acquire(self):
while not self.acquire():
time.sleep(0.05)
# 10 requests per second, burst of 20
bucket = TokenBucket(rate=10, capacity=20)
def throttled_request(url):
bucket.wait_and_acquire()
return requests.get(url)
Post-sync reconciliation is the final verification step. Full process: (1) Count check: SELECT COUNT(*) from both systems. (2) Key check: all source keys exist in target. (3) Value check: sample records match field-by-field. (4) Aggregate check: SUM/AVG of numeric fields match. (5) Freshness check: latest record in target matches latest in source. (6) Generate a reconciliation report. (7) Alert on any discrepancy. This should be automated and run after every sync.
def full_reconciliation(source_url, target_url, headers):
# Fetch counts
src_count = requests.get(f"{source_url}/count", headers=headers).json()["total"]
tgt_count = requests.get(f"{target_url}/count", headers=headers).json()["total"]
report = {
"timestamp": datetime.utcnow().isoformat(),
"source_count": src_count,
"target_count": tgt_count,
"count_match": src_count == tgt_count,
"checks": [],
}
# Sample validation (10 random records)
import random
src_records = requests.get(source_url, params={"limit": 1000}, headers=headers).json()
sample_ids = random.sample([r["id"] for r in src_records], min(10, len(src_records)))
for eid in sample_ids:
src = requests.get(f"{source_url}/{eid}", headers=headers).json()
tgt = requests.get(f"{target_url}/{eid}", headers=headers).json()
match = src["name"] == tgt["name"] and src["email"] == tgt["email"]
report["checks"].append({"id": eid, "match": match})
report["status"] = "PASS" if all(c["match"] for c in report["checks"]) and report["count_match"] else "FAIL"
return report
2 API to SFTP Integration
The pipeline: (1) Call the API and collect all records (with pagination handling). (2) Transform data into the required file format (CSV, XML, JSON). (3) Write to a local temp file. (4) Connect to SFTP using paramiko. (5) Upload the file to the correct directory. (6) Verify the upload (check file size on SFTP matches local). (7) Clean up local temp file. (8) Log and notify. Always use a temp file, never stream directly — if SFTP fails mid-upload, you still have the complete local file to retry.
import paramiko, csv, tempfile, os
def api_to_sftp(api_url, sftp_config, remote_path):
# 1. Fetch from API
records = fetch_all_offset(api_url, headers={"Authorization": "Bearer xxx"})
# 2. Write to temp CSV
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
writer = csv.DictWriter(f, fieldnames=["emp_id", "name", "email", "department"])
writer.writeheader()
writer.writerows(records)
local_path = f.name
# 3. Upload to SFTP
transport = paramiko.Transport((sftp_config["host"], sftp_config["port"]))
transport.connect(username=sftp_config["user"], password=sftp_config["password"])
sftp = paramiko.SFTPClient.from_transport(transport)
try:
sftp.put(local_path, remote_path)
# 4. Verify
remote_size = sftp.stat(remote_path).st_size
local_size = os.path.getsize(local_path)
assert remote_size == local_size, f"Size mismatch: {local_size} vs {remote_size}"
print(f"Uploaded {len(records)} records to {remote_path}")
finally:
sftp.close()
transport.close()
os.unlink(local_path)
Use Python's csv module with DictWriter for clean, consistent output. Key considerations: (1) Handle nested JSON by flattening (address.city becomes address_city). (2) Handle None values (write empty string, not "None"). (3) Handle special characters (commas, quotes, newlines in data — CSV module handles quoting automatically). (4) Set proper encoding (UTF-8 with BOM if Excel compatibility needed). (5) Use consistent date format in output.
import csv
from datetime import datetime
def flatten_record(record, parent_key="", sep="_"):
"""Flatten nested dict: {address: {city: 'Delhi'}} -> {address_city: 'Delhi'}"""
items = []
for k, v in record.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(flatten_record(v, new_key, sep).items())
else:
items.append((new_key, v if v is not None else ""))
return dict(items)
def generate_csv(records, output_path, encoding="utf-8-sig"):
# utf-8-sig adds BOM for Excel compatibility
flat_records = [flatten_record(r) for r in records]
fieldnames = list(flat_records[0].keys()) if flat_records else []
with open(output_path, "w", newline="", encoding=encoding) as f:
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
writer.writerows(flat_records)
return output_path
For large datasets (100K+ records), never load everything into memory. Use streaming/generator pattern: (1) Fetch data page by page from API (generator function). (2) Write each page to CSV immediately (don't accumulate). (3) Use context managers to ensure files are properly closed. (4) For very large files, split into multiple files (e.g., 50K records per file). (5) Monitor memory usage with tracemalloc during testing.
def stream_api_pages(url, headers):
"""Generator that yields records page by page"""
page = 1
while True:
resp = requests.get(url, params={"page": page, "limit": 500}, headers=headers)
data = resp.json()["data"]
if not data:
break
yield from data # Yield individual records
page += 1
def stream_to_csv(url, headers, output_path, fieldnames):
"""Write API data to CSV without loading all into memory"""
with open(output_path, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
count = 0
for record in stream_api_pages(url, headers):
writer.writerow(record)
count += 1
if count % 10000 == 0:
print(f"Written {count} records...")
print(f"Total: {count} records")
Good file naming prevents conflicts and enables automation. Convention: {source}_{type}_{date}_{sequence}.{ext}. Examples: peoplestrong_employees_20240115_001.csv, hrms_attendance_20240115_full.csv. Rules: (1) Include date for traceability. (2) Include sequence number for multiple files per day. (3) Use lowercase, underscores (no spaces). (4) Upload with a temp extension (.tmp), then rename to final — this prevents the consumer from reading a partial file. (5) Include load type (full/delta) in the name.
from datetime import datetime
def generate_filename(source, data_type, load_type="full", ext="csv"):
date_str = datetime.now().strftime("%Y%m%d")
timestamp = datetime.now().strftime("%H%M%S")
return f"{source}_{data_type}_{date_str}_{load_type}_{timestamp}.{ext}"
# Safe upload pattern: write as .tmp, then rename
def safe_sftp_upload(sftp, local_path, remote_dir, filename):
temp_remote = f"{remote_dir}/.{filename}.tmp"
final_remote = f"{remote_dir}/{filename}"
sftp.put(local_path, temp_remote)
sftp.rename(temp_remote, final_remote) # Atomic rename
return final_remote
# Result: peoplestrong_employees_20240115_full_023000.csv
pattern: "{type}_{YYYYMMDD}.csv"Compression reduces upload time and storage. Use gzip for single files (most universally supported), zip for multiple files. For a 100MB CSV, gzip typically achieves 85-90% compression (down to 10-15MB). Always compress before upload, not during (avoid streaming compression to SFTP). Name the file with .gz or .zip extension so the consumer knows it's compressed.
import gzip, shutil, zipfile
def gzip_file(input_path):
output_path = f"{input_path}.gz"
with open(input_path, "rb") as f_in:
with gzip.open(output_path, "wb") as f_out:
shutil.copyfileobj(f_in, f_out)
return output_path
def zip_files(file_paths, output_path):
with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zf:
for fp in file_paths:
zf.write(fp, os.path.basename(fp))
return output_path
# Usage
compressed = gzip_file("employees_20240115.csv")
# employees_20240115.csv (100MB) -> employees_20240115.csv.gz (12MB)
Verify with: (1) File size match — local size == remote size. (2) MD5/SHA256 checksum — compute hash locally, compare with hash of remote file. (3) Row count validation — upload a control file (.ctl) alongside the data file containing the expected row count. (4) Download and compare — re-download the file and diff with local (expensive but thorough). The control file approach is most common in enterprise SFTP integrations.
import hashlib
def compute_md5(file_path):
md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
md5.update(chunk)
return md5.hexdigest()
def upload_with_verification(sftp, local_path, remote_path):
# Upload data file
sftp.put(local_path, remote_path)
# Size check
local_size = os.path.getsize(local_path)
remote_size = sftp.stat(remote_path).st_size
assert local_size == remote_size, f"Size mismatch!"
# Upload control file with checksum + row count
checksum = compute_md5(local_path)
row_count = sum(1 for _ in open(local_path)) - 1 # Minus header
ctl_path = remote_path.replace(".csv", ".ctl")
with sftp.open(ctl_path, "w") as ctl:
ctl.write(f"filename={os.path.basename(remote_path)}\n")
ctl.write(f"row_count={row_count}\n")
ctl.write(f"md5={checksum}\n")
ctl.write(f"timestamp={datetime.utcnow().isoformat()}\n")
print(f"Verified: {row_count} rows, MD5: {checksum}")
SFTP drops happen due to network issues, firewall timeouts, or server-side limits. Solutions: (1) Keep-alive — send periodic packets to keep the connection alive. (2) Auto-reconnect — wrap SFTP operations in a retry decorator that re-establishes the connection. (3) Chunked upload — for large files, upload in chunks so you can resume from where you left off. (4) Connection pooling — reuse connections across uploads. (5) Set timeout values appropriately (not too short, not too long).
class ResilientSFTP:
def __init__(self, host, port, user, password):
self.config = {"host": host, "port": port, "user": user, "password": password}
self.transport = None
self.sftp = None
def connect(self):
self.transport = paramiko.Transport((self.config["host"], self.config["port"]))
self.transport.connect(username=self.config["user"], password=self.config["password"])
self.transport.set_keepalive(30) # Send keepalive every 30s
self.sftp = paramiko.SFTPClient.from_transport(self.transport)
def upload(self, local_path, remote_path, max_retries=3):
for attempt in range(max_retries):
try:
if not self.transport or not self.transport.is_active():
self.connect()
self.sftp.put(local_path, remote_path)
return True
except (paramiko.SSHException, OSError) as e:
print(f"Connection lost (attempt {attempt+1}): {e}")
time.sleep(2 ** attempt)
self.connect() # Reconnect
raise Exception("SFTP upload failed after max retries")
Options: (1) Cron job (Linux) — most reliable for simple schedules. (2) APScheduler (Python) — in-process scheduler. (3) Airflow/Prefect — for complex DAGs with dependencies, retries, alerting. (4) Cloud schedulers — AWS EventBridge + Lambda, Azure Functions timer trigger. For production integration pipelines, use Airflow or equivalent — it provides retry, monitoring, backfill, and dependency management.
# Option 1: Cron (crontab -e)
# Run every day at 2 AM
# 0 2 * * * /usr/bin/python3 /opt/integrations/api_to_sftp.py >> /var/log/sync.log 2>&1
# Option 2: APScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job("cron", hour=2, minute=0)
def daily_sync():
try:
api_to_sftp(
api_url="https://api.peoplestrong.com/employees",
sftp_config={"host": "sftp.payroll.com", "port": 22,
"user": "integration", "password": os.environ["SFTP_PASS"]},
remote_path="/incoming/employees_daily.csv"
)
send_notification("Daily sync completed successfully")
except Exception as e:
send_alert(f"Daily sync FAILED: {e}")
scheduler.start()
FTP (File Transfer Protocol): Unencrypted, uses port 21, transmits credentials in plain text. Insecure — never use for sensitive data. FTPS: FTP + TLS/SSL encryption. Better but complex (multiple ports, firewall issues). SFTP (SSH File Transfer Protocol): Runs over SSH (port 22), encrypted by default, single port, supports key-based auth. Always use SFTP for integration — it's secure, simple, and firewall-friendly. SFTP is NOT "FTP over SSH" — it's a completely different protocol that happens to use SSH as its transport.
PII (Personally Identifiable Information) must be protected. Techniques: (1) Masking — replace part of data (email: j***@example.com). (2) Tokenization — replace with a random token, store mapping separately. (3) Encryption — encrypt PII fields, decrypt at destination. (4) Hashing — one-way hash for fields that need matching but not reading. (5) Column-level filtering — don't include PII columns if not needed. Always log what was masked and maintain an audit trail.
import re, hashlib
def mask_email(email):
if not email or "@" not in email:
return email
local, domain = email.split("@")
return f"{local[0]}{'*' * (len(local)-1)}@{domain}"
def mask_phone(phone):
return re.sub(r"\d(?=\d{4})", "*", phone) # Show last 4 digits
def mask_aadhaar(aadhaar):
return f"XXXX-XXXX-{aadhaar[-4:]}"
def mask_record(record):
masked = record.copy()
if "email" in masked: masked["email"] = mask_email(masked["email"])
if "phone" in masked: masked["phone"] = mask_phone(masked["phone"])
if "aadhaar" in masked: masked["aadhaar"] = mask_aadhaar(masked["aadhaar"])
return masked
Options: (1) Overwrite — sftp.put() overwrites by default. Use when today's file should replace yesterday's. (2) Version — append timestamp to filename, never overwrite. (3) Archive first — move existing file to /archive/ before uploading new one. (4) Skip — if file exists with same hash, don't re-upload. (5) Error — fail and alert if unexpected file exists. Choice depends on business rules.
def upload_with_archive(sftp, local_path, remote_dir, filename):
remote_path = f"{remote_dir}/{filename}"
archive_dir = f"{remote_dir}/archive"
# Check if file exists
try:
sftp.stat(remote_path)
# File exists - archive it
try: sftp.mkdir(archive_dir)
except: pass
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_path = f"{archive_dir}/{filename}.{ts}"
sftp.rename(remote_path, archive_path)
print(f"Archived existing file to {archive_path}")
except FileNotFoundError:
pass # No existing file, proceed
sftp.put(local_path, remote_path)
Notifications are essential for monitoring. Send emails with: (1) Status (success/failure). (2) File details (name, size, row count). (3) Duration. (4) Error details if failed. Use Python's smtplib or a service like SendGrid/SES. For critical pipelines, also send to Slack/Teams. Always notify on both success AND failure — silence should be treated as failure.
import smtplib
from email.mime.text import MIMEText
def send_notification(subject, body, recipients):
msg = MIMEText(body, "html")
msg["Subject"] = subject
msg["From"] = "integration@company.com"
msg["To"] = ", ".join(recipients)
with smtplib.SMTP("smtp.company.com", 587) as server:
server.starttls()
server.login("integration@company.com", os.environ["SMTP_PASS"])
server.sendmail(msg["From"], recipients, msg.as_string())
# After successful upload
send_notification(
subject="[SUCCESS] Employee Sync - 2024-01-15",
body="""<h3>Daily Employee Sync Report</h3>
<ul>
<li>Records: 5,234</li>
<li>File: employees_20240115_full.csv (2.3 MB)</li>
<li>Duration: 45 seconds</li>
<li>Status: Uploaded to SFTP successfully</li>
</ul>""",
recipients=["payroll-team@company.com"]
)
PGP (Pretty Good Privacy) encrypts files so only the intended recipient can read them. In SFTP integrations, even though SFTP encrypts data in transit, PGP encrypts data at rest on the SFTP server. Process: (1) Recipient gives you their public key. (2) You encrypt the file with their public key. (3) Upload encrypted file (.pgp/.gpg extension). (4) Recipient decrypts with their private key. This is mandatory for financial data, PII, and compliance (SOX, HIPAA, PCI).
import gnupg
def pgp_encrypt_file(input_path, recipient_key_id, output_path=None):
gpg = gnupg.GPG()
output_path = output_path or f"{input_path}.pgp"
with open(input_path, "rb") as f:
encrypted = gpg.encrypt_file(
f,
recipients=[recipient_key_id],
output=output_path,
always_trust=True,
armor=False # Binary output (smaller)
)
if encrypted.ok:
print(f"Encrypted: {input_path} -> {output_path}")
return output_path
raise Exception(f"Encryption failed: {encrypted.status}")
# Full pipeline: generate CSV -> encrypt -> upload to SFTP
csv_path = generate_csv(records, "employees.csv")
pgp_path = pgp_encrypt_file(csv_path, "payroll-team@company.com")
safe_sftp_upload(sftp, pgp_path, "/incoming", "employees.csv.pgp")
Design your pipeline with a formatter abstraction. The core ETL extracts and transforms data into a common format (list of dicts). The final step applies the required output format. This way, adding a new format is just adding one function. Common libraries: csv (CSV), xml.etree (XML), json (JSON), openpyxl (Excel). Let the consumer's requirements dictate the format — don't assume CSV.
import json, csv
import xml.etree.ElementTree as ET
from openpyxl import Workbook
def write_csv(records, path):
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=records[0].keys())
w.writeheader()
w.writerows(records)
def write_json(records, path):
Path(path).write_text(json.dumps(records, indent=2, default=str))
def write_xml(records, path, root_tag="employees", row_tag="employee"):
root = ET.Element(root_tag)
for record in records:
row = ET.SubElement(root, row_tag)
for key, val in record.items():
child = ET.SubElement(row, key)
child.text = str(val) if val is not None else ""
ET.ElementTree(root).write(path, encoding="unicode", xml_declaration=True)
def write_excel(records, path):
wb = Workbook()
ws = wb.active
ws.append(list(records[0].keys()))
for r in records:
ws.append(list(r.values()))
wb.save(path)
# Config-driven format selection
FORMATTERS = {"csv": write_csv, "json": write_json, "xml": write_xml, "xlsx": write_excel}
FORMATTERS[config["output_format"]](records, output_path)
Checksums verify file integrity end-to-end. Process: (1) Compute checksum (MD5 or SHA256) of the local file before upload. (2) Upload the file + checksum (in a .md5 or .sha256 sidecar file). (3) Consumer downloads both files. (4) Consumer computes checksum of downloaded file. (5) Compare: if checksums match, file is intact; if not, re-download. SHA256 is preferred over MD5 for security, but MD5 is sufficient for integrity checks.
import hashlib
def compute_checksum(file_path, algorithm="sha256"):
h = hashlib.new(algorithm)
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def upload_with_checksum(sftp, local_path, remote_dir):
filename = os.path.basename(local_path)
checksum = compute_checksum(local_path)
# Upload data file
sftp.put(local_path, f"{remote_dir}/{filename}")
# Upload checksum sidecar file
checksum_content = f"{checksum} {filename}\n"
with sftp.open(f"{remote_dir}/{filename}.sha256", "w") as f:
f.write(checksum_content)
print(f"Uploaded {filename} with SHA256: {checksum}")
3 SFTP to API Integration
Pipeline: (1) Connect to SFTP. (2) List files in the directory, pick the right one (latest, by naming pattern). (3) Download to local temp file. (4) Parse the file (CSV, XML, etc.). (5) Validate records. (6) Transform to target API's schema. (7) Batch-POST to API. (8) Archive the processed file (move to /processed/ on SFTP). (9) Log results and notify. Key: always archive after processing so you don't reprocess the same file.
def sftp_to_api(sftp_config, remote_dir, api_url, api_headers):
# Connect to SFTP
transport = paramiko.Transport((sftp_config["host"], sftp_config["port"]))
transport.connect(username=sftp_config["user"], password=sftp_config["password"])
sftp = paramiko.SFTPClient.from_transport(transport)
# Find latest CSV file
files = [f for f in sftp.listdir(remote_dir) if f.endswith(".csv")]
if not files:
print("No files to process")
return
latest = sorted(files)[-1]
remote_path = f"{remote_dir}/{latest}"
# Download to temp
with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tmp:
sftp.get(remote_path, tmp.name)
local_path = tmp.name
# Parse CSV
with open(local_path, "r") as f:
records = list(csv.DictReader(f))
# Transform and push to API
result = batch_upload(records, api_url, api_headers)
# Archive processed file
archive_path = f"{remote_dir}/processed/{latest}"
try: sftp.mkdir(f"{remote_dir}/processed")
except: pass
sftp.rename(remote_path, archive_path)
sftp.close()
transport.close()
os.unlink(local_path)
return result
Malformed files include: wrong encoding, invalid CSV structure, unexpected delimiters, binary content mixed in, truncated files. Defense: (1) Check file size > 0. (2) Try to parse the header row — if it fails, the file is corrupt. (3) Use try/except on each row. (4) Validate column count matches header count. (5) Move corrupt files to /error/ directory. (6) Alert with details.
def safe_parse_csv(file_path):
valid_records, errors = [], []
try:
with open(file_path, "r", encoding="utf-8") as f:
first_line = f.readline()
if not first_line.strip():
return [], [{"line": 0, "error": "File is empty"}]
f.seek(0)
reader = csv.DictReader(f)
for line_num, row in enumerate(reader, start=2):
try:
if None in row:
errors.append({"line": line_num, "error": "Extra columns"})
continue
if not row.get("emp_id", "").strip():
errors.append({"line": line_num, "error": "Missing emp_id"})
continue
valid_records.append(row)
except Exception as e:
errors.append({"line": line_num, "error": str(e)})
except UnicodeDecodeError:
errors.append({"line": 0, "error": "File encoding is not UTF-8"})
return valid_records, errors
This is the "partial validity" problem. (1) Process all 1000 records through validation. (2) Separate into valid (995) and invalid (5) buckets. (3) Push 995 valid records to API. (4) Write the 5 invalid records to an error file with reasons. (5) Upload error file to SFTP /errors/. (6) Notification: "995 processed, 5 rejected." (7) If >10% fail, reject the entire file (systemic issue).
def process_with_error_report(file_path, api_url, headers, error_threshold=0.1):
valid, errors = safe_parse_csv(file_path)
total = len(valid) + len(errors)
error_rate = len(errors) / total if total > 0 else 0
if error_rate > error_threshold:
raise Exception(f"Error rate {error_rate:.1%} exceeds {error_threshold:.1%}")
result = batch_upload(valid, api_url, headers)
if errors:
error_path = file_path.replace(".csv", "_errors.csv")
with open(error_path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=["line", "error"])
w.writeheader()
w.writerows(errors)
return {"processed": len(valid), "rejected": len(errors)}
Archiving ensures traceability and prevents reprocessing. Pattern: After successful processing, move file from /incoming/ to /processed/YYYY/MM/. Keep a manifest log. Retention policy: 90 days. For failed files, move to /errors/ with timestamp suffix. Never delete without archiving.
def archive_file(sftp, source_path, base_archive_dir):
now = datetime.now()
archive_dir = f"{base_archive_dir}/{now:%Y}/{now:%m}"
for d in [base_archive_dir, f"{base_archive_dir}/{now:%Y}", archive_dir]:
try: sftp.mkdir(d)
except: pass
filename = os.path.basename(source_path)
archive_path = f"{archive_dir}/{filename}"
sftp.rename(source_path, archive_path)
return archive_path
Processing order matters when files have dependencies. Strategies: (1) Timestamp sort — process oldest first (FIFO). (2) Filename sort — if files have sequence numbers, process in order. (3) Dependency-based — process departments before employees. (4) Manifest file — a control file lists the correct order. (5) If independent, process in parallel.
def process_files_in_order(sftp, remote_dir, processing_order=None):
files = sftp.listdir(remote_dir)
csv_files = [f for f in files if f.endswith(".csv")]
if processing_order:
ordered = []
for pattern in processing_order:
for f in csv_files:
if pattern in f:
ordered.append(f)
csv_files = ordered
else:
csv_files.sort()
for filename in csv_files:
process_single_file(sftp, f"{remote_dir}/{filename}")
# departments must load before employees
process_files_in_order(sftp, "/incoming",
processing_order=["departments", "designations", "employees"])
Detection: (1) Maintain a processed files log (filename + checksum + timestamp). (2) Before processing, check if file's checksum matches a previously processed file. (3) If duplicate, skip and log. (4) Same filename but different content = correction, process it. (5) Same checksum = true duplicate, skip.
def is_duplicate(file_path, processed_log_path="processed_files.json"):
current_checksum = compute_checksum(file_path)
if Path(processed_log_path).exists():
log = json.loads(Path(processed_log_path).read_text())
else:
log = {}
# Check if exact content was already processed
if current_checksum in [e["checksum"] for e in log.values()]:
print("DUPLICATE: exact content match, skipping")
return True
return False
Use a parser factory pattern. Detect format from extension or config. Each parser normalizes data into a list of dicts. Handle format-specific quirks: CSV (delimiter, quoting), XML (namespaces), JSON (array vs object), Excel (sheets, types). Fixed-width is common in legacy systems.
def parse_file(file_path):
ext = Path(file_path).suffix.lower()
parsers = {".csv": parse_csv, ".json": parse_json,
".xml": parse_xml, ".xlsx": parse_excel}
parser = parsers.get(ext)
if not parser:
raise ValueError(f"Unsupported format: {ext}")
return parser(file_path)
def parse_csv(path):
with open(path) as f:
sample = f.read(1024)
f.seek(0)
dialect = csv.Sniffer().sniff(sample) # Auto-detect delimiter
return list(csv.DictReader(f, dialect=dialect))
def parse_fixed_width(path):
# Legacy: emp_id(0:10), name(10:40), dept(40:60)
cols = [("emp_id",0,10), ("name",10,40), ("dept",40,60)]
records = []
with open(path) as f:
for line in f:
records.append({n: line[s:e].strip() for n,s,e in cols})
return records
Empty file could mean: no changes (valid), source error (invalid), or generation failure. Check file size first. If 0 bytes: log and skip. If header only (no data rows): "valid empty" — log "0 records." If the file should always have data (daily attendance), treat empty as error and alert. Config: allow_empty: true/false per integration.
Strategy: (1) Try UTF-8 first. (2) If UnicodeDecodeError, try UTF-8-SIG (BOM), Latin-1, CP1252 (Windows). (3) Use chardet library to auto-detect. (4) Document expected encoding in spec. (5) Always write output in UTF-8.
import chardet
def read_with_fallback(file_path):
encodings = ["utf-8", "utf-8-sig", "latin-1", "cp1252"]
for enc in encodings:
try:
with open(file_path, "r", encoding=enc) as f:
records = list(csv.DictReader(f))
print(f"Read with {enc} encoding")
return records
except UnicodeDecodeError:
continue
raise ValueError("Could not decode file")
For SFTP, polling is the only option (no push notifications). Check directory every N minutes. Use a "last seen" state file to track processed files. Compare current file list against state to find new ones.
def poll_sftp_for_new_files(sftp, remote_dir, state_file="sftp_state.json"):
seen = set(json.loads(Path(state_file).read_text())) if Path(state_file).exists() else set()
current_files = set(sftp.listdir(remote_dir))
new_files = current_files - seen
if new_files:
print(f"Found {len(new_files)} new files")
seen.update(new_files)
Path(state_file).write_text(json.dumps(list(seen)))
return sorted(new_files)
A staging table is a temporary DB table where you load raw file data before processing. Benefits: decouple parsing from business logic, validate with SQL, handle duplicates with GROUP BY, easy rollback (clear staging table). Pattern: Clear staging → Load raw data → Validate → Transform → Upsert into production → Archive staging.
-- Staging table (mirrors CSV structure)
-- stg_employees: row_num, emp_id, name, email, department,
-- load_date, is_valid, error_message
-- Step 1: Bulk load raw CSV data into staging
-- Step 2: Validate with SQL
-- Mark invalid: SET is_valid=FALSE WHERE emp_id IS NULL
-- Mark invalid: SET is_valid=FALSE WHERE email NOT LIKE '%@%'
-- Step 3: Upsert valid records to production table
-- Upsert from staging WHERE is_valid = TRUE
-- ON CONFLICT (emp_id) update name, email, department
Detection methods: (1) Control file — source uploads data.csv, then data.ctl with expected row count. Only process when both exist. (2) Trigger file — source uploads data.csv, then data.done (empty). Process only when .done exists. (3) File stability — check size at t=0 and t=30s; if same, complete. (4) Footer/trailer — last row contains expected count. Validate actual vs expected.
Header mismatches: columns renamed, extra columns, missing columns, reordered. Handle: (1) Compare actual header against expected. (2) Missing required columns: reject file. (3) Extra columns: ignore. (4) Renamed columns: use alias mapping. (5) Reordered: DictReader handles this (reads by name, not position).
def validate_headers(file_path, required, aliases=None):
aliases = aliases or {}
with open(file_path) as f:
actual = set(csv.DictReader(f).fieldnames)
normalized = {aliases.get(c, c) for c in actual}
missing = set(required) - normalized
if missing:
raise ValueError(f"Missing required columns: {missing}")
return True
validate_headers("employees.csv",
required=["emp_id", "name", "email"],
aliases={"employee_code": "emp_id", "full_name": "name"})
fieldnames = [f.strip().lower() for f in reader.fieldnames]CSV is all strings. You must validate and convert: dates, numbers, booleans, emails. Use Pydantic for schema-based validation. Handle null/empty explicitly. Log errors with row number and column name.
from pydantic import BaseModel, field_validator
from datetime import date
class EmployeeRecord(BaseModel):
emp_id: str
name: str
salary: float
join_date: date
@field_validator("salary", mode="before")
def parse_salary(cls, v):
if isinstance(v, str):
return float(v.replace(",", "")) # "50,000" -> 50000.0
return v
@field_validator("join_date", mode="before")
def parse_date(cls, v):
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d-%b-%Y"):
try: return datetime.strptime(v, fmt).date()
except: continue
raise ValueError(f"Cannot parse date: {v}")
For 100MB+ files: (1) Never load into memory. Stream line by line. (2) Batch API calls — accumulate 500 records, POST, clear, repeat. (3) Track progress every 10K rows. (4) Use generators for memory efficiency. (5) Monitor rows/sec for performance baseline.
def stream_large_csv_to_api(file_path, api_url, headers, batch_size=500):
stats = {"total": 0, "success": 0, "failed": 0}
batch = []
with open(file_path) as f:
for row in csv.DictReader(f):
batch.append(row)
stats["total"] += 1
if len(batch) >= batch_size:
try:
requests.post(api_url, json={"records": batch}, headers=headers)
stats["success"] += len(batch)
except:
stats["failed"] += len(batch)
batch = []
if stats["total"] % 10000 == 0:
print(f"Progress: {stats['total']} rows")
if batch: # remaining
try:
requests.post(api_url, json={"records": batch}, headers=headers)
stats["success"] += len(batch)
except:
stats["failed"] += len(batch)
return stats
4 Full Load vs Incremental Load
Full load means extracting ALL records from the source and replacing ALL records in the target. It's the simplest approach: clear target, insert everything. Use when: (1) Source system has no way to track changes (no modified_date). (2) Data volume is small (under 10K records). (3) Initial load (first-time setup). (4) After a major data migration or cleanup. (5) Weekly/monthly reconciliation runs. Downside: slow for large datasets, resource-intensive, and causes downtime if you clear-and-reload.
Incremental load means extracting only records that changed since the last sync (new, modified, deleted). Much faster and less resource-intensive than full load. Use when: (1) Large data volume (100K+ records). (2) Source has a reliable modified_date or change tracking. (3) Frequent syncs (hourly/real-time). (4) Network bandwidth is limited. (5) Target system can handle upserts. Methods: timestamp-based, CDC (Change Data Capture), event-driven.
Key differences: Full: Extract all records, replace target completely. Simple but slow. Good for small datasets or reconciliation. Incremental: Extract only changes, upsert to target. Fast but complex. Requires change tracking. Risk: can miss records if change tracking is unreliable. Best practice: use incremental for daily operations, full load weekly for reconciliation. This hybrid approach catches anything incremental missed.
Timestamp-based incremental: (1) Store the last sync timestamp (high watermark). (2) Query source for records WHERE modified_date > last_sync_time. (3) Upsert those records into target. (4) Update the watermark to the MAX(modified_date) from this batch. Critical: use the data's max timestamp, not the current clock time, to avoid missing records modified during the sync window.
def incremental_sync(source_url, target_url, headers, watermark_file="watermark.json"):
# Load watermark
if Path(watermark_file).exists():
watermark = json.loads(Path(watermark_file).read_text())["last_sync"]
else:
watermark = "1970-01-01T00:00:00Z" # First run = get everything
# Fetch only changed records
resp = requests.get(source_url, params={"modified_since": watermark}, headers=headers)
records = resp.json()["data"]
if not records:
print("No changes since last sync")
return
# Upsert to target
result = batch_upload(records, target_url, headers)
# Update watermark to MAX modified_date from the data
max_modified = max(r["modified_date"] for r in records)
Path(watermark_file).write_text(json.dumps({"last_sync": max_modified}))
print(f"Synced {len(records)} changed records. New watermark: {max_modified}")
CDC tracks every change at the database level. Types: (1) Log-based CDC — reads database transaction log (WAL in Postgres, binlog in MySQL). Most accurate, no source modification needed. Tools: Debezium, AWS DMS. (2) Trigger-based — DB triggers write changes to a changelog table. (3) Polling-based — periodically query for changes using timestamps (simplest CDC). Log-based is the gold standard because it captures ALL changes including deletes without modifying the source.
# Polling-based CDC (simplest form)
# Requires a changes table in source DB:
# change_log: id, table_name, record_id, operation (I/U/D),
# changed_data, changed_at
def cdc_sync(source_db, target_url, headers, last_change_id=0):
# Fetch changes since last processed ID
changes = source_db.execute(
"SELECT * FROM change_log WHERE id > ? ORDER BY id",
(last_change_id,)
).fetchall()
for change in changes:
if change["operation"] == "I": # Insert
requests.post(target_url, json=change["changed_data"], headers=headers)
elif change["operation"] == "U": # Update
requests.put(f"{target_url}/{change['record_id']}",
json=change["changed_data"], headers=headers)
elif change["operation"] == "D": # Delete (soft)
requests.patch(f"{target_url}/{change['record_id']}",
json={"status": "deleted"}, headers=headers)
if changes:
new_last_id = changes[-1]["id"]
save_checkpoint(new_last_id)
This is a common real-world problem. Solutions: (1) Hash comparison — compute a hash of each record, compare with stored hashes. Changed hash = changed record. (2) Full snapshot comparison — extract full dataset, compare with previous snapshot, find differences. (3) Ask source to add modified_date — the right long-term solution. (4) DB trigger approach — add a trigger on the source table that updates a timestamp column on every change. (5) Log-based CDC — doesn't need modified_date, reads from DB transaction logs.
import hashlib
def detect_changes_by_hash(current_records, previous_hashes, key_field="emp_id"):
changes = {"new": [], "modified": [], "deleted": []}
current_hashes = {}
for record in current_records:
key = record[key_field]
record_hash = hashlib.md5(json.dumps(record, sort_keys=True).encode()).hexdigest()
current_hashes[key] = record_hash
if key not in previous_hashes:
changes["new"].append(record)
elif previous_hashes[key] != record_hash:
changes["modified"].append(record)
# Detect deletions
deleted_keys = set(previous_hashes.keys()) - set(current_hashes.keys())
changes["deleted"] = list(deleted_keys)
return changes, current_hashes
Deletes are the hardest part of incremental sync. Approaches: (1) Soft delete flag — source sets is_deleted=true, your incremental picks it up via modified_date. Best approach. (2) Tombstone events — CDC publishes a DELETE event. (3) Full snapshot diff — compare current full extract with previous; missing records = deleted. (4) Periodic reconciliation — weekly full load identifies records in target but not in source. (5) Never delete — some systems just mark inactive, never hard delete.
A watermark (or high-water mark) is the marker that tracks your position in the data stream. For timestamp-based: it's the MAX(modified_date) from the last successful sync. For ID-based: it's the last processed ID. For CDC: it's the log sequence number (LSN). Rules: (1) Only advance watermark after SUCCESSFUL processing. (2) Store persistently (file, DB, not memory). (3) Use atomic write (temp file + rename) to prevent corruption. (4) Include metadata (sync run ID, record count) for debugging.
class WatermarkManager:
def __init__(self, path="watermark.json"):
self.path = Path(path)
self.data = self._load()
def _load(self):
if self.path.exists():
return json.loads(self.path.read_text())
return {"value": None, "updated_at": None, "run_id": None}
def get(self):
return self.data["value"]
def advance(self, new_value, run_id, record_count):
# Atomic write: write to temp, then rename
self.data = {
"value": new_value,
"updated_at": datetime.utcnow().isoformat(),
"run_id": run_id,
"records_processed": record_count,
"previous_value": self.data.get("value"),
}
tmp = self.path.with_suffix(".tmp")
tmp.write_text(json.dumps(self.data, indent=2))
tmp.rename(self.path)
Optimization strategies: (1) Switch to incremental — the best solution if possible. (2) Parallel extraction — split data into ranges (A-M, N-Z) and fetch in parallel. (3) Compression — gzip API responses. (4) Reduce payload — request only needed fields. (5) Bulk insert at target (COPY instead of INSERT). (6) Partition the load — full load different segments on different days (departments on Monday, employees on Tuesday). (7) Expand the window — negotiate with stakeholders for more time.
Initial load is the first-time full sync to bootstrap the target. After that, switch to incremental. Implementation: (1) Check if watermark exists. No watermark = initial load (fetch everything). (2) Initial load: full extract, bulk insert, set watermark. (3) Subsequent runs: check watermark, fetch delta, upsert. (4) Initial load might need special handling (larger batch sizes, progress tracking, chunked processing for memory).
def smart_sync(source_url, target_url, headers, watermark_mgr):
watermark = watermark_mgr.get()
if watermark is None:
print("Initial load - fetching ALL records")
records = fetch_all_offset(source_url, headers)
print(f"Initial load: {len(records)} records")
else:
print(f"Incremental load since {watermark}")
resp = requests.get(source_url,
params={"modified_since": watermark}, headers=headers)
records = resp.json()["data"]
print(f"Incremental: {len(records)} changed records")
if records:
result = batch_upload(records, target_url, headers)
max_ts = max(r["modified_date"] for r in records)
watermark_mgr.advance(max_ts, run_id="sync_001", record_count=len(records))
Upsert = INSERT if new, UPDATE if exists. Essential for incremental loads because you don't know if a record is new or modified. Implementation: (1) SQL: INSERT ... ON CONFLICT DO UPDATE (Postgres), MERGE (SQL Server, Oracle). (2) API: PUT (idempotent replace) or PATCH (partial update). (3) Application-level: check if exists, then insert or update. Always prefer database-level upsert — it's atomic and handles race conditions.
-- PostgreSQL Upsert
-- Insert new employees, update existing ones by emp_id
-- ON CONFLICT (emp_id) DO UPDATE SET
-- name = EXCLUDED.name, email = EXCLUDED.email,
-- department = EXCLUDED.department, updated_at = NOW()
# Python application-level upsert
def upsert_to_api(record, api_url, headers):
emp_id = record["emp_id"]
# Check if exists
check = requests.get(f"{api_url}/{emp_id}", headers=headers)
if check.status_code == 200:
# Update existing
requests.put(f"{api_url}/{emp_id}", json=record, headers=headers)
elif check.status_code == 404:
# Insert new
requests.post(api_url, json=record, headers=headers)
else:
check.raise_for_status()
Late-arriving data has a modified_date BEFORE your watermark (e.g., record modified at 10 PM yesterday but not committed to DB until 2 AM today, after your midnight sync). Solutions: (1) Overlap window — query with watermark minus a buffer (e.g., 2 hours). Accept some duplicates (upsert handles them). (2) Monotonic ID — use auto-increment ID instead of timestamp. (3) CDC — log-based CDC captures the actual commit time, not the business timestamp. (4) Reconciliation — weekly full sync catches anything missed.
SCD Type 1: Overwrite old value with new. No history preserved. Simple. Use when history doesn't matter (e.g., correcting a typo in name). SCD Type 2: Keep history. When a value changes, close the old record (set end_date) and insert a new record with the new value. Enables time-travel queries ("What was this employee's department on Jan 1?"). Use for: department changes, designation changes, salary history.
# SCD Type 1: Simple overwrite
def scd_type1(record, db):
db.execute("UPDATE employees SET name=?, dept=? WHERE emp_id=?",
(record["name"], record["dept"], record["emp_id"]))
# SCD Type 2: Preserve history
def scd_type2(record, db):
# Close current record
db.execute(
"UPDATE employee_history SET is_current=FALSE, end_date=? WHERE emp_id=? AND is_current=TRUE",
(datetime.now(), record["emp_id"]))
# Insert new version
db.execute(
"INSERT INTO employee_history (emp_id, name, dept, start_date, is_current) VALUES (?,?,?,?,TRUE)",
(record["emp_id"], record["name"], record["dept"], datetime.now()))
# Query: "What department was EMP001 in on Jan 1, 2024?"
# SELECT dept FROM employee_history
# WHERE emp_id='EMP001' AND start_date <= '2024-01-01' AND (end_date > '2024-01-01' OR is_current=TRUE)
Schema drift = source schema changes without notice (new columns, renamed columns, type changes). Detection: (1) Compare current response schema with stored expected schema before processing. (2) Alert on any changes. (3) Auto-handle additive changes (new optional columns). (4) Fail on breaking changes (removed required columns, type changes). (5) Maintain a schema registry that versioned schemas.
def detect_schema_drift(current_record, expected_schema):
current_fields = set(current_record.keys())
expected_fields = set(expected_schema.keys())
drift = {
"new_fields": current_fields - expected_fields,
"removed_fields": expected_fields - current_fields,
"type_changes": [],
}
for field in current_fields & expected_fields:
current_type = type(current_record[field]).__name__
expected_type = expected_schema[field]
if current_type != expected_type:
drift["type_changes"].append({
"field": field, "expected": expected_type, "actual": current_type})
has_drift = any(drift.values())
if has_drift:
print(f"SCHEMA DRIFT DETECTED: {drift}")
if drift["removed_fields"] or drift["type_changes"]:
raise Exception("Breaking schema change detected!")
return drift
Backfilling is re-processing historical data that was missed or incorrectly processed. Scenarios: (1) Bug in transformation logic — last 30 days of data had wrong department mapping. Need to re-extract and re-load. (2) New field added — need to populate it for existing records. (3) Incremental sync missed records due to clock skew. (4) New target system needs all historical data. Process: set watermark to the desired start date and re-run, or use a special "backfill mode" that processes a date range.
def backfill(source_url, target_url, headers, start_date, end_date):
print(f"Backfilling from {start_date} to {end_date}")
records = requests.get(source_url, params={
"modified_since": start_date,
"modified_before": end_date
}, headers=headers).json()["data"]
print(f"Backfill: {len(records)} records to reprocess")
result = batch_upload(records, target_url, headers)
print(f"Backfill complete: {result}")
# Note: do NOT update watermark during backfill
return result
Completeness checks: (1) Count validation — records fetched from source = records loaded to target. (2) Watermark continuity — no gaps between previous watermark and current. (3) Periodic full reconciliation — compare total counts weekly. (4) Trend analysis — if daily delta is usually 50-100 records and today it's 5, investigate. (5) Missing records check — spot-check specific known changes.
def validate_completeness(source_count, target_loaded, daily_avg, tolerance=0.2):
report = {
"source_records": source_count,
"target_loaded": target_loaded,
"match": source_count == target_loaded,
"daily_avg": daily_avg,
}
# Anomaly detection: is today's count suspiciously low?
if daily_avg > 0 and source_count < daily_avg * (1 - tolerance):
report["warning"] = f"Count ({source_count}) is {tolerance:.0%} below average ({daily_avg})"
if not report["match"]:
report["error"] = f"Count mismatch: {source_count} source vs {target_loaded} target"
return report
Records get missed due to: (1) Clock skew between source DB nodes. (2) Long-running transactions that commit after your watermark advances. (3) Soft delete without updating modified_date. (4) Source API bug filtering records. Detection: weekly full reconciliation comparing source and target counts and IDs. Fix: (1) Add overlap buffer to watermark query. (2) Switch to CDC for guaranteed capture. (3) Implement reconciliation alerts.
Timezone issues cause missed records or duplicate processing. Rules: (1) Store watermark in UTC. (2) Convert source timestamps to UTC before comparison. (3) If source doesn't include timezone info, DOCUMENT which timezone they assume. (4) Be aware of DST (Daylight Saving Time) — clocks jump forward/backward, creating gaps or overlaps. (5) Always use timezone-aware datetime objects, never naive.
from zoneinfo import ZoneInfo
def safe_incremental_query(source_url, watermark_utc, source_tz="Asia/Kolkata"):
# Convert UTC watermark to source's timezone for the API call
utc_dt = datetime.fromisoformat(watermark_utc).replace(tzinfo=ZoneInfo("UTC"))
source_dt = utc_dt.astimezone(ZoneInfo(source_tz))
resp = requests.get(source_url, params={
"modified_since": source_dt.isoformat()
})
records = resp.json()["data"]
# Convert response timestamps back to UTC for watermark
for r in records:
local = datetime.fromisoformat(r["modified_date"])
if local.tzinfo is None:
local = local.replace(tzinfo=ZoneInfo(source_tz))
r["modified_date"] = local.astimezone(ZoneInfo("UTC")).isoformat()
return records
At-most-once: Fire and forget. May lose records. Never use for integration. At-least-once: Retry on failure. May process duplicates. The standard for most integrations (combined with idempotency). Exactly-once: Each record processed exactly once. Theoretically impossible in distributed systems (Two Generals' Problem). Approximated via: idempotent operations + at-least-once delivery + deduplication. In practice, design for at-least-once and make your operations idempotent.
Rollback strategies: (1) Don't advance watermark — simplest. If load fails, watermark stays at old value. Next run re-fetches and retries. (2) Soft delete loaded records — if partial load corrupted target, mark loaded records as invalid. (3) Restore from backup — for critical data, take a snapshot before loading. (4) Compensation events — in event-driven systems, publish "undo" events. (5) Staging table pattern — load into staging first, only promote to production on full success.
def safe_incremental_with_rollback(source_url, target_url, headers, watermark_mgr):
old_watermark = watermark_mgr.get()
records = fetch_changes(source_url, old_watermark, headers)
# Track what we loaded (for potential rollback)
loaded_ids = []
try:
for batch in chunk_list(records, 100):
resp = requests.post(target_url, json={"records": batch}, headers=headers)
resp.raise_for_status()
loaded_ids.extend([r["emp_id"] for r in batch])
# All successful - advance watermark
max_ts = max(r["modified_date"] for r in records)
watermark_mgr.advance(max_ts, "run_001", len(records))
except Exception as e:
print(f"LOAD FAILED: {e}. Rolling back {len(loaded_ids)} records")
# Rollback: mark loaded records for re-processing
for eid in loaded_ids:
requests.patch(f"{target_url}/{eid}",
json={"needs_resync": True}, headers=headers)
# Watermark NOT advanced - next run will retry
raise
5 ETL Pipeline Design Questions
A production-grade daily employee sync pipeline: (1) Schedule — cron/Airflow triggers at 2 AM. (2) Extract — incremental fetch from source API using watermark. (3) Validate — schema validation, data type checks, required field checks. (4) Transform — field mapping, data cleaning, enrichment (lookup department names from codes). (5) Load — batch upsert to target with retry logic. (6) Reconcile — count check, sample verification. (7) Notify — email/Slack with results. (8) Archive — store raw extract for audit. Each step is idempotent and can be restarted independently.
class EmployeeSyncPipeline:
def __init__(self, config):
self.config = config
self.watermark = WatermarkManager(config["watermark_path"])
self.stats = {"extracted": 0, "valid": 0, "loaded": 0, "errors": []}
def run(self):
run_id = f"sync_{datetime.now():%Y%m%d_%H%M%S}"
print(f"[{run_id}] Starting employee sync")
try:
# Step 1: Extract
raw_records = self.extract()
self.stats["extracted"] = len(raw_records)
if not raw_records:
print("No changes detected")
return self.stats
# Step 2: Validate
valid, invalid = self.validate(raw_records)
self.stats["valid"] = len(valid)
# Step 3: Transform
transformed = self.transform(valid)
# Step 4: Load
result = self.load(transformed)
self.stats["loaded"] = result["success"]
# Step 5: Update watermark
max_ts = max(r["modified_date"] for r in raw_records)
self.watermark.advance(max_ts, run_id, len(valid))
# Step 6: Reconcile & Notify
self.reconcile()
self.notify(run_id)
except Exception as e:
self.stats["errors"].append(str(e))
self.notify_failure(run_id, e)
raise
return self.stats
def extract(self):
wm = self.watermark.get()
params = {"modified_since": wm} if wm else {}
return fetch_all_offset(self.config["source_url"],
headers=self.config["source_headers"])
def validate(self, records):
valid, invalid = [], []
for r in records:
try:
EmployeeSchema(**r)
valid.append(r)
except Exception as e:
invalid.append({"record": r, "error": str(e)})
return valid, invalid
def transform(self, records):
return [{
"employee_id": r["emp_code"],
"full_name": f"{r['first_name']} {r['last_name']}".strip(),
"email": r["email"].lower().strip(),
"department": r.get("dept_name", "Unknown"),
} for r in records]
def load(self, records):
return batch_upload(records, self.config["target_url"],
self.config["target_headers"])
Error handling strategy: (1) Isolate stages — each stage writes its output to disk/DB. If stage 3 fails, stage 2's output is preserved — restart from stage 3. (2) Partial failure — within a stage, continue processing valid records. Send failures to DLQ. (3) Checkpoint — after each batch, save progress. Resume from last checkpoint. (4) Alerting — immediate alert on failure with context (stage, record, error). (5) Retry policy — transient errors retry automatically; permanent errors go to DLQ. (6) Compensation — if stage 4 partially loaded before failing, mark those records for re-processing.
Logging levels: (1) INFO — pipeline started, records extracted/loaded, pipeline completed. (2) WARNING — retries, skipped records, anomalous counts. (3) ERROR — failures with full context. Every log entry includes: run_id, timestamp, stage, record count. Monitoring: track pipeline duration, success rate, record count trends. Alert on: failures, duration > 2x normal, count anomalies. Use structured logging (JSON) for searchability.
import logging, json
class PipelineLogger:
def __init__(self, run_id):
self.run_id = run_id
self.logger = logging.getLogger("pipeline")
self.metrics = {"start": datetime.utcnow().isoformat(), "stages": {}}
def stage_start(self, stage_name):
self.metrics["stages"][stage_name] = {"start": datetime.utcnow().isoformat()}
self.logger.info(json.dumps({
"run_id": self.run_id, "stage": stage_name, "event": "start"}))
def stage_end(self, stage_name, record_count):
self.metrics["stages"][stage_name]["end"] = datetime.utcnow().isoformat()
self.metrics["stages"][stage_name]["records"] = record_count
self.logger.info(json.dumps({
"run_id": self.run_id, "stage": stage_name,
"event": "complete", "records": record_count}))
def error(self, stage_name, error, record=None):
self.logger.error(json.dumps({
"run_id": self.run_id, "stage": stage_name,
"error": str(error), "record": record}))
Reconciliation is the process of comparing source and target data to verify they match. Types: (1) Count reconciliation — total records in source = total in target. (2) Value reconciliation — specific field values match (salary sums, employee details). (3) Key reconciliation — all primary keys in source exist in target. (4) Trend reconciliation — day-over-day counts are within expected range. Run automatically after every sync + weekly full reconciliation. Any mismatch triggers an alert and investigation.
Data quality dimensions: (1) Completeness — no missing required fields. (2) Accuracy — values are correct (email format, date ranges). (3) Consistency — same employee shouldn't have different DOBs in two systems. (4) Timeliness — data is current, not stale. (5) Uniqueness — no duplicates. Implement: validation rules per field, quarantine for bad records, data quality dashboard, and trend tracking (quality score over time).
class DataQualityChecker:
def __init__(self):
self.rules = []
self.results = []
def add_rule(self, field, check_fn, error_msg):
self.rules.append({"field": field, "check": check_fn, "msg": error_msg})
def validate(self, record):
errors = []
for rule in self.rules:
value = record.get(rule["field"])
if not rule["check"](value):
errors.append({"field": rule["field"], "value": value, "error": rule["msg"]})
return errors
# Setup quality rules
dq = DataQualityChecker()
dq.add_rule("email", lambda v: v and "@" in v, "Invalid email format")
dq.add_rule("emp_id", lambda v: v and len(v) > 0, "emp_id is required")
dq.add_rule("salary", lambda v: v is None or (isinstance(v, (int,float)) and v >= 0), "Salary must be non-negative")
dq.add_rule("join_date", lambda v: v and v > "1950-01-01", "Join date seems invalid")
A staging area is an intermediate storage between source and target. Can be: (1) A staging DB table. (2) A local file (CSV/JSON). (3) A message queue (Kafka, SQS). Benefits: (1) Decoupling — extract and load run independently. If load fails, extract's data is preserved. (2) Validation — validate in staging before touching production. (3) Auditability — staging preserves what was extracted. (4) Retry — re-load from staging without re-extracting. (5) Performance — bulk operations on staging are faster.
PII handling rules: (1) Encrypt in transit — HTTPS for APIs, SFTP (not FTP) for files. (2) Encrypt at rest — encrypt staging files/tables. PGP for SFTP files. (3) Mask in logs — never log full email, phone, SSN, Aadhaar. (4) Access control — restrict who can see PII in staging. (5) Data minimization — only transfer PII that's needed. (6) Retention — delete staging data after processing. (7) Audit trail — log who accessed PII when. (8) Compliance — follow DPDP (India), GDPR (EU), SOC2.
def mask_for_logging(record):
"""Mask PII before writing to logs"""
masked = record.copy()
pii_fields = ["email", "phone", "aadhaar", "pan", "bank_account"]
for field in pii_fields:
if field in masked and masked[field]:
masked[field] = f"{masked[field][:2]}***{masked[field][-2:]}"
return masked
# In pipeline: always log masked version
logger.info(f"Processing record: {mask_for_logging(record)}")
# Log shows: {email: "sh***om", phone: "98***45", aadhaar: "12***89"}
An idempotent pipeline produces the same result whether you run it once or ten times with the same input. This is crucial for reliability because pipelines get retried (manual re-runs, cron overlap, retry after failure). Design rules: (1) Use upsert, not insert (prevents duplicates). (2) Use deterministic transformations (same input always produces same output). (3) Checkpointed watermark (re-running fetches same data). (4) No side effects outside the pipeline (don't send notifications in retry mode). (5) Each stage reads from a stable input (file/staging), not directly from a live API that may have changed.
Pipeline dependencies: department sync must complete before employee sync (employees reference departments). Solutions: (1) Airflow DAGs — define dependencies as a Directed Acyclic Graph. Task B runs only after Task A succeeds. (2) Status files — pipeline A writes a .done file; pipeline B waits for it. (3) Event-driven — pipeline A publishes "complete" event; pipeline B subscribes. (4) Sequential runner — a master script runs pipelines in order.
# Simple dependency manager
class PipelineOrchestrator:
def __init__(self):
self.pipelines = {}
self.dependencies = {}
def register(self, name, fn, depends_on=None):
self.pipelines[name] = fn
self.dependencies[name] = depends_on or []
def run_all(self):
completed = set()
while len(completed) < len(self.pipelines):
for name, fn in self.pipelines.items():
if name in completed:
continue
deps_met = all(d in completed for d in self.dependencies[name])
if deps_met:
print(f"Running: {name}")
fn()
completed.add(name)
print(f"Completed: {name}")
# Usage
orch = PipelineOrchestrator()
orch.register("departments", sync_departments)
orch.register("designations", sync_designations)
orch.register("employees", sync_employees, depends_on=["departments", "designations"])
orch.register("attendance", sync_attendance, depends_on=["employees"])
orch.run_all()
# Order: departments -> designations -> employees -> attendance
Optimization areas: (1) Extract — parallel page fetching, field filtering, compression, connection pooling. (2) Transform — vectorized operations (pandas/polars instead of row-by-row), lazy evaluation. (3) Load — bulk insert (not row-by-row), batch API calls, async I/O. (4) Infrastructure — more CPU/RAM, SSD storage, faster network. (5) Architecture — incremental over full load, partitioned processing, streaming over batch. Measure first: profile each stage to find the bottleneck. Don't optimize what's already fast.
# Performance monitoring wrapper
import time
def timed_stage(name):
def decorator(fn):
def wrapper(*args, **kwargs):
start = time.time()
result = fn(*args, **kwargs)
elapsed = time.time() - start
record_count = len(result) if hasattr(result, '__len__') else 'N/A'
rate = len(result) / elapsed if elapsed > 0 and hasattr(result, '__len__') else 0
print(f"[PERF] {name}: {elapsed:.2f}s | {record_count} records | {rate:.0f} rec/sec")
return result
return wrapper
return decorator
@timed_stage("Extract")
def extract(): ...
@timed_stage("Transform")
def transform(records): ...
@timed_stage("Load")
def load(records): ...
# Output:
# [PERF] Extract: 12.34s | 5000 records | 405 rec/sec
# [PERF] Transform: 0.23s | 5000 records | 21739 rec/sec
# [PERF] Load: 45.67s | 5000 records | 109 rec/sec
# Bottleneck: Load stage (45s) -> optimize with parallel batch uploads
Built for PeopleStrong Integration Engineer Interview Prep
85+ Scenario-Based Questions | 50+ Code Examples | 5 Sections
All code examples are in Python. Concepts apply to any language.