Add dagster-adbc integration (#272)

Co-authored-by: Emil Sadek <esadek@users.noreply.github.com>

Author: esadek

libraries/dagster-adbc/Makefile

@@ -0,0 +1,15 @@
+install:
+	uv sync
+
+build:
+	uv build
+
+test:
+	uv run pytest
+
+ruff:
+	uv run ruff check --fix
+	uv run ruff format
+
+check:
+	uv run pyright

libraries/dagster-adbc/README.md

@@ -0,0 +1,57 @@
+# dagster-adbc
+
+A Dagster module that provides an integration with [ADBC](https://arrow.apache.org/adbc/current/index.html).
+
+## Installation
+
+The `dagster_adbc` module is available as a PyPI package - install with your preferred python environment manager (We recommend [uv](https://github.com/astral-sh/uv)).
+
+```sh
+uv venv
+source .venv/bin/activate
+uv pip install dagster-adbc
+```
+
+Additionally, the ADBC driver for your database must be installed.
+We recommend [dbc](https://github.com/columnar-tech/dbc) for installing drivers.
+
+```sh
+uv pip install dbc
+dbc install flightsql
+```
+
+## Example Usage
+
+```python
+from dagster import Definitions, EnvVar, asset
+from dagster_adbc import ADBCResource
+
+
+@asset
+def my_table(dremio: ADBCResource) -> None:
+    with dremio.get_connection() as connection, connection.cursor() as cursor:
+        cursor.execute("SELECT * FROM my_table")
+        table = cursor.fetch_arrow_table()
+
+
+defs = Definitions(
+    assets=[my_table],
+    resources={
+        "dremio": ADBCResource(
+            driver="flightsql",
+            uri="grpc+tcp://localhost:32010",
+            db_kwargs={"username": "admin", "password": EnvVar("DREMIO_PASSWORD")},
+        )
+    },
+)
+```
+
+## Development
+
+The `Makefile` provides the tools required to test and lint your local installation.
+
+```sh
+make test
+make ruff
+make check
+```

libraries/dagster-adbc/dagster_adbc/__init__.py

@@ -0,0 +1,8 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
+from dagster_adbc.resource import ADBCResource
+
+__all__ = ["ADBCResource"]
+__version__ = "0.0.1"
+
+DagsterLibraryRegistry.register("dagster-adbc", __version__, is_dagster_package=False)

libraries/dagster-adbc/dagster_adbc/resource.py

@@ -0,0 +1,76 @@
+from collections.abc import Mapping
+from contextlib import contextmanager
+from typing import Any, Generator
+
+from adbc_driver_manager import dbapi
+from dagster import ConfigurableResource
+from pydantic import Field
+
+
+class ADBCResource(ConfigurableResource):
+    """Resource for interacting with a database via ADBC.
+    Wraps an underlying adbc_driver_manager.dbapi connection.
+
+    Examples:
+        .. code-block:: python
+        from dagster import Definitions, EnvVar, asset
+        from dagster_adbc import ADBCResource
+
+
+        @asset
+        def my_table(dremio: ADBCResource) -> None:
+            with dremio.get_connection() as connection, connection.cursor() as cursor:
+                cursor.execute("SELECT * FROM my_table")
+                table = cursor.fetch_arrow_table()
+
+
+        defs = Definitions(
+            assets=[my_table],
+            resources={
+                "dremio": ADBCResource(
+                    driver="flightsql",
+                    uri="grpc+tcp://localhost:32010",
+                    db_kwargs={"username": "admin", "password": EnvVar("DREMIO_PASSWORD")},
+                )
+            },
+        )
+    """
+
+    driver: str | None = Field(default=None, description="The driver to use.")
+    uri: str | None = Field(
+        default=None, description='The "uri" parameter to the database (if applicable).'
+    )
+    profile: str | None = Field(default=None, description="A connection profile to load.")
+    entrypoint: str | None = Field(
+        default=None,
+        description="The driver-specific entrypoint, if different than the default.",
+    )
+    db_kwargs: Mapping[str, str] | None = Field(
+        default=None,
+        description="Key-value parameters to pass to the driver to initialize the database.",
+    )
+    conn_kwargs: Mapping[str, str] | None = Field(
+        default=None,
+        description="Key-value parameters to pass to the driver to initialize the connection.",
+    )
+    autocommit: bool = Field(default=False, description="Whether to enable autocommit.")
+
+    @classmethod
+    def _is_dagster_mainatained(cls) -> bool:
+        return False
+
+    @contextmanager
+    def get_connection(self) -> Generator[dbapi.Connection, Any, None]:
+        connection = dbapi.connect(
+            driver=self.driver,
+            uri=self.uri,
+            profile=self.profile,
+            entrypoint=self.entrypoint,
+            db_kwargs=self.db_kwargs,
+            conn_kwargs=self.conn_kwargs,
+            autocommit=self.autocommit,
+        )
+        try:
+            yield connection
+        finally:
+            connection.close()

libraries/dagster-adbc/dagster_adbc_tests/__init__.py

@@ -0,0 +1,25 @@
+import subprocess
+from typing import Any, Generator
+
+import pyarrow as pa
+import pytest
+
+from dagster_adbc import ADBCResource
+
+
+@pytest.fixture(autouse=True)
+def sqlite_adbc_driver() -> Generator[None, Any, None]:
+    subprocess.run(["dbc", "install", "sqlite"])
+    yield
+    subprocess.run(["dbc", "uninstall", "sqlite"])
+
+
+def test_connection() -> None:
+    resource = ADBCResource(driver="sqlite", uri=":memory:")
+    with resource.get_connection() as connection, connection.cursor() as cursor:
+        cursor.execute("SELECT 1 AS value")
+        table = cursor.fetch_arrow_table()
+
+    assert isinstance(table, pa.Table)
+    assert table.num_rows == 1
+    assert table.column("value")[0].as_py() == 1

libraries/dagster-adbc/dagster_adbc_tests/test_resource.py

@@ -0,0 +1,7 @@
+from importlib.metadata import version
+
+import dagster_adbc
+
+
+def test_version() -> None:
+    assert version("dagster-adbc") == dagster_adbc.__version__

libraries/dagster-adbc/dagster_adbc_tests/test_version.py

@@ -0,0 +1,35 @@
+[project]
+name = "dagster-adbc"
+description = "Dagster integration with ADBC"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "adbc-driver-manager>=1.11.0",
+    "dagster>=1.13.0",
+    "pyarrow>=23.0.1",
+]
+dynamic = ["version"]
+
+[dependency-groups]
+dev = [
+    "dbc>=0.2.0",
+    "pyright>=1.1.408",
+    "pytest>=9.0.3",
+    "ruff>=0.15.9",
+]
+
+[build-system]
+requires = ["setuptools>=42"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["dagster_adbc"]
+
+[tool.setuptools.dynamic]
+version = {attr = "dagster_adbc.__version__"}
+
+[tool.ruff]
+line-length = 100
+
+[tool.ruff.lint]
+select = ["ANN", "E", "F", "I"]