LangChain CSV SQLite Analytics: Safer AI Foundation

What this tutorial does—and does not verify

The supplied research context identifies the general pattern of using LangChain agents with external tools and the broader use case of asking questions about CSV data. It does not provide trusted, current documentation for a particular LangChain release, OpenAI model, package API, tracing product, or web framework. For that reason, this tutorial deliberately does not present unverified agent-framework code as production-ready.

Instead, you will build the deterministic portion that should remain under application control regardless of which model or orchestration framework you select later. The project creates a CSV file, imports it into a local SQLite database, describes the approved schema, validates one read-only SQL statement at a time, opens the database in read-only mode for analytics queries, caps returned rows, and tests the important non-model behavior.

This separation matters. A language model may help choose a tool and formulate a question, but it should not receive a writable database connection, a shell function, unrestricted Python execution, or secrets. Your application should retain control of CSV ingestion, database access, query limits, authorization, logging policy, and the definition of approved business metrics.

Prerequisites and project layout

This example uses Python 3.10 or later and only the Python standard library for the runnable application. SQLite is accessed through Python’s built-in sqlite3 module. Install pytest separately if you want to run the tests.

mkdir csv-sqlite-analytics
cd csv-sqlite-analytics

python -m venv .venv

# macOS and Linux
source .venv/bin/activate

# Windows PowerShell
# .\.venv\Scripts\Activate.ps1

python -m pip install --upgrade pip
python -m pip install pytest

mkdir data tests

Create four files: sample_data.py, database.py, app.py, and tests/test_database.py. The command-line program accepts guarded SQL in this version. A future agent adapter can translate natural-language questions into SQL, but it must call the same validation and execution boundary shown here.

Step 1: Create a repeatable CSV file

A deterministic sample makes the behavior easy to inspect and test. The sample has order identifiers, regions, statuses, categories, quantities, prices, and totals. It is demonstration data only; replace it with a reviewed export only after removing fields that your users and application should not access.

from __future__ import annotations

import csv
from pathlib import Path


ORDERS = [
    ["ORD-1001", "2026-01-05", "North", "Enterprise", "Analytics", "completed", 3, 1200.00],
    ["ORD-1002", "2026-01-06", "South", "SMB", "Support", "completed", 8, 150.00],
    ["ORD-1003", "2026-01-07", "West", "Enterprise", "Security", "completed", 2, 2500.00],
    ["ORD-1004", "2026-01-08", "East", "Mid-Market", "Analytics", "pending", 4, 900.00],
    ["ORD-1005", "2026-01-09", "North", "SMB", "Support", "completed", 12, 125.00],
    ["ORD-1006", "2026-01-11", "West", "Enterprise", "Analytics", "completed", 5, 1450.00],
    ["ORD-1007", "2026-01-13", "South", "Mid-Market", "Security", "cancelled", 1, 2200.00],
    ["ORD-1008", "2026-01-15", "East", "SMB", "Support", "completed", 6, 175.00],
    ["ORD-1009", "2026-01-18", "North", "Mid-Market", "Analytics", "completed", 7, 980.00],
    ["ORD-1010", "2026-01-21", "West", "SMB", "Security", "completed", 2, 2400.00],
    ["ORD-1011", "2026-01-25", "East", "Enterprise", "Analytics", "completed", 4, 1600.00],
    ["ORD-1012", "2026-01-28", "South", "Mid-Market", "Support", "pending", 10, 140.00],
]


def create_sample_csv(destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    with destination.open("w", newline="", encoding="utf-8") as file:
        writer = csv.writer(file)
        writer.writerow([
            "order_id", "order_date", "region", "customer_segment",
            "product_category", "status", "quantity", "unit_price", "order_total",
        ])
        for order_id, order_date, region, segment, category, status, quantity, unit_price in ORDERS:
            writer.writerow([
                order_id, order_date, region, segment, category, status,
                quantity, f"{unit_price:.2f}", f"{quantity * unit_price:.2f}",
            ])


if __name__ == "__main__":
    create_sample_csv(Path("data/orders.csv"))
    print("Created data/orders.csv with 12 records.")

Run python sample_data.py. The standard CSV writer is preferable to hand-built comma-separated strings because it correctly escapes values containing commas, quotes, or line breaks.

Step 2: Import CSV data into SQLite

The importer below normalizes CSV headers into safe database identifiers, creates an orders table, and uses parameterized inserts for values. Imported fields are stored as text. This conservative representation avoids unwanted coercion of values such as identifiers with leading zeroes. Numeric analysis explicitly casts appropriate fields to REAL.

from __future__ import annotationsimport csv
import re
import sqlite3
from pathlib import Path
from typing import AnyTABLE_NAME = "orders"
IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")def normalize_identifier(value: str, used: set[str]) -> str:
name = re.sub(r"[^A-Za-z0-9_]", "_", value.strip().lower())
name = re.sub(r"_+", "_", name).strip("_") or "column"
if name[0].isdigit():
name = f"column_{name}"
candidate = name
suffix = 2
while candidate in used:
candidate = f"{name}_{suffix}"
suffix += 1
used.add(candidate)
return candidatedef quote_identifier(identifier: str) -> str:
if not IDENTIFIER.fullmatch(identifier):
raise ValueError(f"Unsafe identifier: {identifier!r}")
return f'"{identifier}"'def load_csv_into_sqlite(csv_path: Path, sqlite_path: Path) -> list[str]:
if not csv_path.exists():
raise FileNotFoundError(f"CSV file does not exist: {csv_path}")with csv_path.open("r", newline="", encoding="utf-8-sig") as file:
reader...
Continue Reading
Log in for free to read the rest of this article and access exclusive AI tools.
Log in / Register