LangChain CSV SQLite Analytics: Safer AI Foundation

A developer from Gate of AI published a tutorial on building a deterministic CSV-to-SQLite analytics foundation with guarded, read-only SQL, designed as a safe boundary for LangChain-style agents. The project uses only Python's standard library and emphasizes keeping database access, query limits, and authorization under application control rather than the model.

🚀 Technical Briefing:This tutorial is part of our deep-dive series on Agentic Workflows at Gate of AI . For the full technical breakdown, interactive code sandbox, and the native Arabic translation, visit the original article here . Build a deterministic CSV-to-SQLite analytics foundation with guarded, read-only SQL. It is designed as a safe boundary that a LangChain-style agent can call after its framework and model integration have been verified against current official documentation. 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. 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. 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. python 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. 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 . python from future import annotations import csv import re import sqlite3 from pathlib import Path from typing import Any TABLE 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 candidate def 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 = csv.DictReader file if not reader.fieldnames: raise ValueError "CSV must have a header row." source headers = list reader.fieldnames used: set str = set columns = normalize identifier header, used for header in source headers rows = list reader if not rows: raise ValueError "CSV must contain at least one data row." sqlite path.parent.mkdir parents=True, exist ok=True with sqlite3.connect sqlite path as connection: table = quote identifier TABLE NAME connection.execute f"DROP TABLE IF EXISTS {table}" definitions = ", ".join f"{quote identifier column } TEXT" for column in columns connection.execute f"CREATE TABLE {table} {definitions} " insert columns = ", ".join quote identifier column for column in columns placeholders = ", ".join "?" for in columns statement = f"INSERT INTO {table} {insert columns} VALUES {placeholders} " values = tuple row.get header, "" .strip for header in source headers for row in rows connection.executemany statement, values return columns def get schema sqlite path: Path - dict str, Any : with sqlite3.connect sqlite path as connection: connection.row factory = sqlite3.Row columns = connection.execute "PRAGMA table info orders " .fetchall count = connection.execute "SELECT COUNT AS total FROM orders" .fetchone "total" return { "table name": TABLE NAME, "row count": count, "columns": {"name": row "name" , "type": row "type" } for row in columns , } The identifier check is important because SQL parameters protect values, not SQL identifiers such as column names. Headers are normalized before being used to build SQL. Values, meanwhile, are sent through parameterized inserts rather than string interpolation. The following program is the application boundary an agent should call. It rejects comments, semicolons, recursive queries, non-read-only starting keywords, and listed administrative or write operations. It also opens the database through a SQLite read-only URI and fetches no more than 100 visible rows. The URI is a second protective layer: even if validation is changed incorrectly, the query connection is not intended for writes. python from future import annotations import json import re import sqlite3 from pathlib import Path from urllib.parse import quote from database import get schema, load csv into sqlite MAX ROWS = 100 FORBIDDEN = re.compile r"\b INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|REPLACE|VACUUM|ATTACH|DETACH|" r"PRAGMA|REINDEX|ANALYZE|BEGIN|COMMIT|ROLLBACK|SAVEPOINT|RELEASE \b", re.IGNORECASE, def validate read only sql sql: str - str: candidate = sql.strip if not candidate: raise ValueError "Query cannot be empty." if len candidate 4000: raise ValueError "Query exceeds 4000 characters." if ";" in candidate or "--" in candidate or "/ " in candidate or " /" in candidate: raise ValueError "Comments and multiple statements are not allowed." normalized = re.sub r"\s+", " ", candidate .upper if not normalized.startswith "SELECT " or normalized.startswith "WITH " : raise ValueError "Only SELECT or WITH queries are allowed." if "WITH RECURSIVE" in normalized or FORBIDDEN.search candidate : raise ValueError "Query contains a disallowed SQL operation." return candidate def run query sqlite path: Path, sql: str - dict str, object : safe sql = validate read only sql sql uri = f"file:{quote str sqlite path.resolve }?mode=ro" with sqlite3.connect uri, uri=True as connection: connection.row factory = sqlite3.Row cursor = connection.execute safe sql rows = cursor.fetchmany MAX ROWS + 1 return { "row count returned": min len rows , MAX ROWS , "truncated": len rows MAX ROWS, "rows": dict row for row in rows :MAX ROWS , } def main - None: csv path = Path "data/orders.csv" sqlite path = Path "data/orders.sqlite3" load csv into sqlite csv path, sqlite path print json.dumps get schema sqlite path , indent=2 print "Enter read-only SQL, /schema, or /quit." while True: try: request = input "SQL " .strip except EOFError, KeyboardInterrupt : print "\nGoodbye." return if request.lower in {"/quit", "/exit"}: print "Goodbye." return if request.lower == "/schema": print json.dumps get schema sqlite path , indent=2 continue try: print json.dumps run query sqlite path, request , indent=2 except ValueError, sqlite3.Error as error: print f"Rejected or invalid query: {error}" if name == " main ": main Save this file as app.py and run python app.py . Then enter the following query: SELECT product category, ROUND SUM CAST order total AS REAL , 2 AS completed revenue FROM orders WHERE status = 'completed' GROUP BY product category ORDER BY completed revenue DESC LIMIT 1 The explicit cast prevents text ordering and aggregation from being confused with numeric analysis. The result is also scoped to completed records, which is one possible definition of realized revenue in this sample. A real organization must document its own metric definitions; a query cannot resolve ambiguity about booked, invoiced, collected, gross, net, refunded, or recognized revenue. Tests should exercise the ingestion and query guardrails without a model call. This makes failures fast to reproduce and keeps safety behavior independent of prompt wording or model output. python from pathlib import Path import pytest from app import run query, validate read only sql from database import get schema, load csv into sqlite from sample data import create sample csv def test load and schema tmp path: Path - None: csv path = tmp path / "orders.csv" sqlite path = tmp path / "orders.sqlite3" create sample csv csv path load csv into sqlite csv path, sqlite path schema = get schema sqlite path assert schema "table name" == "orders" assert schema "row count" == 12 assert any column "name" == "order total" for column in schema "columns" def test aggregate query tmp path: Path - None: csv path = tmp path / "orders.csv" sqlite path = tmp path / "orders.sqlite3" create sample csv csv path load csv into sqlite csv path, sqlite path result = run query sqlite path, "SELECT region, COUNT AS n FROM orders GROUP BY region" assert result "truncated" is False assert result "row count returned" == 4 @pytest.mark.parametrize "sql", "DELETE FROM orders", "DROP TABLE orders", "SELECT FROM orders; DELETE FROM orders", "SELECT FROM orders -- comment", "WITH RECURSIVE n x AS SELECT 1 SELECT x FROM n", def test disallowed sql sql: str - None: with pytest.raises ValueError : validate read only sql sql Run pytest -q . If a disallowed statement begins to pass, stop and review the change before adding further features. A permissive boundary is not a presentation issue; it changes what the application can do with a model-generated request. When you have current official documentation for the exact LangChain release you plan to deploy, expose two narrow functions as tools: one that returns get schema and one that accepts SQL and calls run query . The model-facing tool description should state that orders is the approved table, source columns are text, numeric calculations require explicit casts, and list-style requests should use a limit. Do not give the agent a raw SQLite connection, filesystem access, arbitrary Python execution, or a function that can modify the database. Do not place API keys in prompts, tool descriptions, CSV values, or logs. Maintain a bounded conversation history and require the agent to use the query tool for factual numerical answers rather than inventing figures. Before using organizational data, review each column and remove data that is unnecessary for the analytics task. For any GCC or Middle East deployment, confirm the applicable organizational requirements for access, retention, residency, and handling of personal or confidential data with the relevant legal, security, and data-governance teams. A local SQLite demonstration does not establish production compliance. The next technical step is not to add more autonomy; it is to add control. Create an approved data dictionary, document metric definitions, allowlist tables and columns, and record sanitized query metadata such as request ID, execution time, row count, truncation status, and error category. Do not record secrets or unrestricted raw sensitive values. For a production analytics store, use a database identity that has access only to approved reporting views and apply authorization before a query reaches the database. Keep result-size limits, query budgets, and a regression suite containing valid aggregations, missing-column requests, empty results, ambiguous terms, and attempted prompt-injection text in dataset fields. This foundation is intentionally modest: deterministic software prepares and protects data, while an agent framework—once independently verified and version-pinned—can supply the conversational layer. That division keeps the important access and safety decisions in code you can inspect and test.