Add DuckDB connector (#3)

README.md

@@ -517,6 +517,17 @@ df = fetch_table(
 )
 ```
 
+#### DuckDB
+
+```python
+from src.connectors.duckdb_conn import fetch_table
+
+df = fetch_table(
+    database="/path/to/my_database.duckdb",
+    table="my_table"
+)
+```
+
 ---
 
 ### Connector Status
@@ -528,9 +539,9 @@ df = fetch_table(
 | PostgreSQL | User/Pass | ✅ | ✅ | Stable |
 | MySQL | User/Pass | ✅ | ✅ | Stable |
 | BigQuery | Service Account JSON | ✅ | ✅ | Stable |
+| DuckDB | File path | ✅ | ✅ | Stable |
 | MongoDB | — | 🔜 | 🔜 | Planned |
 | Redshift | — | 🔜 | 🔜 | Planned |
-| DuckDB | — | 🔜 | 🔜 | Planned |
 | Microsoft Fabric | — | 🔜 | 🔜 | Planned |
 
 > Want to add a connector? See [Contributing](#contributing)
@@ -695,7 +706,6 @@ We want to support every major database. Next targets:
 |----------|-----------|-------|
 | MongoDB | Medium | #1 |
 | Redshift | Easy | #2 |
-| DuckDB | Easy | #3 |
 | Microsoft Fabric | Medium | #4 |
 | Elasticsearch | Hard | #5 |
 
@@ -884,7 +894,7 @@ tests/test_pipeline.py::TestCSVLoading::test_demo_csv_has_rows PASSED
 - [ ] pip package — `pip install multi-agent-data-pipeline`
 - [ ] MongoDB connector
 - [ ] Redshift connector
-- [ ] DuckDB connector
+- [x] DuckDB connector
 - [ ] Microsoft Fabric connector
 - [ ] Async parallel agent execution
 - [ ] Agent memory — learn from past runs

app.py

@@ -630,7 +630,7 @@ def run_pipeline_ui(df):
 
     db_type = st.selectbox(
         "Database",
-        ["Azure Databricks", "Snowflake", "PostgreSQL", "MySQL", "BigQuery"],
+        ["Azure Databricks", "Snowflake", "PostgreSQL", "MySQL", "BigQuery", "DuckDB"],
         label_visibility="collapsed"
     )
 
@@ -756,6 +756,26 @@ def run_pipeline_ui(df):
             else:
                 st.warning("Please fill all fields")
 
+    elif db_type == "DuckDB":
+        col1, col2 = st.columns(2)
+        with col1:
+            database = st.text_input("Database File", placeholder="/path/to/my_database.duckdb")
+        with col2:
+            table = st.text_input("Table", placeholder="my_table")
+
+        if st.button("🔌 Connect & Fetch Table"):
+            if database and table:
+                try:
+                    from src.connectors.duckdb_conn import fetch_table
+                    with st.spinner("Connecting to DuckDB..."):
+                        df = fetch_table(database, table)
+                    st.success(f"Connected — {len(df)} rows fetched")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill all fields")
+
     st.markdown('</div>', unsafe_allow_html=True)
 
     if df is not None and not df.empty:

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "psycopg2-binary>=2.9.0",
     "mysql-connector-python>=8.0.0",
     "google-cloud-bigquery>=3.0.0",
+    "duckdb>=1.0.0",
     "fpdf2>=2.7.0",
 ]
 

requirements.txt

@@ -17,6 +17,7 @@ cryptography==48.0.0
 databricks-sdk==0.112.0
 distro==1.9.0
 docstring_parser==0.18.0
+duckdb==1.5.3
 filelock==3.29.0
 gitdb==4.0.12
 GitPython==3.1.50

src/connectors/duckdb_conn.py

@@ -0,0 +1,17 @@
+import duckdb
+import pandas as pd
+
+def connect(database: str):
+    return duckdb.connect(database)
+
+def list_tables(database: str) -> list:
+    conn = connect(database)
+    tables = [row[0] for row in conn.execute("SHOW TABLES").fetchall()]
+    conn.close()
+    return tables
+
+def fetch_table(database: str, table: str, limit: int = 1000) -> pd.DataFrame:
+    conn = connect(database)
+    df = conn.execute(f"SELECT * FROM {table} LIMIT {limit}").df()
+    conn.close()
+    return df

tests/test_pipeline.py

@@ -128,6 +128,47 @@ def test_demo_csv_has_rows(self):
         assert len(df) > 0, "Demo CSV is empty"
 
 
+class TestDuckDBConnector:
+
+    @pytest.fixture
+    def duckdb_file(self, tmp_path):
+        import duckdb
+        db_path = str(tmp_path / "test.duckdb")
+        conn = duckdb.connect(db_path)
+        conn.execute("""
+            CREATE TABLE transactions (
+                transaction_id VARCHAR,
+                product_name VARCHAR,
+                unit_price DOUBLE
+            )
+        """)
+        conn.execute("""
+            INSERT INTO transactions VALUES
+            ('TXN001', 'Product A', 10.00),
+            ('TXN002', 'Product B', 15.00),
+            ('TXN003', 'Product C', 5.00)
+        """)
+        conn.close()
+        return db_path
+
+    def test_list_tables(self, duckdb_file):
+        from src.connectors.duckdb_conn import list_tables
+        tables = list_tables(duckdb_file)
+        assert "transactions" in tables
+
+    def test_fetch_table(self, duckdb_file):
+        from src.connectors.duckdb_conn import fetch_table
+        df = fetch_table(duckdb_file, "transactions")
+        assert len(df) == 3
+        assert "transaction_id" in df.columns
+        assert "unit_price" in df.columns
+
+    def test_fetch_table_limit(self, duckdb_file):
+        from src.connectors.duckdb_conn import fetch_table
+        df = fetch_table(duckdb_file, "transactions", limit=2)
+        assert len(df) == 2
+
+
 class TestPDFExists:
 
     def test_demo_pdf_exists(self):