SQLAlchemyTableRetriever

Connects to any SQLAlchemy-supported database and executes an SQL query.

Most common position in a pipeline	Before a PromptBuilder
Mandatory init variables	drivername: SQLAlchemy driver name, for example sqlite, postgresql+psycopg2, mysql+pymysql, or mssql+pyodbc. For real database backends you will also need host, port, database, username, and password.
Mandatory run variables	query: An SQL query to execute
Output variables	dataframe: The query result as a Pandas DataFrame table: The same result rendered as a Markdown table error: Error message if the query failed, empty string otherwise
API reference	SQLAlchemy
GitHub link	https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/sqlalchemy
Package name	sqlalchemy-haystack

Overview

SQLAlchemyTableRetriever is a backend-agnostic table retriever: it speaks to anything SQLAlchemy speaks to — PostgreSQL, MySQL, SQLite, MSSQL, and the long tail of dialects covered by third-party drivers. Give it a SQL query and it hands you back the result as both a Pandas DataFrame (under dataframe) and a ready-to-render Markdown table (under table), which is convenient for piping into a prompt. Results are capped at 10,000 rows.

If the query fails, the component does not raise — it returns an empty DataFrame and puts the SQLAlchemy error string in the error output. That makes it safe to drop into a pipeline without wrapping the whole thing in a try/except.

Connection parameters

The init arguments map directly to SQLAlchemy's URL parts:

• drivername — the only strictly required one. Pick the driver that matches your backend, e.g. postgresql+psycopg2, mysql+pymysql, sqlite, mssql+pyodbc.
• host, port, database, username — standard connection bits. Pass whatever your backend needs.
• password — a Haystack Secret. Resolve it from an environment variable with Secret.from_env_var("MY_DB_PASSWORD"), or inline with Secret.from_token("…") (not recommended for anything other than local tinkering).

For SQLite, drivername="sqlite" with database=":memory:" is enough — no host/user/password needed.

init_script

Pass init_script to run one or more SQL statements once, in a single transaction, the first time the component is warmed up. Typical uses:

• Seeding an in-memory SQLite database for demos or tests.
• Creating temporary views or session-level settings before queries run.

Each entry in the list is a single statement.

Usage

Install the sqlalchemy-haystack package, plus the driver for your database:

shell

pip install sqlalchemy-haystack
# For PostgreSQL, also install a driver:
pip install psycopg2-binary

On its own

A self-contained example using an in-memory SQLite database seeded via init_script:

python

from haystack_integrations.components.retrievers.sqlalchemy import SQLAlchemyTableRetriever

retriever = SQLAlchemyTableRetriever(
    drivername="sqlite",
    database=":memory:",
    init_script=[
        "CREATE TABLE employees (name TEXT, salary INTEGER)",
        "INSERT INTO employees VALUES ('Ada', 90000), ('Linus', 85000), ('Grace', 95000)",
    ],
)

result = retriever.run(query="SELECT name, salary FROM employees ORDER BY salary DESC")
print(result["dataframe"])
print(result["table"])

Connecting to a real backend looks the same — swap the driver and pass connection details:

python

from haystack.utils import Secret
from haystack_integrations.components.retrievers.sqlalchemy import SQLAlchemyTableRetriever

retriever = SQLAlchemyTableRetriever(
    drivername="postgresql+psycopg2",
    host="db.example.com",
    port=5432,
    database="analytics",
    username="readonly",
    password=Secret.from_env_var("ANALYTICS_DB_PASSWORD"),
)

In a pipeline

Use the retriever's Markdown table output as context for an LLM — for example, asking an LLM to summarize a query result:

python

from haystack import Pipeline
from haystack.utils import Secret
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.retrievers.sqlalchemy import SQLAlchemyTableRetriever

retriever = SQLAlchemyTableRetriever(
    drivername="postgresql+psycopg2",
    host="db.example.com",
    port=5432,
    database="analytics",
    username="readonly",
    password=Secret.from_env_var("ANALYTICS_DB_PASSWORD"),
)

pipeline = Pipeline()
pipeline.add_component(
    "builder",
    ChatPromptBuilder(
        template=[ChatMessage.from_user("Describe this table: {{ table }}")],
        required_variables="*",
    ),
)
pipeline.add_component("db", retriever)
pipeline.add_component("llm", OpenAIChatGenerator(model="gpt-4o"))

pipeline.connect("db.table", "builder.table")
pipeline.connect("builder.prompt", "llm.messages")

pipeline.run(data={"query": "SELECT employee, salary FROM employees LIMIT 10"})