Build a Security Data Pipeline
Pull real CVE data from NIST's NVD API and load it into Supabase.
Introduction
30 Second Summary
Every organization's security team deals with vulnerability data coming from multiple places. Scanners, government databases, and internal tools all produce reports in wildly different formats.
In this project, you will build a data integration pipeline that pulls real vulnerability records from a government API and a messy CSV file. You'll normalize both sources into a clean, queryable database that a client's team could use immediately.
What You'll Build
A Python pipeline that extracts security vulnerability data from two sources, validates and transforms it into a unified schema, and loads it into Supabase Postgres with production-grade error handling and observability.
By the end of this project, you'll have:
- A working multi-source data pipeline that pulls vulnerability records from the NVD API and a messy CSV file, normalizing both into a unified schema.
- Production-grade Python patterns for API resilience with retry logic, data validation with quarantine routing, and idempotent database upserts.
- A queryable vulnerability dataset in Supabase with structured JSON logging for full observability.
- Secret Mission: a real-time webhook receiver built with FastAPI that accepts security alert events and feeds them through the same pipeline.
Are there any prerequisites?
You should be comfortable with basic Python scripting and the Linux command line. You'll also need a free Supabase account, which you can create during the first step.
Build the NVD API Extractor
Your pipeline needs to pull real vulnerability data from an external source. NIST's National Vulnerability Database (NVD) publishes every known CVE (Common Vulnerabilities and Exposures), and its REST API gives you structured access to this data.
In this step, you'll build a resilient API extractor that handles pagination, rate limiting, and network failures gracefully. These are the exact patterns FDE interviewers test for when evaluating production-readiness.
In this step, get ready to:
- Create the extractors package with the NVD API module.
- Implement retry logic with exponential backoff and jitter.
- Build a paginated extraction function that pulls all matching CVEs.
Create the extractors package
Python uses packages (directories with an __init__.py file) to organize related modules. Your NVD extractor will live in an extractors/ package so you can add more data sources later.
- Create the extractors package directory and its __init__.py file by running these commands:
mkdir extractors
touch extractors/__init__.py- Confirm the files exist by running:
ls extractorsYou should see __init__.py in the output.
- Create a new file called nvd_api.py inside the extractors/ directory and open it in your editor.
- Add the following imports and configuration constants at the top of extractors/nvd_api.py:
import httpx
import time
import random
import logging
from datetime import datetime, timedelta, timezone
logger = logging.getLogger(__name__)
NVD_CVE_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
MAX_RETRIES = 3
BASE_DELAY = 6.0What does this code do?
- httpx is a modern Python HTTP client with built-in timeout handling and connection pooling.
- logger = logging.getLogger(__name__) creates a logger scoped to this module. Every API call will log structured data (URL, status code, timing) that you can later feed into log aggregators.
- BASE_DELAY = 6.0 follows the NVD's official recommendation to sleep six seconds between requests.
- MAX_RETRIES = 3 caps how many times the extractor retries a failed request before giving up.
Implement the retry function with exponential backoff
Real APIs fail. They rate-limit you, time out, or drop connections. Exponential backoff is the production pattern for handling this: wait longer between each retry attempt, and add random jitter so multiple clients don't all retry at the same instant.
- Add the fetch_with_retry() function below the constants in extractors/nvd_api.py:
def fetch_with_retry(client: httpx.Client, url: str, params: dict, headers: dict) -> dict:
"""Fetch URL with exponential backoff and jitter."""
for attempt in range(MAX_RETRIES):
start_time = time.time()
try:
response = client.get(url, params=params, headers=headers, timeout=30.0)
elapsed = time.time() - start_time
logger.info(
"NVD API request",
extra={
"url": url,
"status_code": response.status_code,
"attempt": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
},
)
if response.status_code == 200:
return response.json()
if response.status_code == 403:
logger.warning(
"Rate limited by NVD API",
extra={"attempt": attempt + 1},
)
else:
logger.error(
"NVD API error",
extra={
"status_code": response.status_code,
"attempt": attempt + 1,
},
)
except (httpx.TimeoutException, httpx.ConnectError) as e:
elapsed = time.time() - start_time
logger.error(
"NVD API connection error",
extra={
"error": str(e),
"attempt": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
},
)
if attempt < MAX_RETRIES - 1:
delay = BASE_DELAY * (2 ** attempt)
jitter = random.uniform(0, delay * 0.5)
sleep_time = delay + jitter
logger.info(
f"Retrying in {sleep_time:.2f}s",
extra={"attempt": attempt + 1, "sleep_seconds": round(sleep_time, 2)},
)
time.sleep(sleep_time)
raise RuntimeError(f"Failed to fetch {url} after {MAX_RETRIES} attempts")What does this code do?
This function wraps every HTTP request with production-grade resilience:
- On a successful 200 response, it returns the parsed JSON immediately.
- On HTTP 403 (the NVD's rate-limit signal), it logs a warning and retries after waiting.
- On timeouts or connection errors, it catches the exception and retries.
- The delay doubles each attempt (6s, 12s, 24s) and adds random jitter to prevent thundering herd problems.
- Every request logs structured data via the extra parameter. This makes logs machine-parseable for observability tools.
Why does the NVD return 403 instead of 429?
Most APIs use HTTP 429 (Too Many Requests) for rate limiting. The NVD is different. It returns HTTP 403 (Forbidden) when you exceed the rate limit. Your retry logic must handle 403 specifically.
Build the CVE extraction function with pagination
The NVD API returns results in pages. A single request might match hundreds of CVEs, but the API only sends a batch at a time. Your extraction function needs to loop through pages until it has all the results.
- Add the extract_recent_cves() function below fetch_with_retry() in extractors/nvd_api.py:
def extract_recent_cves(days_back: int = 7, api_key: str = None) -> list[dict]:
"""Extract recent CVEs from NVD API 2.0."""
end_date = datetime.now(timezone.utc)
start_date = end_date - timedelta(days=days_back)
params = {
"pubStartDate": start_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00"),
"pubEndDate": end_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00"),
"resultsPerPage": 20,
"startIndex": 0,
}
headers = {}
if api_key:
headers["apiKey"] = api_key
all_cves = []
with httpx.Client() as client:
while True:
data = fetch_with_retry(client, NVD_CVE_API_URL, params, headers)
vulnerabilities = data.get("vulnerabilities", [])
all_cves.extend(vulnerabilities)
total_results = data.get("totalResults", 0)
current_index = params["startIndex"] + len(vulnerabilities)
logger.info(
"NVD extraction progress",
extra={
"fetched": len(all_cves),
"total_available": total_results,
},
)
if current_index >= total_results:
break
params["startIndex"] = current_index
time.sleep(BASE_DELAY)
logger.info(f"Extracted {len(all_cves)} CVEs from NVD API")
return all_cvesWhat does this code do?
This function orchestrates pulling all matching CVEs from the NVD:
- pubStartDate and pubEndDate filter results to CVEs published in the last N days.
- startIndex and resultsPerPage handle pagination. After each page, the index advances by the number of records returned.
- The optional apiKey header increases the rate limit from 5 requests per 30 seconds to 50.
- A time.sleep(BASE_DELAY) between pages respects the NVD's rate limit recommendation.
- Save extractors/nvd_api.py.
- Verify the module is importable by running:
python -c "from extractors.nvd_api import extract_recent_cves, fetch_with_retry; print('Import successful')"You should see Import successful printed in your terminal.
Seeing an ImportError or ModuleNotFoundError?
- Make sure you have an __init__.py file inside the extractors/ directory. Without it, Python won't recognize the folder as a package.
- Confirm your virtual environment is still activated. You should see the environment name in your terminal prompt.
- Check for typos in the import statement or the filename. The file must be named exactly nvd_api.py (with an underscore, not a hyphen).
help me debug this ImportError in my extractors module
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import httpx
import time
import random
import logging
from datetime import datetime, timedelta, timezone
logger = logging.getLogger(__name__)
NVD_CVE_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
MAX_RETRIES = 3
BASE_DELAY = 6.0
def fetch_with_retry(client: httpx.Client, url: str, params: dict, headers: dict) -> dict:
"""Fetch URL with exponential backoff and jitter."""
for attempt in range(MAX_RETRIES):
start_time = time.time()
try:
response = client.get(url, params=params, headers=headers, timeout=30.0)
elapsed = time.time() - start_time
logger.info(
"NVD API request",
extra={
"url": url,
"status_code": response.status_code,
"attempt": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
},
)
if response.status_code == 200:
return response.json()
if response.status_code == 403:
logger.warning(
"Rate limited by NVD API",
extra={"attempt": attempt + 1},
)
else:
logger.error(
"NVD API error",
extra={
"status_code": response.status_code,
"attempt": attempt + 1,
},
)
except (httpx.TimeoutException, httpx.ConnectError) as e:
elapsed = time.time() - start_time
logger.error(
"NVD API connection error",
extra={
"error": str(e),
"attempt": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
},
)
if attempt < MAX_RETRIES - 1:
delay = BASE_DELAY * (2 ** attempt)
jitter = random.uniform(0, delay * 0.5)
sleep_time = delay + jitter
logger.info(
f"Retrying in {sleep_time:.2f}s",
extra={"attempt": attempt + 1, "sleep_seconds": round(sleep_time, 2)},
)
time.sleep(sleep_time)
raise RuntimeError(f"Failed to fetch {url} after {MAX_RETRIES} attempts")
def extract_recent_cves(days_back: int = 7, api_key: str = None) -> list[dict]:
"""Extract recent CVEs from NVD API 2.0."""
end_date = datetime.now(timezone.utc)
start_date = end_date - timedelta(days=days_back)
params = {
"pubStartDate": start_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00"),
"pubEndDate": end_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00"),
"resultsPerPage": 20,
"startIndex": 0,
}
headers = {}
if api_key:
headers["apiKey"] = api_key
all_cves = []
with httpx.Client() as client:
while True:
data = fetch_with_retry(client, NVD_CVE_API_URL, params, headers)
vulnerabilities = data.get("vulnerabilities", [])
all_cves.extend(vulnerabilities)
total_results = data.get("totalResults", 0)
current_index = params["startIndex"] + len(vulnerabilities)
logger.info(
"NVD extraction progress",
extra={
"fetched": len(all_cves),
"total_available": total_results,
},
)
if current_index >= total_results:
break
params["startIndex"] = current_index
time.sleep(BASE_DELAY)
logger.info(f"Extracted {len(all_cves)} CVEs from NVD API")
return all_cvesYour NVD extractor is ready to pull real vulnerability data with production-grade resilience. Next up, you'll build the CSV extractor and a data validator that catches messy records before they reach your database.
Build the CSV Extractor and Validator
Your NVD API extractor handles one source of vulnerability data. In a real Forward Deployed Engineer engagement, you rarely get the luxury of a single clean data source.
Clients hand you CSVs exported from internal security scanners. These files have inconsistent formats, missing fields, and duplicates. In this step, you'll build an extractor for messy CSV data and a validator that catches bad records before they reach your database.
In this step, get ready to:
- Create a sample CSV with intentionally dirty data to simulate real client exports.
- Build a CSV extractor that reads raw records from the file.
- Build a data validator that normalizes severity levels and quarantines bad records.
Create the messy sample CSV
Real client data is never clean. Security scanners export results in slightly different formats. Some records are missing CVE IDs. Others use non-standard severity labels like "H" instead of "HIGH" or "3" instead of a word.
You'll create a sample CSV that simulates all of these problems so your pipeline can handle them gracefully.
- Create a data/ directory inside your project folder by running:
mkdir data- Create a new file called internal_scan_results.csv inside the data/ directory and paste in the following content:
scan_id,cve_id,severity,description,affected_product,scan_date,scanner
SCAN-001,CVE-2024-21762,CRITICAL,FortiOS out-of-bound write vulnerability,FortiOS 7.4.0,2024-02-15,Nessus
SCAN-002,CVE-2024-3400,HIGH,PAN-OS Command Injection,Palo Alto PAN-OS 11.1,2024-04-12,Qualys
SCAN-003,CVE-2024-1709,high,ConnectWise ScreenConnect auth bypass,ScreenConnect 23.9.7,2024-02-20,Nessus
SCAN-004,,MEDIUM,Unknown vulnerability with missing CVE,Apache HTTP 2.4,2024-03-01,Rapid7
SCAN-005,CVE-2024-27198,H,JetBrains TeamCity authentication bypass,TeamCity 2023.11.3,03/05/2024,Qualys
SCAN-006,CVE-2024-21762,CRITICAL,FortiOS out-of-bound write vulnerability,FortiOS 7.4.0,2024-02-15,Nessus
SCAN-007,CVE-2024-4577,3,PHP CGI argument injection,PHP 8.3.7,2024-06-06,Nessus
SCAN-008,CVE-2024-28986,LOW,SolarWinds Web Help Desk hardcoded credential,Web Help Desk 12.8.3,,Qualys
SCAN-009,CVE-INVALID-FORMAT,HIGH,Bad CVE format test,TestProduct 1.0,2024-07-01,Rapid7
SCAN-010,CVE-2024-38077,CRITICAL,Windows RDL service RCE,Windows Server 2022,2024-07-09,NessusWhat's wrong with this data?
This CSV has several intentional problems that mirror real-world client data:
- SCAN-003 has lowercase severity (high instead of HIGH).
- SCAN-004 is missing its CVE ID entirely.
- SCAN-005 uses "H" for severity and a non-ISO date format (03/05/2024).
- SCAN-006 is a duplicate of SCAN-001.
- SCAN-007 uses a number ("3") for severity.
- SCAN-008 is missing its scan date.
- SCAN-009 has a malformed CVE ID (CVE-INVALID-FORMAT).
- Save data/internal_scan_results.csv.
- Confirm the file exists by running:
ls data/You should see internal_scan_results.csv in the output.
Don't see the file?
Make sure you created the data/ directory first with mkdir data before creating the file inside it. Check that your terminal is in the project root directory.
Help me troubleshoot my CSV file creation.
Build the CSV extractor
The CSV extractor has one job: read every row from the file and hand it off as a raw dictionary. It does not validate or transform the data. Keeping extraction separate from validation makes each piece easier to test and reuse.
- Create a new file called csv_source.py inside the extractors/ directory and paste in the following code:
import csv
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
def extract_scan_results(file_path: str = "data/internal_scan_results.csv") -> list[dict]:
"""Extract records from internal scan results CSV."""
path = Path(file_path)
if not path.exists():
logger.error(f"CSV file not found: {file_path}")
return []
records = []
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
records.append(dict(row))
logger.info(
f"Extracted {len(records)} records from CSV",
extra={"file": file_path, "record_count": len(records)},
)
return recordsWhat does this code do?
- csv.DictReader reads each row as a dictionary using the header row as keys. This means you get {"scan_id": "SCAN-001", "cve_id": "CVE-2024-21762", ...} instead of a raw list of strings.
- The function returns an empty list if the file is missing, logging an error instead of crashing. This is defensive programming for production code.
- Structured logging with the extra parameter attaches the file path and record count as machine-parseable fields.
- Save extractors/csv_source.py.
- Test the extractor by running this command from your project root:
python -c "from extractors.csv_source import extract_scan_results; records = extract_scan_results(); print(f'Extracted {len(records)} records'); print(records[0])"You should see Extracted 10 records followed by the first row printed as a dictionary.
Seeing a ModuleNotFoundError?
Make sure you're running the command from your project root directory (the folder containing the extractors/ folder). Also confirm that extractors/__init__.py exists from the previous step.
Help me fix my CSV extractor import error.
Build the data validator
The extractor pulls raw records without judging them. The validator's job is to decide which records are good enough for your database and which ones get quarantined with a rejection reason.
This separation is a key production pattern. Quarantined records are not thrown away. They get logged so a human can review and fix them later.
- Create the pipeline/ package directory and its __init__.py by running:
mkdir pipeline
touch pipeline/__init__.py- Confirm the directory exists by running:
ls pipeline/You should see __init__.py in the output.
The validator needs a mapping of all the non-standard severity labels your pipeline should recognize. This is where you define what "messy" looks like and how to clean it.
- Create a new file called validator.py inside the pipeline/ directory. Start by adding the imports and constant definitions:
import re
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
# Standard severity levels the pipeline accepts
VALID_SEVERITIES = {"CRITICAL", "HIGH", "MEDIUM", "LOW", "NONE"}
# Maps non-standard severity labels to their canonical form
SEVERITY_MAP = {
"critical": "CRITICAL",
"crit": "CRITICAL",
"c": "CRITICAL",
"4": "CRITICAL",
"high": "HIGH",
"h": "HIGH",
"3": "HIGH",
"medium": "MEDIUM",
"med": "MEDIUM",
"m": "MEDIUM",
"2": "MEDIUM",
"low": "LOW",
"l": "LOW",
"1": "LOW",
"none": "NONE",
"info": "NONE",
"informational": "NONE",
"0": "NONE",
}
# Regex to validate CVE ID format: CVE-YYYY-NNNN+
CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d{4,}$")What do these constants do?
- VALID_SEVERITIES is the set of severity levels your database accepts. Anything not in this set needs to be normalized or rejected.
- SEVERITY_MAP handles all the non-standard labels you'll see in real scanner exports. "H", "3", "high", and "h" all become "HIGH".
- CVE_PATTERN enforces the standard CVE format: the prefix "CVE-", a 4-digit year, a hyphen, and at least 4 digits. This catches malformed IDs like CVE-INVALID-FORMAT.
- Below the constants, add the normalize_severity() function:
def normalize_severity(raw_severity: str) -> str | None:
"""Normalize severity string to standard enum value."""
if not raw_severity:
return None
cleaned = raw_severity.strip().upper()
if cleaned in VALID_SEVERITIES:
return cleaned
return SEVERITY_MAP.get(raw_severity.strip().lower())What does normalize_severity do?
This function handles three scenarios. First, it returns None for empty input (triggering a quarantine later). Second, it checks if the uppercased value already matches a valid severity. Third, it looks up the lowercase value in SEVERITY_MAP for non-standard labels.
- Below normalize_severity(), add the main validate_record() function:
def validate_record(record: dict, source: str) -> tuple[dict | None, str | None]:
"""
Validate a record. Returns (validated_record, None) on success,
or (None, rejection_reason) on failure.
"""
# Check for required CVE ID
cve_id = record.get("cve_id", "").strip()
if not cve_id:
return None, "missing_cve_id"
# Validate CVE ID format
if not CVE_PATTERN.match(cve_id):
return None, f"invalid_cve_format: {cve_id}"
# Normalize and validate severity
raw_severity = record.get("severity", "")
severity = normalize_severity(raw_severity)
if severity is None:
return None, f"invalid_severity: {raw_severity}"
# Check for required description
description = record.get("description", "").strip()
if not description:
return None, "missing_description"
# Build the validated output record
validated = {
"cve_id": cve_id,
"source": source,
"severity": severity,
"description": description,
"affected_product": record.get("affected_product", "").strip() or None,
"published_date": record.get("published_date"),
"raw_data": record,
}
return validated, NoneWhat does validate_record do?
This function acts as a gatekeeper. It returns a tuple: either the cleaned record with None for the rejection reason (success), or None for the record with a human-readable rejection reason (failure).
- Records missing a CVE ID get rejected with missing_cve_id.
- Records with malformed CVE IDs get rejected with the specific bad value logged.
- Records with unrecognizable severity labels get rejected rather than guessing.
- Valid records are shaped into the unified schema with raw_data preserved for audit purposes.
Why return a tuple instead of raising exceptions?
Exceptions are expensive and break control flow. In a batch pipeline processing hundreds of records, you want to collect all failures without stopping. The tuple pattern lets the caller decide what to do with rejections (log them, quarantine them, retry them).
- Save pipeline/validator.py.
- Test the validator against your messy CSV by running:
python -c "
from extractors.csv_source import extract_scan_results
from pipeline.validator import validate_record
records = extract_scan_results()
for row in records:
record = {'cve_id': row.get('cve_id',''), 'severity': row.get('severity',''), 'description': row.get('description',''), 'affected_product': row.get('affected_product',''), 'published_date': row.get('scan_date','')}
validated, reason = validate_record(record, source='csv_test')
status = 'PASS' if validated else f'REJECT ({reason})'
print(f\"{row.get('scan_id')}: {status}\")
"You should see output showing SCAN-001, SCAN-002, SCAN-003, SCAN-006, SCAN-007, SCAN-008, and SCAN-010 passing, while SCAN-004 is rejected for missing_cve_id, SCAN-005 passes with severity normalized from "H" to "HIGH", and SCAN-009 is rejected for invalid_cve_format.
Seeing unexpected rejections?
If all records are being rejected, check that your SEVERITY_MAP keys are all lowercase strings. The function looks up raw_severity.strip().lower() in the map, so the keys must match that format.
If you see an import error for the pipeline package, confirm that pipeline/__init__.py exists (it can be empty).
Help me debug my validator output.
✔️ Awesome, I've got everything!
Great. Double check you've saved both extractors/csv_source.py and pipeline/validator.py.
ⓧ I'd like to double check the full code
Here is the complete extractors/csv_source.py:
import csv
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
def extract_scan_results(file_path: str = "data/internal_scan_results.csv") -> list[dict]:
"""Extract records from internal scan results CSV."""
path = Path(file_path)
if not path.exists():
logger.error(f"CSV file not found: {file_path}")
return []
records = []
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
records.append(dict(row))
logger.info(
f"Extracted {len(records)} records from CSV",
extra={"file": file_path, "record_count": len(records)},
)
return recordsAnd here is the complete pipeline/validator.py:
import re
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
VALID_SEVERITIES = {"CRITICAL", "HIGH", "MEDIUM", "LOW", "NONE"}
SEVERITY_MAP = {
"critical": "CRITICAL",
"crit": "CRITICAL",
"c": "CRITICAL",
"4": "CRITICAL",
"high": "HIGH",
"h": "HIGH",
"3": "HIGH",
"medium": "MEDIUM",
"med": "MEDIUM",
"m": "MEDIUM",
"2": "MEDIUM",
"low": "LOW",
"l": "LOW",
"1": "LOW",
"none": "NONE",
"info": "NONE",
"informational": "NONE",
"0": "NONE",
}
CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d{4,}$")
def normalize_severity(raw_severity: str) -> str | None:
"""Normalize severity string to standard enum value."""
if not raw_severity:
return None
cleaned = raw_severity.strip().upper()
if cleaned in VALID_SEVERITIES:
return cleaned
return SEVERITY_MAP.get(raw_severity.strip().lower())
def validate_record(record: dict, source: str) -> tuple[dict | None, str | None]:
"""
Validate a record. Returns (validated_record, None) on success,
or (None, rejection_reason) on failure.
"""
cve_id = record.get("cve_id", "").strip()
if not cve_id:
return None, "missing_cve_id"
if not CVE_PATTERN.match(cve_id):
return None, f"invalid_cve_format: {cve_id}"
raw_severity = record.get("severity", "")
severity = normalize_severity(raw_severity)
if severity is None:
return None, f"invalid_severity: {raw_severity}"
description = record.get("description", "").strip()
if not description:
return None, "missing_description"
validated = {
"cve_id": cve_id,
"source": source,
"severity": severity,
"description": description,
"affected_product": record.get("affected_product", "").strip() or None,
"published_date": record.get("published_date"),
"raw_data": record,
}
return validated, NoneYou now have both data sources covered: the NVD API extractor from the previous step, and a CSV extractor with a validation layer that catches bad data before it reaches your database. Next up, you'll build the transform and load layers that unify both sources into a single schema and push them into Supabase.
Build the Transform and Load Layers
You have extractors that pull raw data from two very different sources. You also have a validator that catches bad records before they enter the system.
Now you need the glue that turns raw data into a consistent shape and pushes it into your database. In this step, you'll build the transform layer (which maps both NVD and CSV records to your unified schema) and the load layer (which upserts records into Supabase without creating duplicates).
In this step, get ready to:
- Build transformer functions that map NVD API and CSV records to a single unified schema.
- Create an idempotent loader that upserts batches of records into Supabase.
- Add per-batch error handling that reports failures without crashing the pipeline.
Create the transformer module
The NVD API returns deeply nested JSON with optional CVSS scoring fields. Your CSV records come through the validator as flat dictionaries. The transformer's job is to produce the same output shape from both.
- Create a new file called pipeline/transformer.py and add the imports and the transform_nvd_record() function by pasting this code:
import logging
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
def transform_nvd_record(vuln: dict) -> dict:
"""Transform an NVD API vulnerability record to unified schema."""
cve = vuln.get("cve", {})
cve_id = cve.get("id", "")
# Extract the English description from the descriptions array
descriptions = cve.get("descriptions", [])
description = ""
for desc in descriptions:
if desc.get("lang") == "en":
description = desc.get("value", "")
break
# Extract severity from CVSS metrics (try v3.1, then v3.0, then v2)
severity = "NONE"
metrics = cve.get("metrics", {})
if "cvssMetricV31" in metrics:
severity = metrics["cvssMetricV31"][0].get("cvssData", {}).get("baseSeverity", "NONE")
elif "cvssMetricV30" in metrics:
severity = metrics["cvssMetricV30"][0].get("cvssData", {}).get("baseSeverity", "NONE")
elif "cvssMetricV2" in metrics:
base_score = metrics["cvssMetricV2"][0].get("cvssData", {}).get("baseScore", 0)
if base_score >= 9.0:
severity = "CRITICAL"
elif base_score >= 7.0:
severity = "HIGH"
elif base_score >= 4.0:
severity = "MEDIUM"
else:
severity = "LOW"
# Parse the published date into ISO format
published = cve.get("published", "")
published_date = None
if published:
try:
published_date = datetime.fromisoformat(published.replace("Z", "+00:00")).isoformat()
except ValueError:
published_date = None
# Extract affected product from CPE configurations
affected_product = None
configurations = cve.get("configurations", [])
if configurations:
nodes = configurations[0].get("nodes", [])
if nodes:
cpe_match = nodes[0].get("cpeMatch", [])
if cpe_match:
criteria = cpe_match[0].get("criteria", "")
parts = criteria.split(":")
if len(parts) >= 5:
affected_product = f"{parts[3]}:{parts[4]}" if parts[3] != "*" else None
return {
"cve_id": cve_id,
"source": "nvd_api",
"severity": severity.upper(),
"description": description[:1000],
"affected_product": affected_product,
"published_date": published_date,
"raw_data": vuln,
}What does this code do?
The NVD API nests vulnerability data several layers deep. This function reaches into that structure to extract what matters.
- It pulls the English description from a multilingual array.
- It tries three different CVSS metric versions (v3.1, v3.0, v2) in priority order, because not every CVE has the latest scoring.
- For v2 scores (which are numeric, not labels), it maps the number to a severity string using standard thresholds.
- It parses the CPE (Common Platform Enumeration) string to extract a human-readable product name.
- Save pipeline/transformer.py.
Seeing a syntax error?
Check that the f-string on the affected_product line has matching curly braces and a closing quote. Also confirm the return dict uses commas after every key-value pair.
Help me debug a syntax error in my transformer.py
- Next, add the transform_csv_record() function below transform_nvd_record() in the same file:
def transform_csv_record(record: dict) -> dict:
"""Transform a validated CSV record to unified schema."""
# Normalize the published date to ISO format
published_date = record.get("published_date")
if published_date:
try:
parsed = datetime.fromisoformat(published_date)
published_date = parsed.isoformat()
except ValueError:
published_date = None
return {
"cve_id": record["cve_id"],
"source": record["source"],
"severity": record["severity"],
"description": record["description"],
"affected_product": record.get("affected_product"),
"published_date": published_date,
"raw_data": record.get("raw_data"),
}What does this code do?
CSV records arrive already validated and normalized by validate_record(). This function only needs to parse the date into ISO format and pass everything else through to the unified schema.
The output dict has the exact same keys as transform_nvd_record(). That consistency is what makes loading simple.
- Save pipeline/transformer.py.
- Finally, add the transform_records() dispatcher function below transform_csv_record():
def transform_records(validated_records: list[dict], source_type: str) -> list[dict]:
"""Transform a list of validated records."""
transformed = []
for record in validated_records:
try:
if source_type == "nvd":
transformed.append(transform_nvd_record(record))
else:
transformed.append(transform_csv_record(record))
except Exception as e:
logger.error(
f"Transform error",
extra={"cve_id": record.get("cve_id", "unknown"), "error": str(e)},
)
return transformedWhat does this code do?
This is the public entry point that the orchestrator will call. It dispatches to the correct transformer based on source_type and wraps each transformation in a try/except so a single malformed record can't crash the entire batch.
- Save pipeline/transformer.py.
Seeing an import error for datetime?
Make sure the import at the top of the file reads from datetime import datetime, timezone. Both datetime and timezone are needed.
Help me fix my transformer.py imports
✔️ Awesome, I've got everything!
Great. Double check you've saved pipeline/transformer.py with all three functions.
ⓧ I'd like to double check the full code
import logging
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
def transform_nvd_record(vuln: dict) -> dict:
"""Transform an NVD API vulnerability record to unified schema."""
cve = vuln.get("cve", {})
cve_id = cve.get("id", "")
descriptions = cve.get("descriptions", [])
description = ""
for desc in descriptions:
if desc.get("lang") == "en":
description = desc.get("value", "")
break
severity = "NONE"
metrics = cve.get("metrics", {})
if "cvssMetricV31" in metrics:
severity = metrics["cvssMetricV31"][0].get("cvssData", {}).get("baseSeverity", "NONE")
elif "cvssMetricV30" in metrics:
severity = metrics["cvssMetricV30"][0].get("cvssData", {}).get("baseSeverity", "NONE")
elif "cvssMetricV2" in metrics:
base_score = metrics["cvssMetricV2"][0].get("cvssData", {}).get("baseScore", 0)
if base_score >= 9.0:
severity = "CRITICAL"
elif base_score >= 7.0:
severity = "HIGH"
elif base_score >= 4.0:
severity = "MEDIUM"
else:
severity = "LOW"
published = cve.get("published", "")
published_date = None
if published:
try:
published_date = datetime.fromisoformat(published.replace("Z", "+00:00")).isoformat()
except ValueError:
published_date = None
affected_product = None
configurations = cve.get("configurations", [])
if configurations:
nodes = configurations[0].get("nodes", [])
if nodes:
cpe_match = nodes[0].get("cpeMatch", [])
if cpe_match:
criteria = cpe_match[0].get("criteria", "")
parts = criteria.split(":")
if len(parts) >= 5:
affected_product = f"{parts[3]}:{parts[4]}" if parts[3] != "*" else None
return {
"cve_id": cve_id,
"source": "nvd_api",
"severity": severity.upper(),
"description": description[:1000],
"affected_product": affected_product,
"published_date": published_date,
"raw_data": vuln,
}
def transform_csv_record(record: dict) -> dict:
"""Transform a validated CSV record to unified schema."""
published_date = record.get("published_date")
if published_date:
try:
parsed = datetime.fromisoformat(published_date)
published_date = parsed.isoformat()
except ValueError:
published_date = None
return {
"cve_id": record["cve_id"],
"source": record["source"],
"severity": record["severity"],
"description": record["description"],
"affected_product": record.get("affected_product"),
"published_date": published_date,
"raw_data": record.get("raw_data"),
}
def transform_records(validated_records: list[dict], source_type: str) -> list[dict]:
"""Transform a list of validated records."""
transformed = []
for record in validated_records:
try:
if source_type == "nvd":
transformed.append(transform_nvd_record(record))
else:
transformed.append(transform_csv_record(record))
except Exception as e:
logger.error(
f"Transform error",
extra={"cve_id": record.get("cve_id", "unknown"), "error": str(e)},
)
return transformedCreate the loader with idempotent upserts
The loader's job is to push transformed records into Supabase without creating duplicates. If you run the pipeline twice with the same data, the second run should update existing rows instead of inserting new ones.
What is an idempotent upsert?
An upsert combines INSERT and UPDATE into one operation. If a record with the same conflict key already exists, it updates that row. If not, it inserts a new one.
This makes the pipeline idempotent. You can re-run it safely without worrying about duplicate data. FDE interviewers specifically test for this pattern because production pipelines must be safe to retry.
- Create a new file called pipeline/loader.py and paste this code:
import json
import logging
from datetime import datetime, timezone
from supabase import Client
logger = logging.getLogger(__name__)
BATCH_SIZE = 50
def load_to_supabase(client: Client, records: list[dict]) -> dict:
"""
Load records into Supabase with idempotent upserts.
Returns summary dict with counts.
"""
summary = {"inserted": 0, "failed": 0, "total": len(records)}
# Process records in batches of BATCH_SIZE
for i in range(0, len(records), BATCH_SIZE):
batch = records[i : i + BATCH_SIZE]
prepared_batch = []
# Prepare each record for Supabase insertion
for record in batch:
raw_data = record.get("raw_data")
prepared_batch.append(
{
"cve_id": record["cve_id"],
"source": record["source"],
"severity": record["severity"],
"description": record["description"],
"affected_product": record.get("affected_product"),
"published_date": record.get("published_date"),
"ingested_at": datetime.now(timezone.utc).isoformat(),
"raw_data": json.dumps(raw_data) if raw_data else None,
}
)
# Upsert the batch with conflict resolution on cve_id + source
try:
result = client.table("vulnerabilities").upsert(
prepared_batch,
on_conflict="cve_id,source",
).execute()
summary["inserted"] += len(prepared_batch)
logger.info(
f"Loaded batch",
extra={
"batch_start": i,
"batch_size": len(prepared_batch),
},
)
except Exception as e:
summary["failed"] += len(prepared_batch)
logger.error(
f"Batch load failed",
extra={
"batch_start": i,
"batch_size": len(prepared_batch),
"error": str(e),
},
)
return summaryWhat does this code do?
- BATCH_SIZE = 50 controls how many records are sent to Supabase in a single API call. Batching reduces network round trips.
- The on_conflict="cve_id,source" parameter tells Supabase to update (not duplicate) when a record with the same CVE ID and source already exists.
- raw_data is serialized to JSON because the Supabase column type is jsonb.
- ingested_at timestamps every load so you can track when each record entered your database.
- Save pipeline/loader.py.
- Confirm both new files exist by running:
ls pipeline/You should see __init__.py, validator.py, transformer.py, and loader.py in the output.
Getting an import error for supabase Client?
Make sure your virtual environment is activated and that you installed the supabase package in Step 1. The import from supabase import Client requires the supabase==2.31.0 package.
Help me fix my supabase import error in loader.py
✔️ Awesome, I've got everything!
Great. Double check you've saved pipeline/loader.py with the full function.
ⓧ I'd like to double check the full code
import json
import logging
from datetime import datetime, timezone
from supabase import Client
logger = logging.getLogger(__name__)
BATCH_SIZE = 50
def load_to_supabase(client: Client, records: list[dict]) -> dict:
"""
Load records into Supabase with idempotent upserts.
Returns summary dict with counts.
"""
summary = {"inserted": 0, "failed": 0, "total": len(records)}
for i in range(0, len(records), BATCH_SIZE):
batch = records[i : i + BATCH_SIZE]
prepared_batch = []
for record in batch:
raw_data = record.get("raw_data")
prepared_batch.append(
{
"cve_id": record["cve_id"],
"source": record["source"],
"severity": record["severity"],
"description": record["description"],
"affected_product": record.get("affected_product"),
"published_date": record.get("published_date"),
"ingested_at": datetime.now(timezone.utc).isoformat(),
"raw_data": json.dumps(raw_data) if raw_data else None,
}
)
try:
result = client.table("vulnerabilities").upsert(
prepared_batch,
on_conflict="cve_id,source",
).execute()
summary["inserted"] += len(prepared_batch)
logger.info(
f"Loaded batch",
extra={
"batch_start": i,
"batch_size": len(prepared_batch),
},
)
except Exception as e:
summary["failed"] += len(prepared_batch)
logger.error(
f"Batch load failed",
extra={
"batch_start": i,
"batch_size": len(prepared_batch),
"error": str(e),
},
)
return summaryReview the batch error handling pattern
The error handling in the loader is designed so a single failed batch does not kill the entire pipeline run. This is critical for production pipelines where partial success is better than total failure.
Why catch errors per batch instead of per record?
Supabase processes the entire upsert batch atomically. If one record in a batch has a schema mismatch, the whole batch fails. Catching at the batch level lets the pipeline continue with the next batch instead of aborting.
The summary dict (inserted, failed, total) gives the orchestrator everything it needs to report what happened. If 200 records were processed across 4 batches and one batch failed, the summary shows 150 inserted and 50 failed.
Why log with extra fields instead of f-strings?
The extra={} parameter passes structured data alongside the log message. When you set up the JSON formatter in the next step, these fields become searchable keys in the log output. This is how production observability works.
Your transform and load layers are ready. Next up, you'll wire everything together with an orchestrator that runs the full pipeline end-to-end and produces a run report.
Orchestrate and Run the Pipeline
You have all the building blocks in place: extractors that pull data from two sources, a validator that catches bad records, a transformer that normalizes everything, and a loader that upserts into Supabase.
Now you need an orchestrator to wire these pieces together into a single executable pipeline. The orchestrator is the control plane that decides which sources to pull from, routes data through validation and transformation, and produces a final report showing exactly what happened.
In this step, get ready to:
- Create the pipeline orchestrator that ties extraction, validation, transformation, and loading together.
- Create the entry point script with argument parsing and structured JSON logging.
- Run the pipeline end-to-end and verify results in Supabase.
Create the pipeline orchestrator
The orchestrator is responsible for loading environment variables, creating the Supabase client, calling each extractor, routing records through validation and transformation, and handing clean data to the loader. It returns a structured report dict so you can see exactly what happened during the run.
- Create a new file called pipeline/orchestrator.py and add the imports and environment setup:
import json
import logging
import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from supabase import create_client, Client
from extractors.nvd_api import extract_recent_cves
from extractors.csv_source import extract_scan_results
from pipeline.validator import validate_record
from pipeline.transformer import transform_nvd_record, transform_csv_record
from pipeline.loader import load_to_supabase
logger = logging.getLogger(__name__)
def run_pipeline(sources: list[str], days_back: int = 7) -> dict:
"""Run the full integration pipeline."""
load_dotenv()
supabase_url = os.environ.get("SUPABASE_URL")
supabase_key = os.environ.get("SUPABASE_KEY")
nvd_api_key = os.environ.get("NVD_API_KEY")
if not supabase_url or not supabase_key:
raise ValueError("SUPABASE_URL and SUPABASE_KEY must be set in .env")
client: Client = create_client(supabase_url, supabase_key)
report = {
"run_start": datetime.now(timezone.utc).isoformat(),
"sources": sources,
"extracted": 0,
"validated": 0,
"quarantined": 0,
"loaded": 0,
"failed": 0,
"quarantine_details": [],
}
all_transformed = []What does this code do?
The function loads environment variables from your .env file, creates a Supabase client, and initializes a report dictionary that tracks every metric about the run: how many records were extracted, validated, quarantined, loaded, and failed.
The all_transformed list will accumulate all records that pass validation and transformation before loading them in one batch.
- Below the all_transformed = [] line, add the NVD source extraction block:
if "nvd" in sources:
logger.info("Extracting from NVD API...")
nvd_records = extract_recent_cves(days_back=days_back, api_key=nvd_api_key)
report["extracted"] += len(nvd_records)
for vuln in nvd_records:
try:
transformed = transform_nvd_record(vuln)
all_transformed.append(transformed)
report["validated"] += 1
except Exception as e:
report["quarantined"] += 1
report["quarantine_details"].append(
{"source": "nvd", "error": str(e)}
)What does this code do?
When "nvd" is in the sources list, the orchestrator calls extract_recent_cves() with the configured days_back value and API key. NVD records are already structured JSON, so they go straight to the transformer without needing the validator.
If transformation fails for any record (malformed data), it gets quarantined instead of crashing the pipeline. This is production error handling in action.
- Below the NVD block, add the CSV source extraction with validation and the loading logic:
if "csv" in sources:
logger.info("Extracting from CSV...")
csv_records = extract_scan_results()
report["extracted"] += len(csv_records)
for row in csv_records:
record_for_validation = {
"cve_id": row.get("cve_id", ""),
"severity": row.get("severity", ""),
"description": row.get("description", ""),
"affected_product": row.get("affected_product", ""),
"published_date": row.get("scan_date", ""),
}
validated, rejection_reason = validate_record(
record_for_validation, source=f"csv_{row.get('scanner', 'unknown')}"
)
if validated:
transformed = transform_csv_record(validated)
all_transformed.append(transformed)
report["validated"] += 1
else:
report["quarantined"] += 1
report["quarantine_details"].append(
{
"source": "csv",
"scan_id": row.get("scan_id"),
"reason": rejection_reason,
}
)
if all_transformed:
logger.info(f"Loading {len(all_transformed)} records to Supabase...")
load_summary = load_to_supabase(client, all_transformed)
report["loaded"] = load_summary["inserted"]
report["failed"] = load_summary["failed"]
report["run_end"] = datetime.now(timezone.utc).isoformat()
logger.info("Pipeline run complete", extra={"report": report})
return reportWhat does this code do?
CSV records are messy, so each row goes through validate_record() first. The source is set to csv_{scanner_name} so you can trace which scanner produced each record.
Valid records get transformed and added to the batch. Invalid records are quarantined with their rejection reason for review later.
Finally, all transformed records are loaded to Supabase in one call. The report captures the final timestamp and returns everything the caller needs to understand what happened.
- Save pipeline/orchestrator.py.
Seeing an import error?
Make sure your pipeline/__init__.py and extractors/__init__.py files exist. Python needs these to recognize the folders as packages.
help me fix import errors in my pipeline orchestrator
✔️ Awesome, I've got everything!
Great. Double check you have saved pipeline/orchestrator.py.
ⓧ I'd like to double check the full code
import json
import logging
import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from supabase import create_client, Client
from extractors.nvd_api import extract_recent_cves
from extractors.csv_source import extract_scan_results
from pipeline.validator import validate_record
from pipeline.transformer import transform_nvd_record, transform_csv_record
from pipeline.loader import load_to_supabase
logger = logging.getLogger(__name__)
def run_pipeline(sources: list[str], days_back: int = 7) -> dict:
"""Run the full integration pipeline."""
load_dotenv()
supabase_url = os.environ.get("SUPABASE_URL")
supabase_key = os.environ.get("SUPABASE_KEY")
nvd_api_key = os.environ.get("NVD_API_KEY")
if not supabase_url or not supabase_key:
raise ValueError("SUPABASE_URL and SUPABASE_KEY must be set in .env")
client: Client = create_client(supabase_url, supabase_key)
report = {
"run_start": datetime.now(timezone.utc).isoformat(),
"sources": sources,
"extracted": 0,
"validated": 0,
"quarantined": 0,
"loaded": 0,
"failed": 0,
"quarantine_details": [],
}
all_transformed = []
if "nvd" in sources:
logger.info("Extracting from NVD API...")
nvd_records = extract_recent_cves(days_back=days_back, api_key=nvd_api_key)
report["extracted"] += len(nvd_records)
for vuln in nvd_records:
try:
transformed = transform_nvd_record(vuln)
all_transformed.append(transformed)
report["validated"] += 1
except Exception as e:
report["quarantined"] += 1
report["quarantine_details"].append(
{"source": "nvd", "error": str(e)}
)
if "csv" in sources:
logger.info("Extracting from CSV...")
csv_records = extract_scan_results()
report["extracted"] += len(csv_records)
for row in csv_records:
record_for_validation = {
"cve_id": row.get("cve_id", ""),
"severity": row.get("severity", ""),
"description": row.get("description", ""),
"affected_product": row.get("affected_product", ""),
"published_date": row.get("scan_date", ""),
}
validated, rejection_reason = validate_record(
record_for_validation, source=f"csv_{row.get('scanner', 'unknown')}"
)
if validated:
transformed = transform_csv_record(validated)
all_transformed.append(transformed)
report["validated"] += 1
else:
report["quarantined"] += 1
report["quarantine_details"].append(
{
"source": "csv",
"scan_id": row.get("scan_id"),
"reason": rejection_reason,
}
)
if all_transformed:
logger.info(f"Loading {len(all_transformed)} records to Supabase...")
load_summary = load_to_supabase(client, all_transformed)
report["loaded"] = load_summary["inserted"]
report["failed"] = load_summary["failed"]
report["run_end"] = datetime.now(timezone.utc).isoformat()
logger.info("Pipeline run complete", extra={"report": report})
return reportCreate the entry point with structured logging
The entry point script is what you actually run from the terminal. It handles argument parsing so you can choose which sources to process and how far back to look for CVEs. It also configures structured JSON logging so every log line is a parseable JSON object.
- Create a new file called run_pipeline.py in your project root and add the imports, JSON formatter, and logging setup:
import argparse
import json
import logging
import sys
class JSONFormatter(logging.Formatter):
"""Format log records as JSON for structured logging."""
def format(self, record):
log_entry = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
}
if hasattr(record, "url"):
log_entry["url"] = record.url
if hasattr(record, "status_code"):
log_entry["status_code"] = record.status_code
if hasattr(record, "attempt"):
log_entry["attempt"] = record.attempt
if hasattr(record, "elapsed_seconds"):
log_entry["elapsed_seconds"] = record.elapsed_seconds
if hasattr(record, "error"):
log_entry["error"] = record.error
if hasattr(record, "record_count"):
log_entry["record_count"] = record.record_count
if hasattr(record, "batch_start"):
log_entry["batch_start"] = record.batch_start
if hasattr(record, "batch_size"):
log_entry["batch_size"] = record.batch_size
return json.dumps(log_entry)
def setup_logging():
"""Configure structured JSON logging."""
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])Why structured JSON logging?
In production, logs go to aggregators like Datadog, ELK, or CloudWatch Logs. These tools parse JSON natively, so structured logs let you filter and search by any field (status code, attempt number, module).
The JSONFormatter class checks for extra fields that your extractors and loader attach via the extra={} parameter. This means every API call, batch load, and error gets machine-readable context automatically.
- Below setup_logging(), add the main() function and entry point:
def main():
parser = argparse.ArgumentParser(description="Security Data Integration Pipeline")
parser.add_argument(
"--source",
choices=["nvd", "csv", "all"],
default="all",
help="Data source to process (default: all)",
)
parser.add_argument(
"--days-back",
type=int,
default=7,
help="Number of days of CVE data to fetch from NVD (default: 7)",
)
args = parser.parse_args()
setup_logging()
logger = logging.getLogger(__name__)
sources = ["nvd", "csv"] if args.source == "all" else [args.source]
logger.info(
f"Starting pipeline",
extra={"sources": sources, "days_back": args.days_back},
)
from pipeline.orchestrator import run_pipeline
report = run_pipeline(sources=sources, days_back=args.days_back)
print("\n" + "=" * 60)
print("PIPELINE RUN REPORT")
print("=" * 60)
print(f" Sources: {', '.join(report['sources'])}")
print(f" Extracted: {report['extracted']} records")
print(f" Validated: {report['validated']} records")
print(f" Quarantined: {report['quarantined']} records")
print(f" Loaded: {report['loaded']} records")
print(f" Failed: {report['failed']} records")
print(f" Duration: {report['run_start']} -> {report['run_end']}")
if report["quarantine_details"]:
print("\n Quarantined Records:")
for item in report["quarantine_details"]:
print(f" - {item}")
print("=" * 60)
if __name__ == "__main__":
main()What does this code do?
The --source argument lets you run just the NVD extractor, just the CSV extractor, or both. The --days-back argument controls how far back the NVD API query looks for newly published CVEs.
After the pipeline finishes, the script prints a human-readable report showing extraction, validation, quarantine, and load counts. Any quarantined records are listed with their rejection reasons so you can diagnose data quality issues.
- Save run_pipeline.py.
Not sure where to save this file?
The run_pipeline.py file lives in the top-level project directory, alongside your requirements.txt and .env files.
help me figure out where to place my entry point script
✔️ Awesome, I've got everything!
Great. Double check you have saved run_pipeline.py.
ⓧ I'd like to double check the full code
import argparse
import json
import logging
import sys
class JSONFormatter(logging.Formatter):
"""Format log records as JSON for structured logging."""
def format(self, record):
log_entry = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
}
if hasattr(record, "url"):
log_entry["url"] = record.url
if hasattr(record, "status_code"):
log_entry["status_code"] = record.status_code
if hasattr(record, "attempt"):
log_entry["attempt"] = record.attempt
if hasattr(record, "elapsed_seconds"):
log_entry["elapsed_seconds"] = record.elapsed_seconds
if hasattr(record, "error"):
log_entry["error"] = record.error
if hasattr(record, "record_count"):
log_entry["record_count"] = record.record_count
if hasattr(record, "batch_start"):
log_entry["batch_start"] = record.batch_start
if hasattr(record, "batch_size"):
log_entry["batch_size"] = record.batch_size
return json.dumps(log_entry)
def setup_logging():
"""Configure structured JSON logging."""
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
def main():
parser = argparse.ArgumentParser(description="Security Data Integration Pipeline")
parser.add_argument(
"--source",
choices=["nvd", "csv", "all"],
default="all",
help="Data source to process (default: all)",
)
parser.add_argument(
"--days-back",
type=int,
default=7,
help="Number of days of CVE data to fetch from NVD (default: 7)",
)
args = parser.parse_args()
setup_logging()
logger = logging.getLogger(__name__)
sources = ["nvd", "csv"] if args.source == "all" else [args.source]
logger.info(
f"Starting pipeline",
extra={"sources": sources, "days_back": args.days_back},
)
from pipeline.orchestrator import run_pipeline
report = run_pipeline(sources=sources, days_back=args.days_back)
print("\n" + "=" * 60)
print("PIPELINE RUN REPORT")
print("=" * 60)
print(f" Sources: {', '.join(report['sources'])}")
print(f" Extracted: {report['extracted']} records")
print(f" Validated: {report['validated']} records")
print(f" Quarantined: {report['quarantined']} records")
print(f" Loaded: {report['loaded']} records")
print(f" Failed: {report['failed']} records")
print(f" Duration: {report['run_start']} -> {report['run_end']}")
if report["quarantine_details"]:
print("\n Quarantined Records:")
for item in report["quarantine_details"]:
print(f" - {item}")
print("=" * 60)
if __name__ == "__main__":
main()Run the pipeline and verify results
Time to see everything work together. You will start with just the CSV source to get fast feedback, then run the full pipeline including the NVD API.
- Run the pipeline with only the CSV source by running this command:
python run_pipeline.py --source csvYou should see structured JSON log lines printing to your terminal, followed by a formatted pipeline run report.
What should I see?
The report should show 10 extracted records (the full CSV), around 7 validated records, and 3 quarantined records. The quarantined records are SCAN-004 (missing CVE ID), SCAN-006 (duplicate that gets upserted, not quarantined at validation but handled by the loader), and SCAN-009 (invalid CVE format).
Your structured logs will appear as JSON objects with fields like timestamp, level, message, and module. This is what production log aggregators expect.
- Now run the full pipeline with both sources by running:
python run_pipeline.py --source all --days-back 7Why does the NVD call take longer?
The NVD API has a public rate limit of 5 requests per 30 seconds. Your fetch_with_retry() function sleeps 6 seconds between requests to respect this limit. If you have an NVD API key in your .env file, the limit increases to 50 requests per 30 seconds.
- Switch to the Supabase dashboard in your browser.
- Click Table Editor in the left sidebar.
- Select the vulnerabilities table.
- Verify that rows exist with different source values (nvd_api, csv_Nessus, csv_Qualys).
You should see vulnerability records populated in the table with severity levels, descriptions, and timestamps.
What about idempotency?
Run the pipeline again with the same command. You will get the same report, but no duplicate rows appear in Supabase. That is the upsert pattern at work. The composite key (cve_id + source) ensures existing records get updated in place instead of duplicated.
- Review the quarantine details in your terminal output. Look for the rejection reasons to confirm your validator is catching the intentional data quality issues in the CSV.
Pipeline failing with a connection error?
Check that your .env file has the correct SUPABASE_URL and SUPABASE_KEY values. The URL should look like https://yourproject.supabase.co and the key is the anon public key from your Supabase project settings.
If the NVD API is timing out, try running with just --source csv first to confirm your Supabase connection works. NVD rate limiting or network issues will not affect the CSV path.
help me debug my pipeline connection errors
Your pipeline is complete. You have a working end-to-end system that extracts from multiple sources, validates and quarantines bad data, transforms everything to a unified schema, and loads it idempotently into Supabase. Next up, you will take this further with a real-time webhook receiver in the Secret Mission.
Secret mission
Add a Real-Time Webhook Receiver
Your batch pipeline runs on a schedule. But what happens when a SIEM fires a critical alert right now? Build a FastAPI endpoint that accepts real-time security alert webhooks and feeds them through the same validation, transform, and load pipeline you already built.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally and uses Supabase's free tier, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Supabase vulnerabilities table (cloud database table with ingested vulnerability data).
- Local project directory with Python virtual environment, pipeline code, and sample data.
Keep everything running
No action needed. Choose this if you're still actively building or plan to extend the pipeline further.
- Your Supabase vulnerabilities table stays on the free tier with no charges.
- Your local project files remain on your Desktop ready for future runs.
- Your Python virtual environment stays intact with all installed packages.
Pause - I'll come back to this later
Free up local resources while keeping your cloud data intact for later.
- If the FastAPI webhook server is still running, stop it by pressing Ctrl+C in the terminal.
- Deactivate your virtual environment by running:
deactivate- Your Supabase table and data remain available at no cost whenever you return.
Delete - I don't want to use this again
Remove all project resources and start fresh.
- Drop the Supabase table by opening the SQL Editor in your Supabase dashboard and running this query:
DROP TABLE IF EXISTS vulnerabilities;- Delete your local project directory by running these commands in your terminal:
cd ~/Desktop
rm -rf security-data-pipeline- On Windows, use this command instead:
cd ~\Desktop
Remove-Item -Recurse -Force security-data-pipeline- This removes all Python files, the virtual environment, the .env file with your credentials, and the sample CSV data.
Nice Work!
Nice Work!
You just built what a Forward Deployed Engineer delivers in their first week on a client engagement. You took messy, multi-source security data and turned it into a clean, queryable vulnerability dataset in Supabase Postgres.
You've learned how to:
- Build a multi-source data integration pipeline that extracts from the NVD REST API and a messy CSV, validates and normalizes the data, and loads it into Supabase with idempotent upserts for safe re-runs.
- Implement production-grade retry logic with exponential backoff and jitter, data validation with quarantine routing for bad records, and structured JSON logging for observability.
- Design a modular pipeline architecture that cleanly separates extract, transform, validate, and load layers into reusable components.
- Build a real-time webhook receiver with HMAC-SHA256 signature verification that feeds incoming security alerts through the same pipeline.
Ready to quiz yourself?