feat(utils): add AdlsLakeFileSystem for Azure Data Lake Storage (#70)

**New ADLS Gen2 backend and API enhancements:**

* Added the `AdlsLakeFileSystem` class, providing direct access to Azure
Data Lake Storage Gen2 via the Azure SDK, with methods matching
`LakeFileSystem` and additional directory listing support.
* Updated `__init__.py` to support lazy importing of
`AdlsLakeFileSystem`, preventing hard dependency on Azure packages
unless needed.
* Extended the `pyproject.toml` to include `azure-storage-file-datalake`
in the `azure` extra requirements for proper dependency management.

**Documentation improvements:**

* Revised `docs/packages/dataorc-utils/lake/index.md` to explain both
`LakeFileSystem` and `AdlsLakeFileSystem`, their APIs, constructor
options, path handling differences, and authentication details.

**Testing enhancements:**

* Consolidated tests for `LakeFileSystem` to also work for
`AdlsLakeFileSystem`.

Author: TordAreStromsnes

docs/packages/dataorc-utils/lake/index.md

@@ -4,19 +4,28 @@ title: dataorc-utils - Lake
 
 # dataorc-utils — Lake
 
-Filesystem utilities for reading and writing to the Data Lake in Databricks pipelines.
+Filesystem utilities for reading and writing to the Data Lake.
 
 ## Overview
 
-The `lake` module provides a unified interface for file operations on Azure Data Lake Storage,
-abstracting away the differences between local development and Databricks runtime environments.
+The `lake` module provides a unified interface for file operations on Azure Data Lake Storage.
+Two implementations are available:
+
+| Class | Backend | Use case |
+|-------|---------|----------|
+| `LakeFileSystem` | Local / FUSE mount (via `fsspec`) | Databricks with mounted storage |
+| `AdlsLakeFileSystem` | ADLS Gen2 SDK (direct) | Any environment — no mounts or dbutils needed |
+
+Both classes expose the **same core API** (`read_text`, `write_text`, `read_json`, `write_json`,
+`exists`, `delete`), so switching between them requires only changing the constructor.
 
 **Key design principle:** The module is **path-agnostic**. It performs pure I/O operations
-without assuming any specific mounting conventions. Path normalization (e.g., `dls://` → `/mnt/...`)
-is the responsibility of your pipeline code.
+without assuming any specific mounting conventions.
 
 ## Quick start
 
+### LakeFileSystem (mount-based)
+
 ```python
 from dataorc_utils.lake import LakeFileSystem
 
@@ -36,6 +45,40 @@ if fs.exists("old_file.txt"):
     fs.delete("old_file.txt")
 ```
 
+### AdlsLakeFileSystem (direct ADLS Gen2)
+
+!!! note "Requires the `azure` extra"
+    Install with: `pip install dataorc-utils[azure]`
+
+```python
+from dataorc_utils.lake import AdlsLakeFileSystem
+
+# Connect directly to ADLS Gen2 — no mounts or dbutils required
+fs = AdlsLakeFileSystem(
+    account_url="https://testdatadevsc.dfs.core.windows.net",
+    container="bronze",
+    base_path="sales/orders",          # optional prefix inside the container
+)
+
+# Same API from here on
+fs.write_text("metadata.txt", "Pipeline run: 2026-02-02")
+content = fs.read_text("metadata.txt")
+
+fs.write_json("config.json", {"version": 1, "status": "complete"})
+config = fs.read_json("config.json")
+
+if fs.exists("old_file.txt"):
+    fs.delete("old_file.txt")
+
+# Bonus: list files in a directory
+files = fs.list_paths("_metadata/")
+```
+
+Authentication uses `DefaultAzureCredential` by default, which supports
+Managed Identity, Azure CLI (`az login`), and environment variables.
+You can also pass a custom credential via the `credential` parameter
+(e.g. `ManagedIdentityCredential()`).
+
 ## API Reference
 
 ### LakeFileSystem
@@ -77,6 +120,48 @@ fs = LakeFileSystem(base_path="/dbfs/mnt/datalakestore/bronze")
 | `exists(path)` | `bool` | Check if a file or directory exists. |
 | `delete(path)` | `bool` | Delete a file. Returns `True` if deleted, `False` if didn't exist. |
 
+---
+
+### AdlsLakeFileSystem
+
+Direct connection to ADLS Gen2 — no mounts or Databricks utilities required.
+
+#### Constructor
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `account_url` | `str` | Full DFS endpoint, e.g. `"https://<account>.dfs.core.windows.net"` |
+| `container` | `str` | File-system / container name, e.g. `"bronze"` |
+| `base_path` | `str` | Optional prefix inside the container prepended to every path. Defaults to `""`. |
+| `credential` | `Any \| None` | Any Azure credential accepted by the SDK. Defaults to `DefaultAzureCredential()`. |
+
+#### Methods
+
+`AdlsLakeFileSystem` exposes the same text, JSON, and directory methods as `LakeFileSystem`,
+plus one additional method:
+
+##### Text Operations
+
+| Method | Returns | Description |
+|--------|---------|-------------|
+| `read_text(path)` | `str \| None` | Read a UTF-8 text file. Returns `None` if the file doesn't exist. |
+| `write_text(path, content)` | `None` | Write (or overwrite) a UTF-8 text file. |
+
+##### JSON Operations
+
+| Method | Returns | Description |
+|--------|---------|-------------|
+| `read_json(path)` | `dict \| None` | Read and parse a JSON file. Returns `None` if the file doesn't exist or parse fails. |
+| `write_json(path, data, indent=2)` | `None` | Write a dictionary as JSON. |
+
+##### Directory Operations
+
+| Method | Returns | Description |
+|--------|---------|-------------|
+| `exists(path)` | `bool` | Check if a file exists. |
+| `delete(path)` | `bool` | Delete a file. Returns `True` if deleted, `False` otherwise. |
+| `list_paths(path, recursive=True)` | `list[str]` | List file paths under `path`. Returns paths relative to `base_path`. |
+
 ## Usage in Pipelines
 
 ### With CorePipelineConfig
@@ -105,6 +190,8 @@ fs.write_json("_metadata/run_info.json", {
 
 ### Path Handling
 
+#### LakeFileSystem
+
 The module does **not** perform path normalization. Your pipeline code is responsible for
 providing correct absolute paths for the runtime environment.
 
@@ -119,6 +206,12 @@ fs = LakeFileSystem()
 fs.write_text("/dbfs/mnt/datalakestore/bronze/file.txt", "content")
 ```
 
+#### AdlsLakeFileSystem
+
+Paths are always **relative to the container and `base_path`** — no mount prefixes needed.
+For example, with `container="bronze"` and `base_path="sales/orders"`,
+calling `fs.write_text("file.txt", ...)` resolves to `bronze/sales/orders/file.txt`.
+
 ### Error Handling
 
 The module returns `None` for missing files rather than raising exceptions:

packages/dataorc-utils/pyproject.toml

@@ -29,6 +29,7 @@ dev = [
 azure = [
     "azure-identity",
     "azure-keyvault-secrets",
+    "azure-storage-file-datalake",
 ]
 
 [tool.ruff]

packages/dataorc-utils/src/dataorc_utils/lake/__init__.py

@@ -1,5 +1,6 @@
-"""Data lake filesystem utilities for Databricks pipelines."""
+"""Data lake filesystem utilities."""
 
+from .adls_filesystem import AdlsLakeFileSystem
 from .filesystem import LakeFileSystem
 
-__all__ = ["LakeFileSystem"]
+__all__ = ["AdlsLakeFileSystem", "LakeFileSystem"]

packages/dataorc-utils/src/dataorc_utils/lake/adls_filesystem.py

@@ -0,0 +1,143 @@
+"""AdlsLakeFileSystem - ADLS Gen2-backed file operations.
+
+Drop-in alternative for LakeFileSystem that connects directly to
+Azure Data Lake Storage Gen2 via the Azure SDK, removing the
+dependency on Databricks mounts / dbutils.
+
+Requires the ``azure`` extra::
+
+    pip install dataorc-utils[azure]
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from azure.identity import DefaultAzureCredential
+from azure.storage.filedatalake import DataLakeServiceClient
+
+logger = logging.getLogger(__name__)
+
+# Type alias for JSON-serializable data (matches LakeFileSystem)
+JSONValue = dict[str, Any] | list[Any] | str | int | float | bool | None
+
+
+class AdlsLakeFileSystem:
+    """ADLS Gen2-backed file operations — drop-in for LakeFileSystem.
+
+    Args:
+        account_url: Full DFS endpoint, e.g.
+            ``"https://<storage_account>.dfs.core.windows.net"``
+        container: File-system / container name, e.g. ``"bronze"``
+        base_path: Optional prefix inside the container prepended to
+            every path.
+        credential: Any Azure credential accepted by the SDK.
+            Defaults to ``DefaultAzureCredential()``.
+
+    Example::
+
+        fs = AdlsLakeFileSystem(
+            account_url="https://testdatadevsc.dfs.core.windows.net",
+            container="bronze",
+            base_path="raw/my_pipeline",
+        )
+        fs.write_json("data.json", {"key": "value"})
+        data = fs.read_json("data.json")
+    """
+
+    def __init__(
+        self,
+        account_url: str,
+        container: str,
+        base_path: str = "",
+        credential: Any | None = None,
+    ):
+        self._credential = credential or DefaultAzureCredential()
+        self._service = DataLakeServiceClient(
+            account_url=account_url,
+            credential=self._credential,
+        )
+        self._fs_client = self._service.get_file_system_client(
+            file_system=container,
+        )
+        self._base_path = base_path.strip("/")
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _resolve(self, path: str) -> str:
+        """Join *path* with the configured base-path prefix."""
+        path = path.lstrip("/")
+        if self._base_path:
+            return f"{self._base_path}/{path}"
+        return path
+
+    # ------------------------------------------------------------------
+    # Text operations
+    # ------------------------------------------------------------------
+
+    def read_text(self, path: str) -> str | None:
+        """Read a UTF-8 text file. Returns ``None`` if the file does not exist."""
+        resolved = self._resolve(path)
+        try:
+            file_client = self._fs_client.get_file_client(resolved)
+            download = file_client.download_file()
+            return download.readall().decode("utf-8")
+        except Exception:
+            logger.debug("Could not read %s", resolved, exc_info=True)
+            return None
+
+    def write_text(self, path: str, content: str) -> None:
+        """Write (or overwrite) a UTF-8 text file.
+
+        Parent "directories" are created implicitly by ADLS Gen2.
+        """
+        resolved = self._resolve(path)
+        file_client = self._fs_client.get_file_client(resolved)
+        file_client.upload_data(content.encode("utf-8"), overwrite=True)
+
+    # ------------------------------------------------------------------
+    # JSON operations
+    # ------------------------------------------------------------------
+
+    def read_json(self, path: str) -> JSONValue:
+        """Read a JSON file. Returns ``None`` if the file does not exist or parsing fails."""
+        content = self.read_text(path)
+        if content is None:
+            return None
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError as exc:
+            logger.warning("Failed to parse JSON from %s: %s", path, exc)
+            return None
+
+    def write_json(self, path: str, data: JSONValue, indent: int = 2) -> None:
+        """Write a JSON file."""
+        self.write_text(path, json.dumps(data, indent=indent, default=str))
+
+    # ------------------------------------------------------------------
+    # Directory / existence helpers
+    # ------------------------------------------------------------------
+
+    def exists(self, path: str) -> bool:
+        """Check whether a file exists."""
+        resolved = self._resolve(path)
+        try:
+            file_client = self._fs_client.get_file_client(resolved)
+            file_client.get_file_properties()
+            return True
+        except Exception:
+            return False
+
+    def delete(self, path: str) -> bool:
+        """Delete a file. Returns ``True`` if deleted, ``False`` otherwise."""
+        resolved = self._resolve(path)
+        try:
+            file_client = self._fs_client.get_file_client(resolved)
+            file_client.delete_file()
+            return True
+        except Exception:
+            return False