Implementation:CrewAIInc CrewAI RAG PostgreSQL Loader
| Knowledge Sources | |
|---|---|
| Domains | RAG, Data_Loading, Database |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Executes SQL queries against PostgreSQL databases and formats the results into structured readable text for RAG ingestion.
Description
PostgresLoader extends BaseLoader to connect to PostgreSQL databases and execute queries. It requires a db_uri parameter passed via the metadata kwargs, supporting postgresql, postgres, and postgresql+psycopg2 URI schemes.
The loader parses the connection URI using urlparse to extract host, port (default 5432), user, password, and database name. It uses psycopg2 with RealDictCursor for dictionary-based row access.
The query execution formats results as structured text:
- A header line listing all column names.
- A total row count.
- Row-by-row data with each column value on an indented line, filtering out None values.
Content is truncated at 100KB to prevent excessive memory usage. The returned LoaderResult includes metadata with the source query, database name, row count, and column names. Empty results are handled with a descriptive message. Proper connection cleanup is ensured via try/finally.
Usage
Import PostgresLoader when you need to load data from a PostgreSQL database into the RAG knowledge base. It is typically instantiated automatically by the DataType.POSTGRES registry.
Code Reference
Source Location
- Repository: CrewAI
- File: lib/crewai-tools/src/crewai_tools/rag/loaders/postgres_loader.py
- Lines: 1-100
Signature
class PostgresLoader(BaseLoader):
def load(self, source: SourceContent, **kwargs) -> LoaderResult: ...
Import
from crewai_tools.rag.loaders.postgres_loader import PostgresLoader
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| source | SourceContent | Yes | Wraps a SQL query string (e.g., "SELECT * FROM table_name") |
| metadata (via kwargs) | dict | Yes | Must contain db_uri (str) with PostgreSQL connection URI |
Outputs
| Name | Type | Description |
|---|---|---|
| return | LoaderResult | Formatted text with column headers and row data; metadata includes source query, database name, row_count, and columns |
Usage Examples
Basic Usage
from crewai_tools.rag.loaders.postgres_loader import PostgresLoader
from crewai_tools.rag.source_content import SourceContent
loader = PostgresLoader()
source = SourceContent("SELECT id, title, created_at FROM articles ORDER BY created_at DESC LIMIT 50")
result = loader.load(
source,
metadata={"db_uri": "postgresql://user:password@localhost:5432/mydb"},
)
print(result.content)
# Columns: id, title, created_at
# Total rows: 50
#
# Row 1:
# id: 1
# title: Introduction to AI
# created_at: 2025-01-15
# ...
print(result.metadata)
# {'source': 'SELECT ...', 'database': 'mydb', 'row_count': 50, 'columns': ['id', 'title', 'created_at']}