DevSecOps RAG Insight Consolidator
Build a vector database pipeline that answers DevSecOps questions with AI.
Introduction
30 Second Summary
Imagine you're a security lead drowning in reports from four different tools. Each tool has its own format, its own dashboard, and its own way of describing the same problems. Finding patterns across all of them means hours of manual cross-referencing.
In this project, you will build an AI-powered analytics platform that ingests DevSecOps reports into a vector database and answers natural-language questions about them using retrieval-augmented generation.
What You'll Build
A CLI-based analytics consolidator that unifies reports from four internal tools into a single PostgreSQL vector database, then queries them with natural language to get AI-synthesized answers.
By the end of this project, you'll have:
- A pgvector schema storing 1536-dimensional vector embeddings from DevSecOps reports.
- A CLI ingestion pipeline that parses CSVs, validates data, generates OpenAI embeddings, and bulk-inserts records into the database.
- A FastAPI REST API with filtering, aggregation, and single-record endpoints backed by the same database.
- A RAG query engine that embeds your question, performs cosine similarity search, and synthesizes answers with GPT-4o.
- Secret Mission: Build an agentic loop that classifies questions and autonomously decides between semantic search, structured filtering, or summarization.
Are there any prerequisites?
You should be comfortable with Python basics, have some familiarity with SQL, and feel at home on the command line. No prior experience with vector databases or RAG is required.
Build the Database Layer
Your database tables and vector extension are ready from the previous step. Now you need a way for your Python code to talk to PostgreSQL reliably.
In this step, you'll write a connection module that handles opening connections, registering the pgvector type adapter, and cleaning up automatically. Then you'll verify that everything works end to end.
In this step, get ready to:
- Write the database connection module with context managers.
- Understand the schema design that powers the vector database.
- Verify the database connection, tables, and index.
Write the connection module
Every module in this project that touches the database will import from a single db/connection.py file. This keeps connection logic in one place and ensures every connection registers the pgvector type adapter.
- Create a new file called connection.py inside your db folder.
- Add the imports and configuration loader by pasting this code:
import os
from contextlib import contextmanager
import psycopg2
from dotenv import load_dotenv
from pgvector.psycopg2 import register_vector
load_dotenv()
def get_connection_params():
return {
"host": os.getenv("DB_HOST", "localhost"),
"port": os.getenv("DB_PORT", "5432"),
"dbname": os.getenv("DB_NAME", "insight_consolidator"),
"user": os.getenv("DB_USER", "postgres"),
"password": os.getenv("DB_PASSWORD", "postgres"),
}What does this code do?
load_dotenv() reads your .env file and loads its values into environment variables. This means your credentials never need to be hardcoded.
get_connection_params() builds the dictionary that psycopg2 needs to connect. Each value falls back to a sensible default if the environment variable is missing.
- Below get_connection_params(), add the main connection context manager:
@contextmanager
def get_db_connection():
conn = psycopg2.connect(**get_connection_params())
register_vector(conn)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()What does this code do?
get_db_connection() is a Python context manager. When you use it with a with statement, it opens a connection, registers the pgvector type adapter, and yields the connection for you to use.
register_vector(conn) tells psycopg2 how to handle the vector column type. Without this call, PostgreSQL would return raw strings instead of usable arrays.
If your code succeeds, the connection commits automatically. If an exception occurs, it rolls back. Either way, the connection closes in the finally block.
- Below get_db_connection(), add the cursor convenience helper:
@contextmanager
def get_db_cursor():
with get_db_connection() as conn:
cur = conn.cursor()
try:
yield cur
finally:
cur.close()What does this code do?
get_db_cursor() is a shorthand for when you only need a cursor. It opens a connection (with pgvector registered), creates a cursor, yields it, and handles cleanup of both the cursor and connection.
- Save db/connection.py.
Seeing a ModuleNotFoundError?
If Python cannot find psycopg2 or pgvector, make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.
help me fix a ModuleNotFoundError for psycopg2 or pgvector in my project
✔️ Awesome, I've got everything!
Great. Double check you've saved db/connection.py before moving on.
ⓧ I'd like to double check the full code
import os
from contextlib import contextmanager
import psycopg2
from dotenv import load_dotenv
from pgvector.psycopg2 import register_vector
load_dotenv()
def get_connection_params():
return {
"host": os.getenv("DB_HOST", "localhost"),
"port": os.getenv("DB_PORT", "5432"),
"dbname": os.getenv("DB_NAME", "insight_consolidator"),
"user": os.getenv("DB_USER", "postgres"),
"password": os.getenv("DB_PASSWORD", "postgres"),
}
@contextmanager
def get_db_connection():
conn = psycopg2.connect(**get_connection_params())
register_vector(conn)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
@contextmanager
def get_db_cursor():
with get_db_connection() as conn:
cur = conn.cursor()
try:
yield cur
finally:
cur.close()Understand the schema design
The connection module you just wrote will interact with the two tables you created in the previous step. Let's review what your schema gives you.
- Switch back to db/schema.sql in your editor to review the table structure.
What do these tables do?
- reports tracks each ingested file. It records the filename, which source system generated it (CodeScan, ARM, Vault, or Guard), and when it was ingested.
- insights stores individual findings from those reports. Each insight has a category, severity, date, the raw text, optional tags, and a 1536-dimensional vector embedding column.
- The embedding vector(1536) column is where pgvector stores the OpenAI embedding for each insight. This is what makes semantic search possible.
- The ivfflat index with vector_cosine_ops speeds up cosine similarity queries. Instead of comparing your query vector against every single row, the index partitions vectors into 100 lists for approximate nearest-neighbor search.
Why ivfflat instead of HNSW?
IVFFlat has faster build times and lower memory usage than HNSW. For a learning project with sample data, it builds quickly and demonstrates the core concept of approximate nearest-neighbor indexing. HNSW would be a better choice for production workloads with frequent inserts.
Verify the database connection
Now let's confirm that your connection module can reach the database and that the tables, extension, and index are all in place.
- Run this verification script in your terminal:
python -c "
from db.connection import get_db_cursor
with get_db_cursor() as cur:
# Check that both tables exist
cur.execute(\"SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name IN ('reports', 'insights') ORDER BY table_name\")
tables = [row[0] for row in cur.fetchall()]
print(f'Tables found: {tables}')
# Check that the vector extension is enabled
cur.execute(\"SELECT extname, extversion FROM pg_extension WHERE extname = 'vector'\")
ext = cur.fetchone()
print(f'Vector extension: {ext[0]} v{ext[1]}')
# Check that the ivfflat index exists
cur.execute(\"SELECT indexname FROM pg_indexes WHERE indexname = 'insights_embedding_idx'\")
idx = cur.fetchone()
print(f'Index found: {idx[0]}')
print('All checks passed!')
"You should see output showing both tables, the vector extension version, and the index name.
What should I see?
Your terminal should display something like:
Tables found: ['insights', 'reports']
Vector extension: vector v0.8.0
Index found: insights_embedding_idx
All checks passed!The exact vector extension version may differ depending on your PostgreSQL installation. The important thing is that all three checks print their values without errors.
Getting a connection refused error?
Make sure PostgreSQL is running. On Windows, check that the PostgreSQL service is started in the Services app. On macOS/Linux, try running pg_isready to check if the server is accepting connections.
Also verify that the values in your .env file match your actual PostgreSQL credentials. The default username is postgres and the default port is 5432.
help me fix a PostgreSQL connection refused error in my Python project
Seeing vector type not found in the database?
This means the pgvector extension was not created in the insight_consolidator database. Run your schema file again:
psql -U postgres -d insight_consolidator -f db/schema.sqlThe CREATE EXTENSION IF NOT EXISTS vector statement must run against the same database your Python code connects to.
help me fix the vector type not found error in pgvector
Your database layer is locked in. You have a clean connection module that any part of your application can import, and you've confirmed the tables and vector index are ready. Next up, you'll build the ingestion pipeline that reads CSV files, generates embeddings, and loads them into these tables.
Build the CSV Ingestion Pipeline
Your database is ready to store vector embeddings. Now you need a way to get data into it.
In this step, you'll build a CLI tool that reads CSV reports from your DevSecOps tools, generates OpenAI embeddings for each row, and bulk-inserts everything into PostgreSQL. The pipeline validates data before touching the API, retries on transient failures, and shows a progress bar so you know exactly where things stand.
In this step, get ready to:
- Build a CSV parser that validates rows against the expected schema.
- Implement an embedding generator with retry logic using the OpenAI API.
- Wire up bulk insert into PostgreSQL with a Rich progress bar.
Build the CSV parser and validator
Before calling the OpenAI API (which costs money per token), you want to reject bad data early. A validator that checks each row against the expected schema catches problems before any network calls happen.
- Create an ingestion folder inside your project root by running these commands:
- The __init__.py file makes Python treat the folder as an importable package.
mkdir ingestion
touch ingestion/__init__.py- Confirm the folder exists by running:
ls ingestionYou should see __init__.py in the output.
- Create a file called ingest.py inside the ingestion folder.
- Add the following imports, app initialization, validation constants, and validate_row() function:
import csv
import sys
from datetime import date
from pathlib import Path
import numpy as np
import typer
from openai import OpenAI
from psycopg2.extras import execute_values
from rich.console import Console
from rich.progress import track
from tenacity import retry, stop_after_attempt, wait_exponential
from db.connection import get_db_connection
app = typer.Typer()
console = Console()
client = OpenAI()
VALID_SOURCES = {"CodeScan", "ARM", "Vault", "Guard"}
VALID_CATEGORIES = {"code_quality", "security", "deployment", "backup"}
VALID_SEVERITIES = {"low", "medium", "high", "critical"}
def validate_row(row: dict, row_num: int) -> dict:
errors = []
if "category" not in row or row["category"] not in VALID_CATEGORIES:
errors.append(f"Row {row_num}: invalid category '{row.get('category', '')}'")
if "severity" not in row or row["severity"] not in VALID_SEVERITIES:
errors.append(f"Row {row_num}: invalid severity '{row.get('severity', '')}'")
if "raw_text" not in row or not row["raw_text"].strip():
errors.append(f"Row {row_num}: raw_text is empty")
if "insight_date" not in row or not row["insight_date"].strip():
errors.append(f"Row {row_num}: insight_date is missing")
else:
try:
date.fromisoformat(row["insight_date"])
except ValueError:
errors.append(f"Row {row_num}: invalid date format '{row['insight_date']}'")
if errors:
raise ValueError("\n".join(errors))
return rowWhat does this code do?
The module starts by initializing three key objects: a Typer app for CLI argument parsing, a Rich console for formatted terminal output, and an OpenAI client that reads your API key from the environment automatically.
The three sets (VALID_SOURCES, VALID_CATEGORIES, VALID_SEVERITIES) define the only values your database constraints allow. The validate_row() function checks each CSV row against these sets and also validates the date format using Python's date.fromisoformat(). If any check fails, it raises a ValueError with all accumulated errors for that row.
- Save ingestion/ingest.py.
Seeing an import error?
Make sure you created __init__.py inside the ingestion folder. Without it, Python cannot import the module. Also confirm your virtual environment is activated.
help me fix import errors in my ingestion module
Implement the embedding generator
Each row's raw_text needs to be converted into a 1536-dimensional vector using OpenAI's text-embedding-3-small model. Network calls can fail transiently, so you'll wrap the API call with retry logic.
- In ingestion/ingest.py, add the following function below validate_row():
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embeddings(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
input=texts,
model="text-embedding-3-small",
)
return [item.embedding for item in response.data]What does this code do?
The @retry decorator from tenacity automatically retries the function up to 3 times if it raises any exception. The wait between retries grows exponentially starting at 2 seconds, capping at 10 seconds.
The function accepts a list of text strings and returns a list of embedding vectors. This batch approach is more efficient than embedding one string at a time because it reduces the number of API round-trips.
- Save ingestion/ingest.py.
Why batch embeddings?
OpenAI's embedding endpoint accepts multiple texts in a single request. Sending 50 texts at once instead of 50 individual calls saves time and reduces the chance of hitting rate limits.
Wire up bulk insert with Rich progress
Now you'll write the main CLI command that ties everything together: parse the CSV, generate embeddings in batches of 50, insert a parent report record, and bulk-insert all insights with their embeddings.
- In ingestion/ingest.py, add the following ingest() command and main block below generate_embeddings():
@app.command()
def ingest(
file: Path = typer.Option(..., help="Path to the CSV file"),
source: str = typer.Option(..., help="Source system: CodeScan, ARM, Vault, or Guard"),
):
"""Ingest a CSV report into the insight consolidator."""
if source not in VALID_SOURCES:
console.print(f"[red]Error:[/red] Invalid source '{source}'. Must be one of: {VALID_SOURCES}")
raise typer.Exit(code=1)
if not file.exists():
console.print(f"[red]Error:[/red] File not found: {file}")
raise typer.Exit(code=1)
console.print(f"[bold blue]Ingesting[/bold blue] {file.name} from {source}...")
rows = []
with open(file, newline="", encoding="utf-8") as csvfile:
reader = csv.DictReader(csvfile)
for i, row in enumerate(reader, start=1):
validate_row(row, i)
rows.append(row)
console.print(f"[green]Validated {len(rows)} rows[/green]")
batch_size = 50
all_embeddings = []
texts = [row["raw_text"] for row in rows]
for i in track(range(0, len(texts), batch_size), description="Generating embeddings..."):
batch = texts[i : i + batch_size]
embeddings = generate_embeddings(batch)
all_embeddings.extend(embeddings)
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO reports (filename, source_system) VALUES (%s, %s) RETURNING id",
(file.name, source),
)
report_id = cur.fetchone()[0]
values = []
for row, embedding in zip(rows, all_embeddings):
tags = [t.strip() for t in row.get("tags", "").split(",") if t.strip()]
values.append((
report_id,
row["category"],
row["severity"],
row["insight_date"],
row["raw_text"],
tags,
np.array(embedding),
))
execute_values(
cur,
"""INSERT INTO insights
(report_id, category, severity, insight_date, raw_text, tags, embedding)
VALUES %s""",
values,
template="(%s, %s, %s, %s, %s, %s, %s)",
)
cur.close()
console.print(
f"[bold green]Success![/bold green] Ingested {len(rows)} insights from {file.name} (report_id={report_id})"
)
if __name__ == "__main__":
app()What does this code do?
The @app.command() decorator registers ingest() as a Typer CLI command. The two typer.Option(...) parameters make --file and --source required flags when the learner runs the script.
The function first validates the source system and file existence, then reads and validates every CSV row. Next it generates embeddings in batches of 50 using the track() wrapper from Rich, which displays a progress bar in the terminal.
Finally, it opens a database connection, inserts a parent reports record to track the source file, and uses execute_values() from psycopg2 to bulk-insert all insights with their embedding vectors. The np.array() wrapper is required because pgvector expects numpy arrays, not plain Python lists.
- Save ingestion/ingest.py.
Why numpy arrays for pgvector?
Passing plain Python lists to pgvector's cosine distance operator causes a type mismatch error. The pgvector Python adapter requires numpy arrays so it can serialize the data correctly for PostgreSQL's vector type.
Now create the sample data file that the pipeline will ingest.
- Create the data/samples directory by running:
mkdir -p data/samples- Create a file called codescan_report.csv inside data/samples and paste the following sample data:
category,severity,insight_date,raw_text,tags
code_quality,high,2024-11-15,"SQL injection vulnerability detected in UserController.java line 42. Raw query concatenation without parameterized queries.","sql-injection,java,controller"
code_quality,medium,2024-11-15,"Cyclomatic complexity of 25 in PaymentService.processPayment() exceeds threshold of 15.","complexity,java,payment"
security,critical,2024-11-14,"Hardcoded AWS credentials found in config/database.yml. Access key AKIA*** exposed.","credentials,aws,config"
security,high,2024-11-14,"Outdated dependency log4j 2.14.1 with known CVE-2021-44228 (Log4Shell) detected in pom.xml.","dependency,log4j,cve"
deployment,low,2024-11-13,"Deployment pipeline took 45 minutes exceeding 30-minute SLA. Bottleneck in integration test stage.","pipeline,slow,integration-tests"
backup,medium,2024-11-13,"Vault backup for production database completed with 3 warning: 2 tables skipped due to lock timeout.","backup,timeout,production"
code_quality,low,2024-11-12,"Documentation coverage at 34% for api-gateway module. Below 60% threshold.","documentation,coverage,api-gateway"
security,high,2024-11-12,"Cross-site scripting (XSS) vulnerability in search endpoint. User input rendered without sanitization.","xss,frontend,sanitization"What is this sample data?
This CSV simulates a report from CodeScan, one of the four fictional DevSecOps tools your consolidator supports. It contains 8 rows spanning all four categories (code_quality, security, deployment, backup) and multiple severity levels. Each row's raw_text is what gets embedded into a vector for semantic search later.
- Save data/samples/codescan_report.csv.
- Make sure your .env file has a valid OpenAI API key in the OPENAI_API_KEY field (replace your-api-key-here with your actual key).
- Run the ingestion pipeline by executing this command from your project root:
python -m ingestion.ingest --file data/samples/codescan_report.csv --source CodeScanYou should see a progress bar while embeddings generate, followed by a success message showing 8 ingested insights and a report ID.
Seeing an authentication error?
If you see an error mentioning an invalid API key, open your .env file and confirm OPENAI_API_KEY contains your actual OpenAI key. The placeholder value your-api-key-here does not work.
If you see a database connection error, make sure PostgreSQL is running and the DB_PASSWORD in .env matches your PostgreSQL password.
help me debug my ingestion pipeline errors
✔️ Awesome, I've got everything!
Great. Double check that you've saved both files.
ⓧ I'd like to double check the full code
import csv
import sys
from datetime import date
from pathlib import Path
import numpy as np
import typer
from openai import OpenAI
from psycopg2.extras import execute_values
from rich.console import Console
from rich.progress import track
from tenacity import retry, stop_after_attempt, wait_exponential
from db.connection import get_db_connection
app = typer.Typer()
console = Console()
client = OpenAI()
VALID_SOURCES = {"CodeScan", "ARM", "Vault", "Guard"}
VALID_CATEGORIES = {"code_quality", "security", "deployment", "backup"}
VALID_SEVERITIES = {"low", "medium", "high", "critical"}
def validate_row(row: dict, row_num: int) -> dict:
errors = []
if "category" not in row or row["category"] not in VALID_CATEGORIES:
errors.append(f"Row {row_num}: invalid category '{row.get('category', '')}'")
if "severity" not in row or row["severity"] not in VALID_SEVERITIES:
errors.append(f"Row {row_num}: invalid severity '{row.get('severity', '')}'")
if "raw_text" not in row or not row["raw_text"].strip():
errors.append(f"Row {row_num}: raw_text is empty")
if "insight_date" not in row or not row["insight_date"].strip():
errors.append(f"Row {row_num}: insight_date is missing")
else:
try:
date.fromisoformat(row["insight_date"])
except ValueError:
errors.append(f"Row {row_num}: invalid date format '{row['insight_date']}'")
if errors:
raise ValueError("\n".join(errors))
return row
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embeddings(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
input=texts,
model="text-embedding-3-small",
)
return [item.embedding for item in response.data]
@app.command()
def ingest(
file: Path = typer.Option(..., help="Path to the CSV file"),
source: str = typer.Option(..., help="Source system: CodeScan, ARM, Vault, or Guard"),
):
"""Ingest a CSV report into the insight consolidator."""
if source not in VALID_SOURCES:
console.print(f"[red]Error:[/red] Invalid source '{source}'. Must be one of: {VALID_SOURCES}")
raise typer.Exit(code=1)
if not file.exists():
console.print(f"[red]Error:[/red] File not found: {file}")
raise typer.Exit(code=1)
console.print(f"[bold blue]Ingesting[/bold blue] {file.name} from {source}...")
rows = []
with open(file, newline="", encoding="utf-8") as csvfile:
reader = csv.DictReader(csvfile)
for i, row in enumerate(reader, start=1):
validate_row(row, i)
rows.append(row)
console.print(f"[green]Validated {len(rows)} rows[/green]")
batch_size = 50
all_embeddings = []
texts = [row["raw_text"] for row in rows]
for i in track(range(0, len(texts), batch_size), description="Generating embeddings..."):
batch = texts[i : i + batch_size]
embeddings = generate_embeddings(batch)
all_embeddings.extend(embeddings)
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO reports (filename, source_system) VALUES (%s, %s) RETURNING id",
(file.name, source),
)
report_id = cur.fetchone()[0]
values = []
for row, embedding in zip(rows, all_embeddings):
tags = [t.strip() for t in row.get("tags", "").split(",") if t.strip()]
values.append((
report_id,
row["category"],
row["severity"],
row["insight_date"],
row["raw_text"],
tags,
np.array(embedding),
))
execute_values(
cur,
"""INSERT INTO insights
(report_id, category, severity, insight_date, raw_text, tags, embedding)
VALUES %s""",
values,
template="(%s, %s, %s, %s, %s, %s, %s)",
)
cur.close()
console.print(
f"[bold green]Success![/bold green] Ingested {len(rows)} insights from {file.name} (report_id={report_id})"
)
if __name__ == "__main__":
app()Your ingestion pipeline is working. Next up, you'll build a FastAPI REST API on top of this same database so you can query your insights programmatically.
Build the FastAPI REST API
Your ingestion pipeline fills the database with embeddings, but right now the only way to access that data is through raw SQL.
In this step, you'll build a FastAPI REST API that exposes your insights through structured endpoints for creating, filtering, aggregating, and retrieving records. FastAPI auto-generates interactive Swagger documentation so you can test every endpoint immediately from your browser.
In this step, get ready to:
- Define Pydantic request and response schemas with validation patterns.
- Build CRUD endpoints with category, severity, and date filters.
- Start the server and verify all endpoints via Swagger UI.
Define Pydantic schemas and create the FastAPI app
Pydantic models define the shape of data flowing in and out of your API. They validate incoming requests automatically and serialize outgoing responses to JSON.
- Create the api/ directory inside your insight-consolidator project folder by running this command:
mkdir api- Create a new file called schemas.py inside the api/ folder.
- Add the imports and the InsightCreate request model by pasting this code:
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, Field
class InsightCreate(BaseModel):
category: str = Field(..., pattern="^(code_quality|security|deployment|backup)$")
severity: str = Field(..., pattern="^(low|medium|high|critical)$")
insight_date: date
raw_text: str = Field(..., min_length=1)
tags: list[str] = Field(default_factory=list)
source_system: str = Field(..., pattern="^(CodeScan|ARM|Vault|Guard)$")What does this code do?
InsightCreate defines what the API expects when a client posts a new insight.
- Field(..., pattern="...") uses regex to restrict values to the exact categories, severities, and sources your database accepts.
- Field(..., min_length=1) ensures the raw text is never empty.
- default_factory=list makes tags optional with an empty list default.
- Add the InsightResponse and SummaryResponse models below InsightCreate in the same file:
class InsightResponse(BaseModel):
id: int
report_id: int
category: str
severity: str
insight_date: date
raw_text: str
tags: list[str]
created_at: datetime
class SummaryResponse(BaseModel):
by_category: dict[str, int]
by_severity: dict[str, int]
weekly_trend: list[dict[str, int | str]]
total: intWhat do these models do?
- InsightResponse defines the shape of individual insight records returned by the API. It includes database-generated fields like id and created_at.
- SummaryResponse structures the aggregation endpoint's output with counts by category, severity, a weekly trend, and a total count.
- Save api/schemas.py.
✔️ Awesome, I've got everything!
Great. Make sure you've saved api/schemas.py before moving on.
ⓧ I'd like to double check the full code
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, Field
class InsightCreate(BaseModel):
category: str = Field(..., pattern="^(code_quality|security|deployment|backup)$")
severity: str = Field(..., pattern="^(low|medium|high|critical)$")
insight_date: date
raw_text: str = Field(..., min_length=1)
tags: list[str] = Field(default_factory=list)
source_system: str = Field(..., pattern="^(CodeScan|ARM|Vault|Guard)$")
class InsightResponse(BaseModel):
id: int
report_id: int
category: str
severity: str
insight_date: date
raw_text: str
tags: list[str]
created_at: datetime
class SummaryResponse(BaseModel):
by_category: dict[str, int]
by_severity: dict[str, int]
weekly_trend: list[dict[str, int | str]]
total: intNow you need the FastAPI application itself and a helper function to generate embeddings for new insights created through the API.
- Create a new file called main.py inside the api/ folder.
- Add the imports, app initialization, and embedding helper by pasting this code:
from datetime import date
from typing import Optional
import numpy as np
from fastapi import FastAPI, HTTPException, Query
from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
from api.schemas import InsightCreate, InsightResponse, SummaryResponse
from db.connection import get_db_connection
# Initialize the FastAPI app with metadata
app = FastAPI(title="DevSecOps Insight Consolidator", version="1.0.0")
client = OpenAI()
# Reusable embedding helper with retry logic
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small",
)
return response.data[0].embeddingWhat does this code do?
- FastAPI(title=..., version=...) creates the app instance. The title and version appear in the auto-generated Swagger docs.
- generate_embedding() calls the same OpenAI model you used during ingestion, but for a single text string. The @retry decorator handles transient API failures with exponential backoff.
- The imports pull in your Pydantic schemas and the database connection context manager you built in earlier steps.
- Save api/main.py.
You won't see a change yet. The next chunk adds the endpoints that use this setup.
Implement the CRUD endpoints
Each endpoint handles a different access pattern: creating a single insight, listing with filters, fetching by ID, and summarizing the whole dataset.
- In api/main.py, add the POST endpoint below the generate_embedding function:
@app.post("/insights", response_model=InsightResponse, status_code=201)
def create_insight(insight: InsightCreate):
# Generate an embedding for the new insight's text
embedding = generate_embedding(insight.raw_text)
with get_db_connection() as conn:
cur = conn.cursor()
# Create or find a report for this single insert
cur.execute(
"INSERT INTO reports (filename, source_system) VALUES (%s, %s) RETURNING id",
("api_single_insert", insight.source_system),
)
report_id = cur.fetchone()[0]
cur.execute(
"""INSERT INTO insights
(report_id, category, severity, insight_date, raw_text, tags, embedding)
VALUES (%s, %s, %s, %s, %s, %s, %s)
RETURNING id, report_id, category, severity, insight_date, raw_text, tags, created_at""",
(
report_id,
insight.category,
insight.severity,
insight.insight_date,
insight.raw_text,
insight.tags,
np.array(embedding),
),
)
row = cur.fetchone()
cur.close()
return InsightResponse(
id=row[0],
report_id=row[1],
category=row[2],
severity=row[3],
insight_date=row[4],
raw_text=row[5],
tags=row[6],
created_at=row[7],
)What does this endpoint do?
POST /insights accepts a JSON body matching InsightCreate, generates a vector embedding for the text, inserts the record into PostgreSQL, and returns the full record with its database-assigned ID.
It creates a parent reports row with filename api_single_insert to maintain the foreign key relationship even for single API inserts.
- Add the GET list endpoint with filter support below create_insight:
@app.get("/insights", response_model=list[InsightResponse])
def list_insights(
category: Optional[str] = Query(None, pattern="^(code_quality|security|deployment|backup)$"),
severity: Optional[str] = Query(None, pattern="^(low|medium|high|critical)$"),
from_date: Optional[date] = Query(None, alias="from"),
to_date: Optional[date] = Query(None, alias="to"),
):
# Build WHERE clause dynamically from query params
conditions = []
params = []
if category:
conditions.append("category = %s")
params.append(category)
if severity:
conditions.append("severity = %s")
params.append(severity)
if from_date:
conditions.append("insight_date >= %s")
params.append(from_date)
if to_date:
conditions.append("insight_date <= %s")
params.append(to_date)
where_clause = " AND ".join(conditions) if conditions else "1=1"
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
f"""SELECT id, report_id, category, severity, insight_date, raw_text, tags, created_at
FROM insights WHERE {where_clause}
ORDER BY created_at DESC LIMIT 100""",
params,
)
rows = cur.fetchall()
cur.close()
return [
InsightResponse(
id=r[0], report_id=r[1], category=r[2], severity=r[3],
insight_date=r[4], raw_text=r[5], tags=r[6], created_at=r[7],
)
for r in rows
]What does this endpoint do?
- GET /insights returns up to 100 insights ordered by newest first.
- Query parameters (category, severity, from, to) are all optional. They compose into a dynamic WHERE clause using parameterized queries to prevent SQL injection.
- alias="from" lets the API use ?from=2024-11-01 in the URL while keeping a valid Python variable name (from is a reserved keyword).
- Add the summary and single-record endpoints below list_insights:
@app.get("/insights/summary", response_model=SummaryResponse)
def get_summary():
with get_db_connection() as conn:
cur = conn.cursor()
# Count insights grouped by category
cur.execute("SELECT category, COUNT(*) FROM insights GROUP BY category")
by_category = dict(cur.fetchall())
# Count insights grouped by severity
cur.execute("SELECT severity, COUNT(*) FROM insights GROUP BY severity")
by_severity = dict(cur.fetchall())
# Get weekly trend for the last 12 weeks
cur.execute(
"""SELECT DATE_TRUNC('week', insight_date)::date AS week, COUNT(*)
FROM insights GROUP BY week ORDER BY week DESC LIMIT 12"""
)
weekly_trend = [{"week": str(r[0]), "count": r[1]} for r in cur.fetchall()]
cur.execute("SELECT COUNT(*) FROM insights")
total = cur.fetchone()[0]
cur.close()
return SummaryResponse(
by_category=by_category,
by_severity=by_severity,
weekly_trend=weekly_trend,
total=total,
)
@app.get("/insights/{insight_id}", response_model=InsightResponse)
def get_insight(insight_id: int):
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"""SELECT id, report_id, category, severity, insight_date, raw_text, tags, created_at
FROM insights WHERE id = %s""",
(insight_id,),
)
row = cur.fetchone()
cur.close()
# Return 404 if no record matches the given ID
if not row:
raise HTTPException(status_code=404, detail="Insight not found")
return InsightResponse(
id=row[0], report_id=row[1], category=row[2], severity=row[3],
insight_date=row[4], raw_text=row[5], tags=row[6], created_at=row[7],
)What do these endpoints do?
- GET /insights/summary runs four aggregate queries and returns a high-level dashboard view: counts by category, counts by severity, a weekly trend, and the total.
- GET /insights/{insight_id} fetches a single record by primary key and raises a 404 if it does not exist.
Why is /insights/summary defined before /insights/{insight_id}?
FastAPI evaluates routes in order. If /insights/{insight_id} came first, a request to /insights/summary would match the path parameter as insight_id="summary" and fail with a validation error.
- Save api/main.py.
Seeing a red underline on the imports?
- Make sure your virtual environment is activated. Your editor needs to use the Python interpreter from .venv to resolve the installed packages.
- Check that api/schemas.py exists and is saved. The from api.schemas import ... line depends on it.
Help me fix import errors in my FastAPI project
✔️ Awesome, I've got everything!
Great. Double check you've saved api/main.py before starting the server.
ⓧ I'd like to double check the full code
from datetime import date
from typing import Optional
import numpy as np
from fastapi import FastAPI, HTTPException, Query
from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
from api.schemas import InsightCreate, InsightResponse, SummaryResponse
from db.connection import get_db_connection
app = FastAPI(title="DevSecOps Insight Consolidator", version="1.0.0")
client = OpenAI()
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small",
)
return response.data[0].embedding
@app.post("/insights", response_model=InsightResponse, status_code=201)
def create_insight(insight: InsightCreate):
embedding = generate_embedding(insight.raw_text)
with get_db_connection() as conn:
cur = conn.cursor()
# Create or find a report for this single insert
cur.execute(
"INSERT INTO reports (filename, source_system) VALUES (%s, %s) RETURNING id",
("api_single_insert", insight.source_system),
)
report_id = cur.fetchone()[0]
cur.execute(
"""INSERT INTO insights
(report_id, category, severity, insight_date, raw_text, tags, embedding)
VALUES (%s, %s, %s, %s, %s, %s, %s)
RETURNING id, report_id, category, severity, insight_date, raw_text, tags, created_at""",
(
report_id,
insight.category,
insight.severity,
insight.insight_date,
insight.raw_text,
insight.tags,
np.array(embedding),
),
)
row = cur.fetchone()
cur.close()
return InsightResponse(
id=row[0],
report_id=row[1],
category=row[2],
severity=row[3],
insight_date=row[4],
raw_text=row[5],
tags=row[6],
created_at=row[7],
)
@app.get("/insights", response_model=list[InsightResponse])
def list_insights(
category: Optional[str] = Query(None, pattern="^(code_quality|security|deployment|backup)$"),
severity: Optional[str] = Query(None, pattern="^(low|medium|high|critical)$"),
from_date: Optional[date] = Query(None, alias="from"),
to_date: Optional[date] = Query(None, alias="to"),
):
conditions = []
params = []
if category:
conditions.append("category = %s")
params.append(category)
if severity:
conditions.append("severity = %s")
params.append(severity)
if from_date:
conditions.append("insight_date >= %s")
params.append(from_date)
if to_date:
conditions.append("insight_date <= %s")
params.append(to_date)
where_clause = " AND ".join(conditions) if conditions else "1=1"
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
f"""SELECT id, report_id, category, severity, insight_date, raw_text, tags, created_at
FROM insights WHERE {where_clause}
ORDER BY created_at DESC LIMIT 100""",
params,
)
rows = cur.fetchall()
cur.close()
return [
InsightResponse(
id=r[0], report_id=r[1], category=r[2], severity=r[3],
insight_date=r[4], raw_text=r[5], tags=r[6], created_at=r[7],
)
for r in rows
]
@app.get("/insights/summary", response_model=SummaryResponse)
def get_summary():
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute("SELECT category, COUNT(*) FROM insights GROUP BY category")
by_category = dict(cur.fetchall())
cur.execute("SELECT severity, COUNT(*) FROM insights GROUP BY severity")
by_severity = dict(cur.fetchall())
cur.execute(
"""SELECT DATE_TRUNC('week', insight_date)::date AS week, COUNT(*)
FROM insights GROUP BY week ORDER BY week DESC LIMIT 12"""
)
weekly_trend = [{"week": str(r[0]), "count": r[1]} for r in cur.fetchall()]
cur.execute("SELECT COUNT(*) FROM insights")
total = cur.fetchone()[0]
cur.close()
return SummaryResponse(
by_category=by_category,
by_severity=by_severity,
weekly_trend=weekly_trend,
total=total,
)
@app.get("/insights/{insight_id}", response_model=InsightResponse)
def get_insight(insight_id: int):
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"""SELECT id, report_id, category, severity, insight_date, raw_text, tags, created_at
FROM insights WHERE id = %s""",
(insight_id,),
)
row = cur.fetchone()
cur.close()
if not row:
raise HTTPException(status_code=404, detail="Insight not found")
return InsightResponse(
id=row[0], report_id=row[1], category=row[2], severity=row[3],
insight_date=row[4], raw_text=row[5], tags=row[6], created_at=row[7],
)Test the API with Swagger UI
FastAPI generates an interactive documentation page at /docs where you can test every endpoint directly from your browser.
- Start the FastAPI development server by running this command from your project root:
uvicorn api.main:app --reloadWhat does this command do?
- uvicorn is the ASGI server that runs your FastAPI app.
- api.main:app tells uvicorn to look for the app object inside api/main.py.
- --reload watches for file changes and restarts the server automatically during development.
- You should see output indicating the server is running on http://127.0.0.1:8000.
- Open your browser and navigate to http://localhost:8000/docs.
You should see the Swagger UI page titled "DevSecOps Insight Consolidator" with all four endpoints listed.
- Click on **GET /insights** to expand it.
- Click **Try it out**.
- Leave all filter fields empty and click **Execute**.
You should see a JSON response containing the 8 insights you ingested from the CSV in the previous step.
- Try the **GET /insights/summary** endpoint the same way. You should see counts broken down by category and severity.
- Try **GET /insights** again with category set to security. You should see only the 3 security insights returned.
Server won't start?
- If you see ModuleNotFoundError, make sure your virtual environment is activated and you ran the command from the project root folder (the one containing the api/ and db/ directories).
- If port 8000 is already in use, stop any other server running on that port or add --port 8001 to the uvicorn command.
- If you see a database connection error, confirm PostgreSQL is running and your .env credentials are correct.
Help me debug my FastAPI server startup
Your API is live and serving data from the same PostgreSQL database your ingestion pipeline feeds. Next up, you'll build the RAG query engine that embeds natural-language questions and retrieves semantically similar insights.
Build the RAG Query Engine
Your API can serve insights over HTTP, but the real power of a vector database is answering questions you haven't pre-defined.
In this step, you'll build a RAG (Retrieval-Augmented Generation) query engine that takes a natural-language question, finds the most semantically relevant insights using cosine similarity, and synthesizes a grounded answer with GPT-4o.
In this step, get ready to:
- Embed a user question into a 1536-dimensional vector.
- Perform cosine similarity search against your pgvector database.
- Synthesize an AI-grounded answer with GPT-4o and display it in the terminal.
Embed the user question
The first stage of any RAG pipeline is turning the user's question into the same vector space as your stored data.
By embedding the question with the same model (text-embedding-3-small) that created your insight embeddings, you can compare them directly using cosine distance.
- Create a new file called rag/query.py in the rag folder of your project.
- Add the imports, client setup, and embedding function by pasting this code:
import sys
import numpy as np
from openai import OpenAI
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from tenacity import retry, stop_after_attempt, wait_exponential
from db.connection import get_db_connection
console = Console()
client = OpenAI()
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small",
)
return response.data[0].embeddingWhat does this code do?
The imports bring in Rich for terminal formatting, OpenAI for embeddings and chat, numpy for vector handling, and tenacity for retry logic.
The generate_embedding() function calls text-embedding-3-small to convert any text into a 1536-dimensional vector. The @retry decorator handles transient API failures with exponential backoff.
- Save rag/query.py.
- You won't see a change yet. The next function uses this embedding to search the database.
Perform cosine similarity search
With the question embedded, you need to find which stored insights are closest to it in vector space.
The pgvector extension provides the <=> operator for cosine distance. Lower distance means higher similarity.
- In rag/query.py, add the similarity search function below generate_embedding() by pasting this code:
def similarity_search(query_embedding: list[float], top_k: int = 5) -> list[dict]:
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"""SELECT id, category, severity, insight_date, raw_text, tags,
1 - (embedding <=> %s) AS similarity
FROM insights
ORDER BY embedding <=> %s
LIMIT %s""",
(np.array(query_embedding), np.array(query_embedding), top_k),
)
rows = cur.fetchall()
cur.close()
return [
{
"id": r[0],
"category": r[1],
"severity": r[2],
"insight_date": str(r[3]),
"raw_text": r[4],
"tags": r[5],
"similarity": round(float(r[6]), 4),
}
for r in rows
]What does this code do?
The SQL query uses embedding <=> %s to calculate cosine distance between the question vector and every stored insight. The ORDER BY sorts results by distance (lowest first) and LIMIT %s returns only the top 5.
The expression 1 - (embedding <=> %s) converts cosine distance into cosine similarity (a number between 0 and 1, where 1 means identical). The query embedding must be passed as a numpy array for pgvector to accept it.
Why numpy arrays instead of Python lists?
When using psycopg2 with pgvector, passing a plain Python list causes a type mismatch error. The pgvector adapter expects numpy arrays so it can properly serialize the vector for PostgreSQL.
- Save rag/query.py.
- You won't see a change yet. The next chunk builds the synthesis and display logic that ties everything together.
Synthesize with GPT-4o and display
You now have the retrieval half of RAG. The generation half sends the retrieved context to GPT-4o with a system prompt that keeps the answer grounded in your actual data.
- In rag/query.py, add the synthesis function, the main query orchestrator, and the entry point below similarity_search() by pasting this code:
def synthesize_answer(question: str, context: list[dict]) -> str:
context_text = "\n\n".join(
f"[{c['category']}|{c['severity']}|{c['insight_date']}] {c['raw_text']}"
for c in context
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "developer",
"content": (
"You are a DevSecOps analytics assistant. Answer questions based on "
"the provided context from ingested reports. Be specific, cite dates "
"and severities when relevant. If the context doesn't contain enough "
"information, say so."
),
},
{
"role": "user",
"content": f"Question: {question}\n\nContext:\n{context_text}",
},
],
temperature=0.3,
)
return response.choices[0].message.content
def query(question: str):
"""Run a RAG query against the insight database."""
console.print(Panel(question, title="Question", border_style="blue"))
with console.status("[bold green]Embedding question..."):
query_embedding = generate_embedding(question)
with console.status("[bold green]Searching similar insights..."):
results = similarity_search(query_embedding)
if not results:
console.print("[yellow]No relevant insights found in the database.[/yellow]")
return
console.print(f"\n[dim]Found {len(results)} relevant insights (top similarity: {results[0]['similarity']})[/dim]\n")
with console.status("[bold green]Synthesizing answer with GPT-4o..."):
answer = synthesize_answer(question, results)
console.print(Panel(Markdown(answer), title="Answer", border_style="green"))
if __name__ == "__main__":
if len(sys.argv) < 2:
console.print("[red]Usage:[/red] python -m rag.query \"your question here\"")
sys.exit(1)
query(sys.argv[1])What does this code do?
- synthesize_answer() formats the retrieved insights into a context string and sends it to GPT-4o with a developer prompt that instructs the model to stay grounded in the provided data.
- query() orchestrates the full RAG pipeline: embed the question, search for similar insights, and synthesize the final answer. Rich status spinners give the user feedback during each stage.
- The __main__ block lets you run the file as a module with python -m rag.query followed by your question in quotes.
Why temperature 0.3?
A low temperature makes GPT-4o produce more deterministic, factual answers. For an analytics assistant that should report data accurately, you want minimal creative hallucination. Higher temperatures (0.7+) are better for creative writing tasks.
- Save rag/query.py.
- Test the RAG engine by running this command from your project root:
python -m rag.query "What are the most critical security issues?"You should see a blue "Question" panel, a progress note about found insights with a similarity score, and then a green "Answer" panel containing GPT-4o's synthesized response citing specific findings from your ingested data.
Seeing a ModuleNotFoundError?
- Make sure you're running the command from your project root directory (the folder containing the rag/ folder), not from inside rag/ itself.
- Confirm your virtual environment is activated. The prompt should show (.venv) at the start.
- Check that your .env file has a valid OPENAI_API_KEY set. The OpenAI client reads it automatically via python-dotenv.
Why isn't my RAG query finding any results or giving me an import error?
✔️ Awesome, I've got everything!
Great. Double check you've saved rag/query.py and move on.
ⓧ I'd like to double check the full code
Here is the complete rag/query.py file for comparison:
import sys
import numpy as np
from openai import OpenAI
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from tenacity import retry, stop_after_attempt, wait_exponential
from db.connection import get_db_connection
console = Console()
client = OpenAI()
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small",
)
return response.data[0].embedding
def similarity_search(query_embedding: list[float], top_k: int = 5) -> list[dict]:
with get_db_connection() as conn:
cur = conn.cursor()
cur.execute(
"""SELECT id, category, severity, insight_date, raw_text, tags,
1 - (embedding <=> %s) AS similarity
FROM insights
ORDER BY embedding <=> %s
LIMIT %s""",
(np.array(query_embedding), np.array(query_embedding), top_k),
)
rows = cur.fetchall()
cur.close()
return [
{
"id": r[0],
"category": r[1],
"severity": r[2],
"insight_date": str(r[3]),
"raw_text": r[4],
"tags": r[5],
"similarity": round(float(r[6]), 4),
}
for r in rows
]
def synthesize_answer(question: str, context: list[dict]) -> str:
context_text = "\n\n".join(
f"[{c['category']}|{c['severity']}|{c['insight_date']}] {c['raw_text']}"
for c in context
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "developer",
"content": (
"You are a DevSecOps analytics assistant. Answer questions based on "
"the provided context from ingested reports. Be specific, cite dates "
"and severities when relevant. If the context doesn't contain enough "
"information, say so."
),
},
{
"role": "user",
"content": f"Question: {question}\n\nContext:\n{context_text}",
},
],
temperature=0.3,
)
return response.choices[0].message.content
def query(question: str):
"""Run a RAG query against the insight database."""
console.print(Panel(question, title="Question", border_style="blue"))
with console.status("[bold green]Embedding question..."):
query_embedding = generate_embedding(question)
with console.status("[bold green]Searching similar insights..."):
results = similarity_search(query_embedding)
if not results:
console.print("[yellow]No relevant insights found in the database.[/yellow]")
return
console.print(f"\n[dim]Found {len(results)} relevant insights (top similarity: {results[0]['similarity']})[/dim]\n")
with console.status("[bold green]Synthesizing answer with GPT-4o..."):
answer = synthesize_answer(question, results)
console.print(Panel(Markdown(answer), title="Answer", border_style="green"))
if __name__ == "__main__":
if len(sys.argv) < 2:
console.print("[red]Usage:[/red] python -m rag.query \"your question here\"")
sys.exit(1)
query(sys.argv[1])Your RAG engine is live. You can now ask natural-language questions about your DevSecOps data and get AI-synthesized answers grounded in real insights. Next up, you'll write tests to make sure the ingestion and API layers hold up under edge cases.
Write and Run the Tests
Your RAG query engine is working. You can ask questions in plain English and get AI-synthesized answers grounded in your actual DevSecOps data.
Before shipping any codebase, you need confidence that it works as expected. In this step, you will write automated tests for the ingestion validator and the API endpoints using pytest and mock objects. Mocking lets you test your logic without making real database connections or OpenAI API calls.
In this step, get ready to:
- Write ingestion tests that validate CSV row parsing and error handling.
- Write API tests using FastAPI's TestClient and mocked database connections.
- Run the full test suite with pytest and confirm all tests pass.
Write ingestion tests
The ingestion pipeline has a validate_row function that checks each CSV row before it ever touches the database. Testing this function in isolation ensures bad data never sneaks through.
- Create a tests/ directory inside your insight-consolidator project folder by running this command:
mkdir tests- Create a new file called test_ingest.py inside the tests/ directory.
- Copy the following code into tests/test_ingest.py:
from unittest.mock import MagicMock, patch
from datetime import date
import pytest
# We need to set up the path for imports
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from ingestion.ingest import validate_row, VALID_CATEGORIES, VALID_SEVERITIES
class TestValidateRow:
def test_valid_row(self):
row = {
"category": "code_quality",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": "Test insight text",
"tags": "test,example",
}
result = validate_row(row, 1)
assert result == row
def test_invalid_category(self):
row = {
"category": "invalid",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid category"):
validate_row(row, 1)
def test_invalid_severity(self):
row = {
"category": "security",
"severity": "extreme",
"insight_date": "2024-11-15",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid severity"):
validate_row(row, 1)
def test_empty_raw_text(self):
row = {
"category": "security",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": " ",
}
with pytest.raises(ValueError, match="raw_text is empty"):
validate_row(row, 1)
def test_invalid_date(self):
row = {
"category": "security",
"severity": "high",
"insight_date": "not-a-date",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid date format"):
validate_row(row, 1)What does this test file do?
The sys.path.insert line at the top adds the project root to Python's import path. This lets pytest find your ingestion package without needing a package install.
The TestValidateRow class groups five tests that exercise every validation branch: valid data passes through, and invalid categories, severities, empty text, and bad dates all raise ValueError with a descriptive message.
- Save tests/test_ingest.py.
Seeing an import error?
Make sure your terminal is in the project root folder (the one containing db/, ingestion/, api/). The sys.path.insert line resolves relative to where the test file lives.
Also confirm your virtual environment is activated. help me fix my pytest import error
Write API tests
Testing API endpoints requires calling your routes without a live database. FastAPI's TestClient lets you simulate HTTP requests. Combined with unittest.mock, you can replace the real database connection with a controlled fake.
- Create a new file called test_api.py inside the tests/ directory.
- Copy the following code into tests/test_api.py:
from unittest.mock import patch, MagicMock
import pytest
from fastapi.testclient import TestClient
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from api.main import app
client = TestClient(app)
class TestListInsights:
@patch("api.main.get_db_connection")
def test_list_insights_no_filters(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.return_value = []
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights")
assert response.status_code == 200
assert response.json() == []
@patch("api.main.get_db_connection")
def test_list_insights_with_category_filter(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.return_value = []
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights?category=security")
assert response.status_code == 200
class TestGetInsight:
@patch("api.main.get_db_connection")
def test_not_found(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchone.return_value = None
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights/999")
assert response.status_code == 404
class TestSummary:
@patch("api.main.get_db_connection")
def test_summary_endpoint(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.side_effect = [
[("security", 5), ("code_quality", 3)],
[("high", 4), ("medium", 4)],
[("2024-11-11", 8)],
]
mock_cur.fetchone.return_value = (8,)
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights/summary")
assert response.status_code == 200
data = response.json()
assert "by_category" in data
assert "by_severity" in data
assert "total" in dataWhat does this test file do?
Each test class targets a different API endpoint. The @patch("api.main.get_db_connection") decorator replaces the real database connection with a MagicMock. This means your tests never touch PostgreSQL.
The mock cursor returns pre-defined data (empty lists, None, or tuples) so you can test how your endpoints behave in each scenario. The summary test uses side_effect to return different results for each successive fetchall call.
- Save tests/test_api.py.
Getting a module not found error for httpx?
FastAPI's TestClient depends on httpx under the hood. Confirm it is installed by checking your requirements.txt includes httpx==0.28.1. If you missed it, run pip install httpx==0.28.1 to add it.
help me fix my TestClient import error
Run the test suite
Both test files are ready. Time to execute the full suite and confirm everything passes.
- Run the test suite with verbose output by running this command:
pytest -vYou should see each test listed by name with a PASSED status next to it. All 9 tests should pass.
What should I see?
The -v flag tells pytest to show individual test names instead of just dots. You should see output like tests/test_ingest.py::TestValidateRow::test_valid_row PASSED for each test, followed by a summary line showing all tests collected and passed.
Tests failing or not collected?
Make sure you are running pytest -v from the project root directory (where requirements.txt lives). Pytest discovers tests by searching for files named test_*.py in subdirectories.
If a specific test fails, read the assertion error carefully. The error message tells you exactly which line did not match.
help me debug my failing pytest tests
✔️ Awesome, I've got everything!
Great. Double check you have saved both test files.
ⓧ I'd like to double check the full code
Here is the complete tests/test_ingest.py:
from unittest.mock import MagicMock, patch
from datetime import date
import pytest
# We need to set up the path for imports
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from ingestion.ingest import validate_row, VALID_CATEGORIES, VALID_SEVERITIES
class TestValidateRow:
def test_valid_row(self):
row = {
"category": "code_quality",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": "Test insight text",
"tags": "test,example",
}
result = validate_row(row, 1)
assert result == row
def test_invalid_category(self):
row = {
"category": "invalid",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid category"):
validate_row(row, 1)
def test_invalid_severity(self):
row = {
"category": "security",
"severity": "extreme",
"insight_date": "2024-11-15",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid severity"):
validate_row(row, 1)
def test_empty_raw_text(self):
row = {
"category": "security",
"severity": "high",
"insight_date": "2024-11-15",
"raw_text": " ",
}
with pytest.raises(ValueError, match="raw_text is empty"):
validate_row(row, 1)
def test_invalid_date(self):
row = {
"category": "security",
"severity": "high",
"insight_date": "not-a-date",
"raw_text": "Test text",
}
with pytest.raises(ValueError, match="invalid date format"):
validate_row(row, 1)And here is the complete tests/test_api.py:
from unittest.mock import patch, MagicMock
import pytest
from fastapi.testclient import TestClient
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from api.main import app
client = TestClient(app)
class TestListInsights:
@patch("api.main.get_db_connection")
def test_list_insights_no_filters(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.return_value = []
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights")
assert response.status_code == 200
assert response.json() == []
@patch("api.main.get_db_connection")
def test_list_insights_with_category_filter(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.return_value = []
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights?category=security")
assert response.status_code == 200
class TestGetInsight:
@patch("api.main.get_db_connection")
def test_not_found(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchone.return_value = None
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights/999")
assert response.status_code == 404
class TestSummary:
@patch("api.main.get_db_connection")
def test_summary_endpoint(self, mock_conn):
mock_cur = MagicMock()
mock_cur.fetchall.side_effect = [
[("security", 5), ("code_quality", 3)],
[("high", 4), ("medium", 4)],
[("2024-11-11", 8)],
]
mock_cur.fetchone.return_value = (8,)
mock_conn.return_value.__enter__.return_value.cursor.return_value = mock_cur
response = client.get("/insights/summary")
assert response.status_code == 200
data = response.json()
assert "by_category" in data
assert "by_severity" in data
assert "total" in dataYour project now has automated tests covering both the data validation layer and the REST API. Next up, you will tackle the Secret Mission and build an agentic loop that classifies questions and autonomously picks the right tool.
Secret mission
Build the Agentic Loop
Your RAG query engine always uses the same strategy: embed the question, find similar vectors, synthesize. But what if some questions are better answered by a structured filter or a summary aggregation? Build an autonomous agent that classifies each question and picks the right tool automatically.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs locally with a PostgreSQL database on your machine. The only pay-per-use resource is the OpenAI API, which only charges when you make requests.
Resources you used:
- The insight_consolidator PostgreSQL database (contains the reports and insights tables with vector data).
- Your project folder on the Desktop (all Python source code, virtual environment, sample data, and tests).
- Your OpenAI API key (stored in your .env file and used for embedding and chat completion calls).
Keep everything running
No action needed. Choose this if you are still actively building on this project or plan to extend it with the Secret Mission.
- Your PostgreSQL database will continue running as a background service. It consumes minimal resources when idle.
- Your project folder and virtual environment stay on your Desktop ready to use.
- Your OpenAI API key remains active but only costs money when you run commands that call the API.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your files and database stay intact for when you return.
- Stop the PostgreSQL service to free memory. On macOS, run:
brew services stop postgresql@16- On Windows, open the Services app (search for "Services" in the Start menu), find the PostgreSQL service, right-click it, and select Stop.
- If you had the FastAPI server running with uvicorn, press Ctrl+C in that terminal window to stop it.
- Your project folder, database data, and .env file all remain on disk. When you come back, start the PostgreSQL service again and reactivate your virtual environment.
Delete - I don't want to use this again
Remove all project resources and start fresh. This cannot be undone.
- Drop the insight_consolidator database by running this command:
dropdb insight_consolidatorWhat does this command do?
The dropdb utility is a PostgreSQL tool that permanently removes a database and all its contents. This deletes the reports table, the insights table, the pgvector index, and all ingested data.
- Delete your project folder. On macOS or Linux, run:
cd ~/Desktop
rm -rf insight-consolidator- On Windows, run:
cd ~\Desktop
rmdir /s /q insight-consolidator- Delete your OpenAI API key. Go to the API keys page in your OpenAI account and click the trashcan icon next to the key you created for this project.
Deleting your API key is permanent
Once you delete an API key, any application using that key will stop working. Only delete it if you are sure you will not use it for other projects.
Nice Work!
Nice Work!
Nice work. You just built a production-grade AI analytics platform from scratch. Your DevSecOps Insight Consolidator ingests real reports, stores them as vector embeddings, serves them through a REST API, and answers natural-language questions using RAG.
You've learned how to:
- Design a PostgreSQL + pgvector schema for storing 1536-dimensional vector embeddings and query them with cosine similarity search.
- Build a CLI ingestion pipeline that validates CSVs, generates OpenAI embeddings with retry logic, and bulk-inserts records into your database.
- Implement a RAG query engine that embeds user questions, performs semantic similarity search, and synthesizes answers with GPT-4o grounded in your actual data.
- Secret Mission: Built an agentic loop that autonomously classifies questions and selects the right tool (semantic search, structured filter, or summarization) before synthesizing a final answer.
Ready to quiz yourself?