Reliable Data Pipelines with Python & n8n: From SQL to AI Search

An AI assistant is only as good as the data behind it. If your data is out of date, duplicated, or copied across by hand, its answers will be wrong. ETL automation (extract, transform, load) keeps that data fresh without manual exports and without slowing down your main database.

At TENSIX (formerly HK Engineering), I combine n8n for visual workflows with small Python services for the heavy processing. Below is the setup I use to keep a normal SQL database in sync with a vector search engine (the index an AI uses to find relevant text).

1. The Setup: n8n for Workflows, Python for Heavy Lifting

Teams usually make one of two mistakes. They write one big Python script that runs on a timer, with no easy way to see when it fails. Or they push heavy data processing into a drag-and-drop tool, which can crash on large files.

I split the work into three layers instead:

1. Orchestration Tier (n8n): Handles webhook ingestion, cron scheduling, OAuth token refresh loops, rate-limiting backoffs, and error notification routing (Slack / Email alerts).
2. Processing Tier (Python / FastAPI): Executes CPU-intensive data normalization, markdown chunking, deduplication hashing, and batch OpenAI embedding generation.
3. Storage Tier (PostgreSQL + pgvector): Serves as the transactional source of truth and vector index, utilizing row-level locks and transactional upserts (`ON CONFLICT DO UPDATE`).

2. Copy Only What Changed, Without Slowing Your Database

To avoid slowing down the live database, the pipeline only picks up rows that changed since the last run (using timestamps). It also stores a fingerprint (a hash) of each piece of text, so text that has not changed is not processed again.

import hashlib
from typing import Dict, Any, List
import psycopg2
from psycopg2.extras import execute_values

def calculate_chunk_hash(text: str, metadata: Dict[str, Any]) -> str:
    """Generate SHA-256 hash of text content and key metadata fields to detect drift."""
    raw_payload = f"{text.strip()}::{metadata.get('updated_at', '')}"
    return hashlib.sha256(raw_payload.encode('utf-8')).hexdigest()

def upsert_vector_batch(conn, chunks: List[Dict[str, Any]]) -> int:
    """Execute high-speed transactional batch upsert into PostgreSQL pgvector table."""
    upsert_sql = """
        INSERT INTO enterprise_vectors (doc_id, chunk_index, content, content_hash, embedding, updated_at)
        VALUES %s
        ON CONFLICT (doc_id, chunk_index) DO UPDATE SET
            content = EXCLUDED.content,
            content_hash = EXCLUDED.content_hash,
            embedding = EXCLUDED.embedding,
            updated_at = NOW()
        WHERE enterprise_vectors.content_hash != EXCLUDED.content_hash;
    """
    
    # Prepare tuple values for execute_values
    records = [
        (c['doc_id'], c['chunk_index'], c['content'], c['hash'], c['embedding'], c['updated_at'])
        for c in chunks
    ]
    
    with conn.cursor() as cur:
        execute_values(cur, upsert_sql, records, page_size=500)
        conn.commit()
        return cur.rowcount

3. When Another App Fails: Retries and a Failure Log

Outside services such as Salesforce, HubSpot, or Stripe sometimes time out or refuse requests for a while (HTTP 429, "too many requests"). In n8n, I retry these calls with growing waits between attempts (exponential backoff), and keep a Dead Letter Queue (a failure log table) in PostgreSQL.

If a record still fails after 3 retries, the raw data and the error are saved to the etl_dead_letter_queue table and an alert goes out by Slack or email. The rest of the pipeline keeps running, and the failed record can be checked and replayed later.

4. What This Means for Your Business

Moving from manual exports to automated pipelines that retry on their own means less time spent fixing data by hand. Your reports and your internal AI search (RAG, an assistant that answers from your own documents) always work from current data.

To discuss custom software or automation for your company, visit TENSIX (formerly HK Engineering) or read Hemal Shah's profile.