--- title: "Ingest CSV into pgvector with Row Context — EnConvert API" description: "Ingest CSV into pgvector with row context preserved on every chunk. Stop returning results that match in vector space but miss the point. Free tier." canonical: "https://www.enconvert.com/solutions/ingest-csv-into-pgvector-with-row-context-enconvert-api" locale: "en" --- # How to Ingest a CSV into pgvector Without Losing Row Context The default approach to CSV ingestion is a loop: open the file, iterate rows, join columns into a string, embed it, insert into pgvector. What gets lost is everything that made the row meaningful — which column the value came from, where in the file it sits, how it relates to adjacent rows. Queries return vectors that are close in embedding space to the question but carry no structural context to confirm they actually answer it. A row containing the words "30-day return" is not the same as the row that defines your refund policy, and flat string insertion cannot tell the difference. EnConvert's ingest endpoint processes the CSV and returns a JSONL file where each chunk preserves row context, column labeling, and source metadata — so when that chunk lands in pgvector, retrieval is both semantically accurate and structurally grounded. ## Example ```python import json import os import time import psycopg2 import requests from openai import OpenAI from pgvector import Vector from pgvector.psycopg2 import register_vector from psycopg2.extras import execute_values ENCONVERT_API_KEY = os.environ["ENCONVERT_API_KEY"] OPENAI_API_KEY = os.environ["OPENAI_API_KEY"] DATABASE_URL = os.environ["DATABASE_URL"] # e.g. postgresql://user:password@localhost:5432/yourdb CSV_PATH = "/path/to/your/data.csv" # Step 1: Upload the CSV to EnConvert (multipart/form-data, field name "files") with open(CSV_PATH, "rb") as csv_file: submit_response = requests.post( "https://api.enconvert.com/v2/ingest/files", headers={"X-API-Key": ENCONVERT_API_KEY}, files={"files": csv_file}, timeout=60, ) submit_response.raise_for_status() job_id = submit_response.json()["job_id"] # Step 2: Poll until the job completes (give up after 5 minutes) deadline = time.monotonic() + 300 while True: job_response = requests.get( f"https://api.enconvert.com/v2/ingest/{job_id}", headers={"X-API-Key": ENCONVERT_API_KEY}, timeout=60, ) job_response.raise_for_status() job = job_response.json() if job["status"] == "completed": break if job["status"] in ("failed", "canceled"): raise RuntimeError(f"Ingest {job['status']}: {job.get('error_message')}") if time.monotonic() > deadline: raise TimeoutError(f"Ingest job {job_id} is still {job['status']}") time.sleep(2) # Step 3: Download the JSONL, one chunk per line. output_url is a signed URL # (valid 15 minutes): plain GET, no API key. output_response = requests.get(job["output_url"], timeout=60) output_response.raise_for_status() lines = output_response.content.decode("utf-8").split("\n") chunks = [json.loads(line) for line in lines if line.strip()] # Step 4: A CSV comes back as a single chunk holding the whole table: a header # row of column labels, a separator row, then one line per CSV row. Pair every # row with the header so each embedded row keeps its column labels. EnConvert # converts at most 5,000 rows per CSV; its truncation note is not a table row. records = [] for chunk in chunks: table_lines = [line for line in chunk["content"].split("\n") if line.startswith("|")] header = table_lines[:2] for row_index, row in enumerate(table_lines[2:]): records.append( { "content": "\n".join(header + [row]), "source_file": chunk["metadata"]["source_url"], # the uploaded filename "row_index": row_index, } ) # Step 5: Create the pgvector table conn = psycopg2.connect(DATABASE_URL) cur = conn.cursor() cur.execute("CREATE EXTENSION IF NOT EXISTS vector") register_vector(conn) cur.execute( """ CREATE TABLE IF NOT EXISTS csv_rows ( id BIGSERIAL PRIMARY KEY, content TEXT NOT NULL, source_file TEXT, row_index INTEGER, embedding vector(1536) ) """ ) # Step 6: Embed the rows in batches and insert them. OpenAI allows up to 2,048 # inputs and 300,000 tokens per embeddings request (8,192 tokens per input), so # 100 rows is safe for typical CSVs; lower BATCH_SIZE for very wide ones. BATCH_SIZE = 100 openai_client = OpenAI(api_key=OPENAI_API_KEY) for start in range(0, len(records), BATCH_SIZE): batch = records[start : start + BATCH_SIZE] embeddings = openai_client.embeddings.create( model="text-embedding-3-small", input=[record["content"] for record in batch], ) execute_values( cur, "INSERT INTO csv_rows (content, source_file, row_index, embedding) VALUES %s", [ ( record["content"], record["source_file"], record["row_index"], Vector(item.embedding), ) for record, item in zip(batch, embeddings.data) ], ) conn.commit() cur.close() conn.close() print(f"Inserted {len(records)} rows into csv_rows") ``` ## Output Raw CSV rows joined as flat strings return results that match on surface keywords versus EnConvert chunks that preserve column labels and row context so retrieval is structurally grounded. ## Get started [Get free API key — 500 ops per month, no card required.](https://www.enconvert.com/auth?mode=signup) ## Related solutions - [How to Scrape Wikipedia with LlamaIndex](https://www.enconvert.com/solutions/scrape-wikipedia-with-llamaindex-enconvert.md) - [How to Ingest Markdown into LangChain with EnConvert](https://www.enconvert.com/solutions/ingest-markdown-into-langchain-with-heading-context.md) - [Best JigsawStack Alternative for Structured Data Extraction](https://www.enconvert.com/solutions/jigsawstack-alternative-for-structured-data-enconvert.md) - [How to Ingest Audio Transcripts into Qdrant with EnConvert](https://www.enconvert.com/solutions/ingest-audio-transcripts-into-qdrant-enconvert-api.md) - [How to Ingest CSV into CrewAI with EnConvert](https://www.enconvert.com/solutions/ingest-csv-into-crewai-with-enconvert.md) - [How to Ingest CSV into LlamaIndex with EnConvert](https://www.enconvert.com/solutions/ingest-csv-into-llamaindex-with-enconvert.md)