From 27c6d5805a7c8ad76410a445ff6f47a978e6cd62 Mon Sep 17 00:00:00 2001 From: issackhant Date: Wed, 10 Jun 2026 03:03:46 +0100 Subject: [PATCH] Add DuckDB connector (#3) - New connector src/connectors/duckdb_conn.py following the standard connect/list_tables/fetch_table pattern - DuckDB option in the Database Connectors UI (file path + table) - Tests covering list_tables, fetch_table and the row limit - README: usage snippet, connector status table and roadmap updated --- README.md | 16 +++++++++++--- app.py | 22 ++++++++++++++++++- pyproject.toml | 1 + requirements.txt | 1 + src/connectors/duckdb_conn.py | 17 +++++++++++++++ tests/test_pipeline.py | 41 +++++++++++++++++++++++++++++++++++ 6 files changed, 94 insertions(+), 4 deletions(-) create mode 100644 src/connectors/duckdb_conn.py diff --git a/README.md b/README.md index 11caab8..3692ff0 100644 --- a/README.md +++ b/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 diff --git a/app.py b/app.py index c6e46aa..ee1a5fc 100644 --- a/app.py +++ b/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('', unsafe_allow_html=True) if df is not None and not df.empty: diff --git a/pyproject.toml b/pyproject.toml index 311d41f..84b9cab 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/requirements.txt b/requirements.txt index f91dc07..4b43fa2 100644 --- a/requirements.txt +++ b/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 diff --git a/src/connectors/duckdb_conn.py b/src/connectors/duckdb_conn.py new file mode 100644 index 0000000..f2ed072 --- /dev/null +++ b/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 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index fb1e38e..56a6fda 100644 --- a/tests/test_pipeline.py +++ b/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):