From 64534a2a3b88c055b15cf006032ab5c5406eda86 Mon Sep 17 00:00:00 2001 From: Leonid Molotiievskyi Date: Mon, 29 Jun 2026 17:22:20 +0200 Subject: [PATCH 1/5] enable package build on main branch --- .github/workflows/publish-dev-rc.yml | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/publish-dev-rc.yml b/.github/workflows/publish-dev-rc.yml index 1a79c868..3a314f1d 100644 --- a/.github/workflows/publish-dev-rc.yml +++ b/.github/workflows/publish-dev-rc.yml @@ -23,10 +23,10 @@ jobs: outputs: rc_version: ${{ steps.compute.outputs.rc_version }} steps: - - name: Guard – dev branch only - if: github.ref != 'refs/heads/dev' + - name: Guard – dev or main branch only + if: github.ref != 'refs/heads/dev' && github.ref != 'refs/heads/main' run: | - echo "::error::This workflow may only be triggered from the 'dev' branch. Current ref: ${{ github.ref }}" + echo "::error::This workflow may only be triggered from the 'dev' or 'main' branch. Current ref: ${{ github.ref }}" exit 1 - name: Validate version input format @@ -105,8 +105,8 @@ jobs: # ── 2. Pip-install smoke test ───────────────────────────────────────────── # NOTE: The full pytest suite is intentionally NOT run here. This workflow is - # guarded to the 'dev' branch, and every commit on 'dev' is already tested by - # publish-dev.yml + # guarded to the 'dev' and 'main' branches, and every commit on those branches + # is already tested by publish-dev.yml / publish-main.yml respectively. test-pip-install: name: Smoke Test – pip install runs-on: ubuntu-latest From b434280e60686b99bc8e4a7f4bd3cd2967581f1f Mon Sep 17 00:00:00 2001 From: Leonid Molotiievskyi Date: Mon, 29 Jun 2026 21:10:36 +0200 Subject: [PATCH 2/5] Add ability to trigger the deploy job directly from the wranglerspy repository (#1031) --- .github/workflows/publish-dev-rc.yml | 28 +++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/.github/workflows/publish-dev-rc.yml b/.github/workflows/publish-dev-rc.yml index 3a314f1d..2cfaff17 100644 --- a/.github/workflows/publish-dev-rc.yml +++ b/.github/workflows/publish-dev-rc.yml @@ -108,7 +108,7 @@ jobs: # guarded to the 'dev' and 'main' branches, and every commit on those branches # is already tested by publish-dev.yml / publish-main.yml respectively. test-pip-install: - name: Smoke Test – pip install + name: smoke test runs-on: ubuntu-latest needs: [compute-version] permissions: @@ -130,7 +130,7 @@ jobs: # ── 3. Publish Python package to CodeArtifact ──────────────────────────── publish-codeartifact: - name: Publish RC to CodeArtifact + name: Publish RC Package runs-on: ubuntu-latest needs: [compute-version, test-pip-install] permissions: @@ -173,4 +173,26 @@ jobs: --repository ${{ vars.CODEARTIFACT_REPOSITORY }} - name: Publish to CodeArtifact - run: twine upload --repository codeartifact dist/* \ No newline at end of file + run: twine upload --repository codeartifact dist/* + + # ── 4. Trigger QA deployment in Lambda-Recipes ─────────────────────────── + # CROSS_REPO_PAT: a GitHub PAT with Actions read/write on Lambda-Recipes, + # stored as a repository or org secret. Required to dispatch workflows across repositories. + trigger-deploy-qa: + name: Trigger QA Deploy + runs-on: ubuntu-latest + needs: [compute-version, publish-codeartifact] + steps: + - name: Deploy QA + uses: convictional/trigger-workflow-and-wait@v1.6.5 + with: + owner: wrangleworks + repo: Lambda-Recipes + workflow_file_name: deploy-qa.yml + github_token: ${{ secrets.CROSS_REPO_PAT }} + ref: main + wait_interval: 15 + client_payload: '{"wrangles_version": "${{ needs.compute-version.outputs.rc_version }}"}' + propagate_failure: true + trigger_workflow: true + wait_workflow: true \ No newline at end of file From 74b94c8a9a15baa43df976ba789484eea1135c7f Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Wed, 1 Jul 2026 00:12:53 +0300 Subject: [PATCH 3/5] 825 add connectors for duckdb access (#1024) * Add DuckDB and Microsoft Access connectors Add DuckDB read/write/run connector with recipe schemas Add Microsoft Access read/write/run connector via ODBC Export new connectors from wrangles.connectors Document optional duckdb and pyodbc dependencies Add connector tests for DuckDB and Access * fix translate tests * fix duckdb tests --------- Co-authored-by: Eric Hills <53243273+ebhills@users.noreply.github.com> --- README.md | 2 + requirements-full.txt | 2 + tests/connectors/test_access.py | 109 ++++++++++++ tests/connectors/test_duckdb.py | 72 ++++++++ wrangles/connectors/__init__.py | 4 +- wrangles/connectors/access.py | 298 ++++++++++++++++++++++++++++++++ wrangles/connectors/duckdb.py | 205 ++++++++++++++++++++++ 7 files changed, 691 insertions(+), 1 deletion(-) create mode 100644 tests/connectors/test_access.py create mode 100644 tests/connectors/test_duckdb.py create mode 100644 wrangles/connectors/access.py create mode 100644 wrangles/connectors/duckdb.py diff --git a/README.md b/README.md index dab5f823..1469b7d1 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,8 @@ Connectors for databases, cloud storage, and external services require additiona | Capability | Install | |---|---| | Microsoft SQL Server | `pip install pymssql sqlalchemy` | +| Microsoft Access | `pip install pyodbc` | +| DuckDB | `pip install duckdb` | | PostgreSQL | `pip install psycopg2-binary sqlalchemy` | | MySQL | `pip install pymysql sqlalchemy` | | MongoDB | `pip install pymongo[srv]` | diff --git a/requirements-full.txt b/requirements-full.txt index 38b05081..15406ae9 100644 --- a/requirements-full.txt +++ b/requirements-full.txt @@ -8,5 +8,7 @@ # SQL databases sqlalchemy>=2.0,<3.0 +duckdb>=1.0.0 +pyodbc>=5.0.0 pymssql>=2.3.3 psycopg2-binary>=2.9.10 diff --git a/tests/connectors/test_access.py b/tests/connectors/test_access.py new file mode 100644 index 00000000..ec31891f --- /dev/null +++ b/tests/connectors/test_access.py @@ -0,0 +1,109 @@ +import pandas as pd +from types import SimpleNamespace +from unittest.mock import Mock + +from wrangles.connectors import access + + +class MockTables: + def __init__(self, exists): + self.exists = exists + + def fetchone(self): + return object() if self.exists else None + + +class MockCursor: + def __init__(self, table_exists=False): + self.table_exists = table_exists + self.executed = [] + self.executemany_calls = [] + + def tables(self, table=None, tableType=None): + return MockTables(self.table_exists) + + def execute(self, sql, params=()): + self.executed.append((sql, params)) + return self + + def executemany(self, sql, rows): + self.executemany_calls.append((sql, list(rows))) + return self + + +class MockConnection: + def __init__(self, cursor): + self.cursor_obj = cursor + self.committed = False + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def cursor(self): + return self.cursor_obj + + def commit(self): + self.committed = True + + +def test_connection_string_requires_database_or_connection_string(): + try: + access._connection_string() + assert False + except ValueError as e: + assert str(e) == 'database or connection_string must be provided' + + +def test_read_sql(monkeypatch): + data = pd.DataFrame({'Col1': ['Data1', 'Data2']}) + mock_connect = Mock(return_value=MockConnection(MockCursor())) + monkeypatch.setattr(access, "_pyodbc", SimpleNamespace(connect=mock_connect)) + monkeypatch.setattr(pd, "read_sql", Mock(return_value=data)) + + df = access.read( + database='database.accdb', + command='SELECT * from df_mock' + ) + + assert df.equals(data) + mock_connect.assert_called_once_with('DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=database.accdb;') + + +def test_write_sql_creates_table_and_inserts(monkeypatch): + cursor = MockCursor(table_exists=False) + monkeypatch.setattr(access, "_pyodbc", SimpleNamespace(connect=Mock(return_value=MockConnection(cursor)))) + + result = access.write( + df=pd.DataFrame({'Col1': ['Data1', 'Data2'], 'Col2': [1, 2]}), + database='database.accdb', + table='WrWx' + ) + + assert result is None + assert cursor.executed[0][0] == 'CREATE TABLE [WrWx] ([Col1] LONGTEXT, [Col2] INTEGER)' + assert cursor.executemany_calls[0] == ( + 'INSERT INTO [WrWx] ([Col1], [Col2]) VALUES (?, ?)', + [('Data1', 1), ('Data2', 2)] + ) + + +def test_run(monkeypatch): + cursor = MockCursor() + connection = MockConnection(cursor) + monkeypatch.setattr(access, "_pyodbc", SimpleNamespace(connect=Mock(return_value=connection))) + + result = access.run( + database='database.accdb', + command=['DELETE FROM WrWx WHERE Col1 = ?', 'UPDATE WrWx SET Col1 = ?'], + params=('Data1',) + ) + + assert result is None + assert cursor.executed == [ + ('DELETE FROM WrWx WHERE Col1 = ?', ('Data1',)), + ('UPDATE WrWx SET Col1 = ?', ('Data1',)) + ] + assert connection.committed diff --git a/tests/connectors/test_duckdb.py b/tests/connectors/test_duckdb.py new file mode 100644 index 00000000..fa7272f4 --- /dev/null +++ b/tests/connectors/test_duckdb.py @@ -0,0 +1,72 @@ +import importlib.util + +import pandas as pd +import pytest + +from wrangles.connectors import duckdb + + +pytestmark = pytest.mark.skipif( + importlib.util.find_spec('duckdb') is None, + reason='duckdb optional dependency is not installed' +) + + +def test_read_sql(tmp_path): + database = tmp_path / 'test.duckdb' + duckdb.run( + database=str(database), + command='CREATE TABLE df_mock AS SELECT \'Data1\' AS Col1, 1 AS Col2 UNION ALL SELECT \'Data2\', 2' + ) + + df = duckdb.read( + database=str(database), + command='SELECT * from df_mock ORDER BY Col2' + ) + + assert df.equals( + pd.DataFrame( + { + 'Col1': ['Data1', 'Data2'], + 'Col2': pd.array([1, 2], dtype='int32') + } + ) + ) + + +def test_write_sql(tmp_path): + database = tmp_path / 'write.duckdb' + df = pd.DataFrame({'Col1': ['Data1', 'Data2']}) + + duckdb.write( + df=df, + database=str(database), + table='temp_mock' + ) + + assert duckdb.read( + database=str(database), + command='SELECT * from temp_mock' + ).equals(df) + + +def test_run(tmp_path): + database = tmp_path / 'run.duckdb' + df = pd.DataFrame({'Col1': ['Data1', 'Data2']}) + duckdb.write( + df=df, + database=str(database), + table='test_table' + ) + + duckdb.run( + database=str(database), + command='CREATE TABLE test_table_copy AS SELECT * FROM test_table' + ) + + df_copy = duckdb.read( + database=str(database), + command='SELECT * from test_table_copy' + ) + + assert df.equals(df_copy) diff --git a/wrangles/connectors/__init__.py b/wrangles/connectors/__init__.py index e8039f7f..1c50c4c0 100644 --- a/wrangles/connectors/__init__.py +++ b/wrangles/connectors/__init__.py @@ -3,8 +3,10 @@ """ from . import akeneo +from . import access from . import ckan from . import concurrent +from . import duckdb from . import excel from . import file from . import http @@ -26,4 +28,4 @@ from . import train from . import jinja from . import _formatting -from . import input \ No newline at end of file +from . import input diff --git a/wrangles/connectors/access.py b/wrangles/connectors/access.py new file mode 100644 index 00000000..f7485f44 --- /dev/null +++ b/wrangles/connectors/access.py @@ -0,0 +1,298 @@ +""" +Connector to read/write from Microsoft Access databases using ODBC. +""" +import logging as _logging +from typing import Union as _Union + +import pandas as _pd +from pandas.api import types as _pd_types + +from ..utils import ( + LazyLoader as _LazyLoader, + wildcard_expansion as _wildcard_expansion, +) + + +_pyodbc = _LazyLoader('pyodbc') + +_schema = {} + + +def _quote_identifier(identifier: str) -> str: + return f"[{str(identifier).replace(']', ']]')}]" + + +def _connection_string( + database: str = None, + connection_string: str = None, + driver: str = 'Microsoft Access Driver (*.mdb, *.accdb)', + password: str = None, +) -> str: + if connection_string: + return connection_string + if database is None: + raise ValueError('database or connection_string must be provided') + + conn = f"DRIVER={{{driver}}};DBQ={database};" + if password: + conn += f"PWD={password};" + return conn + + +def _access_type(dtype) -> str: + if _pd_types.is_bool_dtype(dtype): + return 'BIT' + if _pd_types.is_integer_dtype(dtype): + return 'INTEGER' + if _pd_types.is_float_dtype(dtype): + return 'DOUBLE' + if _pd_types.is_datetime64_any_dtype(dtype): + return 'DATETIME' + return 'LONGTEXT' + + +def _table_exists(cursor, table: str) -> bool: + return cursor.tables(table=table, tableType='TABLE').fetchone() is not None + + +def _create_table(cursor, table: str, df: _pd.DataFrame) -> None: + columns = ', '.join( + f"{_quote_identifier(column)} {_access_type(dtype)}" + for column, dtype in df.dtypes.items() + ) + cursor.execute(f"CREATE TABLE {_quote_identifier(table)} ({columns})") + + +def read( + command: str, + database: str = None, + connection_string: str = None, + driver: str = 'Microsoft Access Driver (*.mdb, *.accdb)', + password: str = None, + columns: _Union[str, list] = None, + params: _Union[list, tuple] = None, + **kwargs +) -> _pd.DataFrame: + """ + Read data from a Microsoft Access database. + + >>> from wrangles.connectors import access + >>> df = access.read(database='database.accdb', command='SELECT * FROM table') + + :param command: SQL command to select data. + :param database: Access database file path. Not required if connection_string is supplied. + :param connection_string: Full ODBC connection string. If provided, database, driver, and password are ignored. + :param driver: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + :param password: Optional database password. + :param columns: (Optional) Subset of columns to be returned. This is less efficient than specifying in the SQL command. + :param params: (Optional) Variables to pass to a parameterized query. + """ + target = database or connection_string + _logging.info(f": Reading data from Microsoft Access :: {target}") + + conn_string = _connection_string(database, connection_string, driver, password) + with _pyodbc.connect(conn_string) as conn: + df = _pd.read_sql(command, conn, params=params, **kwargs) + + if columns is not None: + columns = _wildcard_expansion(df.columns, columns) + df = df[columns] + + return df + + +_schema['read'] = r""" +type: object +description: Import data from a Microsoft Access Database +required: + - command +properties: + database: + type: string + description: Access database file path. Not required if connection_string is supplied. + connection_string: + type: string + description: Full ODBC connection string. If provided, database, driver, and password are ignored. + driver: + type: string + description: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + password: + type: string + description: Optional database password. + command: + type: string + description: |- + SQL command to select data. + Note - using variables here can make your recipe vulnerable + to sql injection. Use params if using variables from + untrusted sources. + columns: + type: + - string + - array + description: A list with a subset of the columns to import. This is less efficient than specifying in the command. + params: + type: array + description: Variables to pass to a parameterized query. +""" + + +def write( + df: _pd.DataFrame, + table: str, + database: str = None, + connection_string: str = None, + driver: str = 'Microsoft Access Driver (*.mdb, *.accdb)', + password: str = None, + action: str = 'INSERT', + columns: _Union[str, list] = None, +) -> None: + """ + Write data to a Microsoft Access database. + + >>> from wrangles.connectors import access + >>> access.write(df, database='database.accdb', table='table') + + :param df: Pandas Dataframe to be written. + :param table: Table to be exported to. + :param database: Access database file path. Not required if connection_string is supplied. + :param connection_string: Full ODBC connection string. If provided, database, driver, and password are ignored. + :param driver: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + :param password: Optional database password. + :param action: INSERT appends, REPLACE recreates the table, FAIL errors if the table exists. Defaults to INSERT. + :param columns: (Optional) Subset of the columns to be written. If not provided, all columns will be output. + """ + target = database or connection_string + _logging.info(f": Writing data to Microsoft Access :: {target} / {table}") + + action = action.upper() + if action not in ('INSERT', 'REPLACE', 'FAIL'): + raise ValueError('Invalid action. Expected INSERT, REPLACE, or FAIL.') + + if columns is not None: + columns = _wildcard_expansion(df.columns, columns) + df = df[columns] + + conn_string = _connection_string(database, connection_string, driver, password) + with _pyodbc.connect(conn_string) as conn: + cursor = conn.cursor() + exists = _table_exists(cursor, table) + + if action == 'FAIL' and exists: + raise ValueError(f"Table already exists: {table}") + if action == 'REPLACE' and exists: + cursor.execute(f"DROP TABLE {_quote_identifier(table)}") + exists = False + if not exists: + _create_table(cursor, table, df) + + if not df.empty: + table_name = _quote_identifier(table) + column_names = ', '.join(_quote_identifier(column) for column in df.columns) + placeholders = ', '.join('?' for _ in df.columns) + sql = f"INSERT INTO {table_name} ({column_names}) VALUES ({placeholders})" + cursor.executemany(sql, df.where(_pd.notnull(df), None).itertuples(index=False, name=None)) + + conn.commit() + + +_schema['write'] = """ +type: object +description: Export data to a Microsoft Access Database +required: + - table +properties: + database: + type: string + description: Access database file path. Not required if connection_string is supplied. + connection_string: + type: string + description: Full ODBC connection string. If provided, database, driver, and password are ignored. + driver: + type: string + description: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + password: + type: string + description: Optional database password. + table: + type: string + description: The table to write to + action: + type: string + description: INSERT appends, REPLACE recreates the table, FAIL errors if the table exists. Defaults to INSERT. + enum: + - INSERT + - REPLACE + - FAIL + columns: + type: + - string + - array + description: A list of the columns to write to the table. If omitted, all columns will be written. +""" + + +def run( + command: _Union[str, list], + database: str = None, + connection_string: str = None, + driver: str = 'Microsoft Access Driver (*.mdb, *.accdb)', + password: str = None, + params: _Union[list, tuple] = None, +) -> None: + """ + Run a command on a Microsoft Access database. + + >>> wrangles.connectors.access.run( + >>> database='database.accdb', + >>> command='' + >>> ) + + :param command: SQL command or a list of SQL commands to execute. + :param database: Access database file path. Not required if connection_string is supplied. + :param connection_string: Full ODBC connection string. If provided, database, driver, and password are ignored. + :param driver: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + :param password: Optional database password. + :param params: Variables to pass to a parameterized query. + """ + target = database or connection_string + _logging.info(f": Executing Microsoft Access Command :: {target}") + + if isinstance(command, str): + command = [command] + + conn_string = _connection_string(database, connection_string, driver, password) + with _pyodbc.connect(conn_string) as conn: + cursor = conn.cursor() + for sql in command: + cursor.execute(sql, params or ()) + conn.commit() + + +_schema['run'] = r""" +type: object +description: Run a command against a Microsoft Access Database +required: + - command +properties: + database: + type: string + description: Access database file path. Not required if connection_string is supplied. + connection_string: + type: string + description: Full ODBC connection string. If provided, database, driver, and password are ignored. + driver: + type: string + description: ODBC driver name. Defaults to Microsoft Access Driver (*.mdb, *.accdb). + password: + type: string + description: Optional database password. + command: + type: + - string + - array + description: SQL command or a list of SQL commands to execute + params: + type: array + description: Variables to pass to a parameterized query. +""" diff --git a/wrangles/connectors/duckdb.py b/wrangles/connectors/duckdb.py new file mode 100644 index 00000000..22d1115c --- /dev/null +++ b/wrangles/connectors/duckdb.py @@ -0,0 +1,205 @@ +""" +Connector to read/write from DuckDB database files. +""" +import logging as _logging +from typing import Union as _Union + +import pandas as _pd + +from ..utils import ( + LazyLoader as _LazyLoader, + wildcard_expansion as _wildcard_expansion, +) + + +_duckdb = _LazyLoader('duckdb') + +_schema = {} + + +def _quote_identifier(identifier: str) -> str: + return '.'.join( + f'"{part.replace(chr(34), chr(34) * 2)}"' + for part in str(identifier).split('.') + ) + + +def read( + database: str, + command: str, + columns: _Union[str, list] = None, + params: _Union[list, tuple, dict] = None, + **kwargs +) -> _pd.DataFrame: + """ + Read data from a DuckDB database. + + >>> from wrangles.connectors import duckdb + >>> df = duckdb.read(database='database.duckdb', command='SELECT * FROM table') + + :param database: The database to connect to including the file path. Use ':memory:' for an in-memory database. + :param command: SQL command to select data. + :param columns: (Optional) Subset of columns to be returned. This is less efficient than specifying in the SQL command. + :param params: (Optional) Variables to pass to a parameterized query. + """ + _logging.info(f": Reading data from DuckDB :: {database}") + + with _duckdb.connect(database=database, **kwargs) as conn: + df = conn.execute(command, params or ()).fetchdf() + + if columns is not None: + columns = _wildcard_expansion(df.columns, columns) + df = df[columns] + + return df + + +_schema['read'] = r""" +type: object +description: Import data from a DuckDB Database +required: + - database + - command +properties: + database: + type: string + description: The database to connect to including the file path. Use ':memory:' for an in-memory database. + command: + type: string + description: |- + SQL command to select data. + Note - using variables here can make your recipe vulnerable + to sql injection. Use params if using variables from + untrusted sources. + columns: + type: + - string + - array + description: A list with a subset of the columns to import. This is less efficient than specifying in the command. + params: + type: + - array + - object + description: Variables to pass to a parameterized query. +""" + + +def write( + df: _pd.DataFrame, + database: str, + table: str, + action: str = 'INSERT', + columns: _Union[str, list] = None, + **kwargs +) -> None: + """ + Write data to a DuckDB database. + + >>> from wrangles.connectors import duckdb + >>> duckdb.write(df, database='database.duckdb', table='table') + + :param df: Pandas Dataframe to be written. + :param database: The database to connect to including the file path. Use ':memory:' for an in-memory database. + :param table: Table to be exported to. + :param action: INSERT appends, REPLACE recreates the table, FAIL errors if the table exists. Defaults to INSERT. + :param columns: (Optional) Subset of the columns to be written. If not provided, all columns will be output. + """ + _logging.info(f": Writing data to DuckDB :: {database} / {table}") + + action = action.upper() + if columns is not None: + columns = _wildcard_expansion(df.columns, columns) + df = df[columns] + + table_name = _quote_identifier(table) + with _duckdb.connect(database=database, **kwargs) as conn: + conn.register('_wrangles_df', df) + + if action == 'FAIL': + conn.execute(f'CREATE TABLE {table_name} AS SELECT * FROM _wrangles_df') + elif action == 'REPLACE': + conn.execute(f'CREATE OR REPLACE TABLE {table_name} AS SELECT * FROM _wrangles_df') + elif action == 'INSERT': + conn.execute(f'CREATE TABLE IF NOT EXISTS {table_name} AS SELECT * FROM _wrangles_df WHERE false') + conn.execute(f'INSERT INTO {table_name} SELECT * FROM _wrangles_df') + else: + raise ValueError('Invalid action. Expected INSERT, REPLACE, or FAIL.') + + +_schema['write'] = """ +type: object +description: Export data to a DuckDB Database +required: + - database + - table +properties: + database: + type: string + description: The database to connect to including the file path. Use ':memory:' for an in-memory database. + table: + type: string + description: The table to write to + action: + type: string + description: INSERT appends, REPLACE recreates the table, FAIL errors if the table exists. Defaults to INSERT. + enum: + - INSERT + - REPLACE + - FAIL + columns: + type: + - string + - array + description: A list of the columns to write to the table. If omitted, all columns will be written. +""" + + +def run( + database: str, + command: _Union[str, list], + params: _Union[list, tuple, dict] = None, + **kwargs +) -> None: + """ + Run a command on a DuckDB database. + + >>> wrangles.connectors.duckdb.run( + >>> database='database.duckdb', + >>> command='' + >>> ) + + :param database: The database to connect to including the file path. Use ':memory:' for an in-memory database. + :param command: SQL command or a list of SQL commands to execute. + :param params: Variables to pass to a parameterized query. + """ + _logging.info(f": Executing DuckDB Command :: {database}") + + if isinstance(command, str): + command = [command] + + with _duckdb.connect(database=database, **kwargs) as conn: + for sql in command: + conn.execute(sql, params or ()) + + +_schema['run'] = r""" +type: object +description: Run a command against a DuckDB Database +required: + - database + - command +properties: + database: + type: string + description: The database to connect to including the file path. Use ':memory:' for an in-memory database. + command: + type: + - string + - array + description: SQL command or a list of SQL commands to execute + params: + type: + - array + - object + description: Variables to pass to a parameterized query. +""" From 369dfff887ab404e22d0bd68abbb68ddc2c03cd0 Mon Sep 17 00:00:00 2001 From: Leonid Molotiievskyi Date: Wed, 1 Jul 2026 13:55:00 +0200 Subject: [PATCH 4/5] build full version of the package (#1035) --- .github/workflows/publish-dev-rc.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/publish-dev-rc.yml b/.github/workflows/publish-dev-rc.yml index 2cfaff17..036cf046 100644 --- a/.github/workflows/publish-dev-rc.yml +++ b/.github/workflows/publish-dev-rc.yml @@ -162,6 +162,12 @@ jobs: - name: Install build tooling run: python -m pip install build twine --user + - name: Patch setup.py to use requirements-full.txt + run: | + sed -i "s/requirements\.txt/requirements-full.txt/" setup.py + echo "setup.py requirements line after patch:" + grep "requirements" setup.py + - name: Build sdist and wheel run: python -m build --sdist --wheel --outdir dist/ . From ab581495a07d9a75d2f38dfd1a485fd879af247a Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Thu, 2 Jul 2026 00:44:14 +0300 Subject: [PATCH 5/5] Add split.dictionary to_lists mode (#1021) * Add split.dictionary to_lists mode * fix tests * renamed parameters --- tests/recipes/wrangles/test_split.py | 169 +++++++++++++++++++++++++++ wrangles/recipe_wrangles/split.py | 50 ++++++-- 2 files changed, 210 insertions(+), 9 deletions(-) diff --git a/tests/recipes/wrangles/test_split.py b/tests/recipes/wrangles/test_split.py index c316f195..509f9df8 100644 --- a/tests/recipes/wrangles/test_split.py +++ b/tests/recipes/wrangles/test_split.py @@ -1281,6 +1281,175 @@ def test_split_dictionary_empty(self): ) assert df.empty and df.columns.to_list() == ['Col'] + def test_split_dictionary_to_lists(self): + """ + Test splitting dictionary keys and values to parallel lists + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output: + - Keys + - Values + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'My Dict': [ + {"a": 123, "b": "kshdf"}, + {"b": 123, "c": "kshdf"} + ] + }) + ) + assert df['Keys'].to_list() == [["a", "b"], ["b", "c"]] + assert df['Values'].to_list() == [[123, "kshdf"], [123, "kshdf"]] + + def test_split_dictionary_to_lists_json(self): + """ + Test splitting JSON dictionary keys and values to lists + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output: + - Keys + - Values + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'My Dict': ['{"a": 123, "b": "kshdf"}'] + }) + ) + assert df['Keys'][0] == ["a", "b"] + assert df['Values'][0] == [123, "kshdf"] + + def test_split_dictionary_to_lists_multiple_inputs(self): + """ + Test splitting multiple dictionaries to keys and values lists + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: + - Dict1 + - Dict2 + output: + - Keys + - Values + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'Dict1': [{"a": 1, "b": 2}], + 'Dict2': [{"b": 3, "c": 4}] + }) + ) + assert df['Keys'][0] == ["a", "b", "c"] + assert df['Values'][0] == [1, 3, 4] + + def test_split_dictionary_to_lists_where(self): + """ + Test split.dictionary output_format to_lists using where + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output: + - Keys + - Values + output_format: to_lists + where: numbers > 3 + """, + dataframe=pd.DataFrame({ + 'My Dict': [ + {"a": 1}, + {"b": 2}, + {"c": 3} + ], + 'numbers': [3, 4, 5] + }) + ) + assert df['Keys'].to_list() == ["", ["b"], ["c"]] + assert df['Values'].to_list() == ["", [2], [3]] + + def test_split_dictionary_to_lists_default_output(self): + """ + Test split.dictionary output_format to_lists default output columns + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'My Dict': [{"a": 1}] + }) + ) + assert df['Keys'][0] == ["a"] + assert df['Values'][0] == [1] + + def test_split_dictionary_to_lists_output_error(self): + """ + Test split.dictionary output_format to_lists validates output column count + """ + with pytest.raises(ValueError, match="exactly two output columns"): + wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output: Keys + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'My Dict': [{"a": 1}] + }) + ) + + def test_split_dictionary_invalid_output_format(self): + """ + Test split.dictionary validates output_format + """ + with pytest.raises(ValueError, match="output_format must be one of"): + wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output_format: invalid + """, + dataframe=pd.DataFrame({ + 'My Dict': [{"a": 1}] + }) + ) + + def test_split_dictionary_to_lists_empty(self): + """ + Test split.dictionary output_format to_lists with an empty column + """ + df = wrangles.recipe.run( + """ + wrangles: + - split.dictionary: + input: My Dict + output: + - Keys + - Values + output_format: to_lists + """, + dataframe=pd.DataFrame({ + 'My Dict': [] + }) + ) + assert df.empty and df.columns.to_list() == ['My Dict', 'Keys', 'Values'] + class TestTokenize: """ diff --git a/wrangles/recipe_wrangles/split.py b/wrangles/recipe_wrangles/split.py index 0d1a3468..01f18cd0 100644 --- a/wrangles/recipe_wrangles/split.py +++ b/wrangles/recipe_wrangles/split.py @@ -17,7 +17,8 @@ def dictionary( df: _pd.DataFrame, input: _Union[str, int, _list], output: _Union[str, _list] = None, - default: dict = None + default: dict = None, + output_format: str = "columns" ) -> _pd.DataFrame: """ type: object @@ -30,7 +31,7 @@ def dictionary( - input properties: input: - type: + type: - string - integer - array @@ -39,22 +40,36 @@ def dictionary( If providing multiple dictionaries and the dictionaries contain overlapping values, the last value will be returned. output: - type: + type: - string - array description: |- - (Optional) Subset of keys to extract from the dictionary. - If not provided, all keys will be returned. + In columns output_format, this is an optional subset of keys to extract + from the dictionary. If not provided, all keys will be returned. Columns can be renamed with the following syntax: output: - key1: new_column_name1 - key2: new_column_name2 + In to_lists output_format, this must be two output columns for the keys + and values lists. If not provided, Keys and Values will be used. default: type: object description: >- Provide a set of default headings and values if they are not found within the input - """ + output_format: + type: string + enum: + - columns + - to_lists + description: |- + How to split the dictionary. + columns creates one output column for each dictionary key. + to_lists creates two output columns containing lists of keys and values. + """ + if output_format not in ["columns", "to_lists"]: + raise ValueError("output_format must be one of: columns, to_lists") + if default is None: default = {} _logging.debug(f": Splitting dictionaries :: input :: {input}") @@ -73,11 +88,28 @@ def _parse_dict_or_json(val): raise ValueError(f'{val} is not a valid Dictionary') from None - # Generate new columns for each key in the dictionary - df_temp = _pd.DataFrame([ + # Merge each row's dictionaries so duplicate keys follow existing behavior: + # later input columns overwrite earlier input columns. + dicts = [ dict(_itertools.chain.from_iterable(_parse_dict_or_json(d) for d in ([default] + row.tolist()))) for row in df[input].values - ]) + ] + + if output_format == "to_lists": + if output is None: + output = ["Keys", "Values"] + elif not isinstance(output, _list): + output = [output] + + if len(output) != 2: + raise ValueError("split.dictionary with output_format to_lists requires exactly two output columns") + + df[output[0]] = [[key for key in item.keys()] for item in dicts] + df[output[1]] = [[value for value in item.values()] for item in dicts] + return df + + # Generate new columns for each key in the dictionary + df_temp = _pd.DataFrame(dicts) # If user has defined how they'd like the output columns if output is not None: