Merge dds_ingestion code

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.11", "3.12"]
     steps:
       - uses: actions/checkout@v4
 

README.md

@@ -1,6 +1,16 @@
-# EEADataLakehouseIngestion
+# EEADataLakehouse
 
-Data preparation and validation utilities for the EEA data lakehouse ingestion pipeline.
+Two main areas of concern and class domains for the EEA data lakehouse:
+
+- `eea_datalakehouse.data_preparation` — data acquisition, vocabulary
+  acquisition, data exploration, and transformation to parquet. Each stage is
+  an abstract base class; concrete per-dataset pipelines subclass them since
+  the actual logic varies by dataset and data flow.
+- `eea_datalakehouse.dds_ingestion` — notebook-side client for the Dremio Document
+  Service (DDS) Ingest API (DI-8.4/8.5). `FolderIngest` transfers a folder of
+  data files to S3 (via DDS-issued presigned URLs only — never an S3 SDK or
+  S3 credentials) and registers it as a Dremio table, coordinated entirely
+  through the DDS REST API.
 
 ## Install
 
@@ -13,22 +23,68 @@ pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.0"
 ## Usage
 
 ```python
-from eea_datalakehouse_ingestion import IngestionPipeline, IngestionValidationError
+from pathlib import Path
+
+from eea_datalakehouse.data_preparation import DataValidationError, ParquetTransformer
+
+
+class MyDatasetTransformer(ParquetTransformer):
+    def to_parquet(self, records, destination: Path) -> Path:
+        ...  # dataset-specific parquet write
+
 
-pipeline = IngestionPipeline(required_fields=["id", "value"])
+transformer = MyDatasetTransformer(required_fields=["id", "value"])
 
 raw_source = [{"id": 1, "value": "a"}, {"id": 2, "value": "b"}]
-records = pipeline.prepare(raw_source)
+records = transformer.prepare(raw_source)
 
 try:
-    pipeline.validate(records)
-except IngestionValidationError as exc:
+    transformer.validate(records)
+except DataValidationError as exc:
     print(f"Invalid data: {exc}")
 ```
 
+Dremio credentials are read from the injected kernel env vars `_DREMIO_USER` /
+`_DREMIO_PWD`, and the service URL from `DDS_BASE_URL`. None of these are ever
+logged or printed.
+
+```python
+from eea_datalakehouse.dds_ingestion import FolderIngest
+
+outcome = FolderIngest(
+    folder="./my_data",
+    target_catalog_path="biodiversity.uploads",
+    data_format="parquet",      # one of parquet | csv | json
+    intent="read_only",         # or "editable"
+    conflict_mode="fail",
+    parallelism=4,               # concurrent uploads (default 4)
+).run()
+
+print(outcome.commit.table_path, outcome.commit.record_count)
+```
+
+`run()` performs `begin → upload(all files) → commit`. Re-running the same
+session resumes by skipping files the server reports as already uploaded.
+
 - `prepare(source)` normalizes an iterable of raw records into a list of plain dicts.
 - `validate(records)` checks that every record contains the configured `required_fields`,
-  raising `IngestionValidationError` on the first invalid record.
+  raising `DataValidationError` on the first invalid record.
+- `to_parquet(records, destination)` is where a concrete subclass writes its own dataset's
+  schema out to parquet.
+
+## Layout
+
+| Path | Purpose |
+|---|---|
+| `data_preparation/acquisition.py` | `DataAcquirer` — fetch a dataset's raw source data |
+| `data_preparation/vocabulary.py` | `VocabularyLoader` — load the controlled vocabulary to validate against |
+| `data_preparation/exploration.py` | `Explorer` — inspect acquired data before transformation |
+| `data_preparation/transformation.py` | `ParquetTransformer` — prepare/validate/write to parquet |
+| `dds_ingestion/credentials.py` | env-var creds + redacted `DremioCreds` |
+| `dds_ingestion/models.py` | typed request/response models for the DDS ingest contract |
+| `dds_ingestion/client.py` | thin, unit-testable HTTP client (`IngestClient`) |
+| `dds_ingestion/progress.py` | tqdm progress bar with graceful fallback |
+| `dds_ingestion/folder.py` | `FolderIngest` orchestration (scan/parallel/resume) |
 
 ## Development
 
@@ -64,12 +120,12 @@ first on `PATH`. After installing, restart the kernel (**Kernel > Restart
 Kernel...**) so the import below picks up the newly installed package:
 
 ```python
-from eea_datalakehouse_ingestion import IngestionPipeline
+from eea_datalakehouse.data_preparation import ParquetTransformer
 ```
 
 Alternatively, download the wheel attached to the GitHub Release page for
 that tag and install the local file instead of pulling from git:
 
 ```python
-%pip install /path/to/EEADataLakehouseIngestion-0.1.0-py3-none-any.whl
+%pip install /path/to/EEADataLakehouse-0.1.0-py3-none-any.whl
 ```

pyproject.toml

@@ -3,15 +3,16 @@ requires = ["setuptools>=68"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "EEADataLakehouseIngestion"
+name = "EEADataLakehouse"
 version = "0.1.0"
 description = "Data preparation and validation utilities for the EEA data lakehouse ingestion pipeline."
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
+license = { text = "EUPL-1.2" }
 authors = [{ name = "EEA Data Hub" }]
 dependencies = [
-    #"pandas>=2.2,<3",
-    # "pyarrow>=17,<22",         # parquet writer + the Arrow types the ingest lane expects
+    "pandas>=3",
+    "pyarrow>=25",               # parquet writer + the Arrow types the ingest lane expects
     "openpyxl>=3.1,<4",          # .xlsx reader — handles sparse cells correctly (see bathing_water/README.md)
     "xlrd>=2.0,<3",              # legacy .xls, some older SDI deliveries
     "lxml>=5.2,<7",              # ISO 19115-3 metadata records
@@ -29,10 +30,38 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-dev = ["pytest>=7"]
+dev = [
+    "pytest>=7",
+    "ruff>=0.15",
+    "mypy>=2.1",
+    "respx>=0.23",       # mocks httpx for the ingestion test suite
+    "types-tqdm",
+]
 
 [project.urls]
 Repository = "https://github.com/eeadata/EEALakeHouse.python"
 
 [tool.setuptools.packages.find]
 where = ["src"]
+
+[tool.ruff]
+target-version = "py311"
+line-length = 100
+src = ["src", "tests"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "C4", "SIM"]
+
+[tool.mypy]
+python_version = "3.11"
+strict = true
+warn_return_any = false
+
+[[tool.mypy.overrides]]
+module = ["respx.*"]
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-q"
+filterwarnings = ["ignore::DeprecationWarning"]

requirements.txt

@@ -1,5 +1,5 @@
-#pandas>=2.2,<3
-#pyarrow>=17                 # parquet writer + the Arrow types the ingest lane expects
+pandas>=3
+pyarrow>=25                 # parquet writer + the Arrow types the ingest lane expects
 openpyxl>=3.1,<4            # .xlsx reader — handles sparse cells correctly (see bathing_water/README.md)
 xlrd>=2.0,<3                # legacy .xls, some older SDI deliveries
 lxml>=5.2,<7                # ISO 19115-3 metadata records
@@ -15,11 +15,6 @@ pyproj>=3.6,<4              # only where a reprojection is a deliberate, documen
 
 # --- acquire ---------------------------------------------------------------
 requests>=2.32,<3
-#httpx>=0.28,<1              # also the dds_ingest dependency
-#beautifulsoup4>=4.12,<5     # SDI "direct download" is an HTML landing page, not a file
-#tqdm>=4.66,<5
-
-# --- build backend for the mounted dds_ingest source -----------------------
-# Present so the entrypoint can install /opt/dds_ingest offline, with
-# --no-build-isolation, without pip reaching out to PyPI on every start.
-hatchling>=1.25,<2
+httpx>=0.28,<1              # also the ingestion (DDS) client dependency
+beautifulsoup4>=4.12,<5     # SDI "direct download" is an HTML landing page, not a file
+tqdm>=4.66,<5

src/eea_datalakehouse/__init__.py

@@ -0,0 +1,3 @@
+__version__ = "0.1.0"
+
+__all__ = ["__version__"]

src/eea_datalakehouse/data_preparation/__init__.py

@@ -0,0 +1,12 @@
+from .acquisition import DataAcquirer
+from .exploration import Explorer
+from .transformation import DataValidationError, ParquetTransformer
+from .vocabulary import VocabularyLoader
+
+__all__ = [
+    "DataAcquirer",
+    "VocabularyLoader",
+    "Explorer",
+    "ParquetTransformer",
+    "DataValidationError",
+]

src/eea_datalakehouse/data_preparation/acquisition.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class DataAcquirer(ABC):
+    """Base class for a dataset's data-acquisition stage.
+
+    A concrete subclass knows how to fetch one dataset's raw source data (an
+    SDI download, an API, a file share, ...) and hands back whatever raw form
+    is most convenient for that dataset's exploration/transformation steps to
+    consume next.
+    """
+
+    @abstractmethod
+    def acquire(self) -> Any:
+        """Fetch and return the dataset's raw source data."""

src/eea_datalakehouse/data_preparation/exploration.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Explorer(ABC):
+    """Base class for exploring acquired data before it is transformed."""
+
+    @abstractmethod
+    def explore(self, data: Any) -> Any:
+        """Inspect ``data`` and return exploration results (profile, schema, summary, ...)."""

src/eea_datalakehouse_ingestion/pipeline.py → src/eea_datalakehouse/data_preparation/transformation.py

@@ -1,14 +1,20 @@
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import Any, Iterable, Mapping, Sequence
 
 
-class IngestionValidationError(Exception):
-    """Raised when a record fails validation before ingestion."""
+class DataValidationError(Exception):
+    """Raised when a record fails validation before being transformed to parquet."""
 
 
-class IngestionPipeline:
-    """Stages and validates records before they are loaded into the lakehouse.
+class ParquetTransformer(ABC):
+    """Base class for a dataset's transform-to-parquet stage.
+
+    ``prepare`` and ``validate`` are dataset-agnostic staging steps shared by
+    every flow; ``to_parquet`` is where a concrete subclass encodes the
+    schema and write logic specific to its own dataset.
 
     Parameters
     ----------
@@ -19,10 +25,6 @@ class IngestionPipeline:
     def __init__(self, required_fields: Sequence[str] = ()) -> None:
         self.required_fields = tuple(required_fields)
 
-    def hello_world(self) -> str:
-        """Return a greeting, useful for a quick install sanity check."""
-        return "Hello, world!"
-
     def prepare(self, source: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
         """Normalize an iterable of raw records into a list of plain dicts."""
         return [dict(record) for record in source]
@@ -31,12 +33,16 @@ def validate(self, records: Sequence[Mapping[str, Any]]) -> bool:
         """Check that every record contains all required fields.
 
         Returns ``True`` when every record is valid, otherwise raises
-        ``IngestionValidationError`` describing the first invalid record.
+        ``DataValidationError`` describing the first invalid record.
         """
         for index, record in enumerate(records):
             missing = [field for field in self.required_fields if field not in record]
             if missing:
-                raise IngestionValidationError(
+                raise DataValidationError(
                     f"record {index} is missing required fields: {missing}"
                 )
         return True
+
+    @abstractmethod
+    def to_parquet(self, records: Sequence[Mapping[str, Any]], destination: Path) -> Path:
+        """Write validated records to a parquet file at ``destination``."""

src/eea_datalakehouse/data_preparation/vocabulary.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class VocabularyLoader(ABC):
+    """Base class for loading the controlled vocabulary a dataset validates against."""
+
+    @abstractmethod
+    def load_vocabulary(self) -> Any:
+        """Fetch and return the dataset's controlled vocabulary or reference data."""

src/eea_datalakehouse/dds_ingestion/__init__.py

@@ -0,0 +1,72 @@
+"""Ingestion into DDS (Dremio Document Service), notebook-side (DI-8.4/8.5).
+
+A generated Jupyter notebook imports :class:`FolderIngest` to transfer a folder
+of data files to S3 (via DDS-issued presigned URLs only) and register it as a
+Dremio table, coordinated entirely through the DDS REST API.
+
+Typical use inside a notebook (creds + base URL come from the kernel env)::
+
+    from eea_datalakehouse.dds_ingestion import FolderIngest
+
+    outcome = FolderIngest(
+        folder="./my_data",
+        target_catalog_path="biodiversity.uploads",
+        data_format="parquet",
+        intent="read_only",
+        parallelism=4,
+    ).run()
+    print(outcome.commit.table_path, outcome.commit.record_count)
+"""
+
+from __future__ import annotations
+
+from .client import IngestApiError, IngestClient
+from .credentials import (
+    DremioCreds,
+    MissingCredentialsError,
+    load_base_url,
+    load_creds,
+)
+from .folder import (
+    DEFAULT_PARALLELISM,
+    FolderIngest,
+    IngestOutcome,
+    ingest_folder,
+    scan_folder,
+)
+from .models import (
+    BeginResult,
+    CommitResult,
+    DataFormat,
+    FileSpec,
+    Intent,
+    Progress,
+    S3Plan,
+    StatusResult,
+    UploadPart,
+    UploadTarget,
+)
+
+__all__ = [
+    "DEFAULT_PARALLELISM",
+    "BeginResult",
+    "CommitResult",
+    "DataFormat",
+    "DremioCreds",
+    "FileSpec",
+    "FolderIngest",
+    "IngestApiError",
+    "IngestClient",
+    "IngestOutcome",
+    "Intent",
+    "MissingCredentialsError",
+    "Progress",
+    "S3Plan",
+    "StatusResult",
+    "UploadPart",
+    "UploadTarget",
+    "ingest_folder",
+    "load_base_url",
+    "load_creds",
+    "scan_folder",
+]