diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5ffbb5d..f229cf1 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/README.md b/README.md index a6e8f7a..949f2d9 100644 --- a/README.md +++ b/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 ``` diff --git a/pyproject.toml b/pyproject.toml index 15ba9dc..931b677 100644 --- a/pyproject.toml +++ b/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"] diff --git a/requirements.txt b/requirements.txt index ac6dcba..10b0b32 100644 --- a/requirements.txt +++ b/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 diff --git a/src/eea_datalakehouse/__init__.py b/src/eea_datalakehouse/__init__.py new file mode 100644 index 0000000..de48b44 --- /dev/null +++ b/src/eea_datalakehouse/__init__.py @@ -0,0 +1,3 @@ +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/src/eea_datalakehouse/data_preparation/__init__.py b/src/eea_datalakehouse/data_preparation/__init__.py new file mode 100644 index 0000000..e20f658 --- /dev/null +++ b/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", +] diff --git a/src/eea_datalakehouse/data_preparation/acquisition.py b/src/eea_datalakehouse/data_preparation/acquisition.py new file mode 100644 index 0000000..b1c6981 --- /dev/null +++ b/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.""" diff --git a/src/eea_datalakehouse/data_preparation/exploration.py b/src/eea_datalakehouse/data_preparation/exploration.py new file mode 100644 index 0000000..057f69d --- /dev/null +++ b/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, ...).""" diff --git a/src/eea_datalakehouse_ingestion/pipeline.py b/src/eea_datalakehouse/data_preparation/transformation.py similarity index 57% rename from src/eea_datalakehouse_ingestion/pipeline.py rename to src/eea_datalakehouse/data_preparation/transformation.py index b82a672..b56108c 100644 --- a/src/eea_datalakehouse_ingestion/pipeline.py +++ b/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``.""" diff --git a/src/eea_datalakehouse/data_preparation/vocabulary.py b/src/eea_datalakehouse/data_preparation/vocabulary.py new file mode 100644 index 0000000..4a32c97 --- /dev/null +++ b/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.""" diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py new file mode 100644 index 0000000..6a57109 --- /dev/null +++ b/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", +] diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py new file mode 100644 index 0000000..ad5bde4 --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -0,0 +1,239 @@ +"""Thin, unit-testable HTTP client for the DDS Ingest API (v0.1). + +This layer knows nothing about local folders, parallelism or progress bars: it +only translates the four ingest endpoints to/from the typed models in +:mod:`eea_datalakehouse.dds_ingestion.models`. The orchestration logic lives in +:mod:`eea_datalakehouse.dds_ingestion.folder`. + +Authentication uses a Bearer token (the Dremio PAT, which is the kernel's +``_DREMIO_PWD``) sent **only on the DDS API calls** — never on the presigned S3 +upload, which carries its own auth in the URL. The token is held in a private +header dict and is never logged by this module. +""" + +from __future__ import annotations + +from types import TracebackType +from typing import Any + +import httpx + +from .credentials import DremioCreds +from .models import ( + BeginResult, + CommitResult, + DataFormat, + FileSpec, + Intent, + StatusResult, + UploadTarget, +) + +DEFAULT_TIMEOUT = 60.0 + + +class IngestApiError(RuntimeError): + """A DDS ingest API call returned a non-success status.""" + + def __init__(self, status_code: int, message: str) -> None: + super().__init__(f"DDS ingest API error {status_code}: {message}") + self.status_code = status_code + self.message = message + + +class IngestClient: + """Client for the ``/api/v1/ingest/*`` endpoints. + + The same :class:`httpx.Client` is reused for the presigned uploads so we + avoid creating a fresh connection pool per file. + """ + + def __init__( + self, + base_url: str, + creds: DremioCreds, + *, + timeout: float = DEFAULT_TIMEOUT, + http_client: httpx.Client | None = None, + ) -> None: + self._base_url = base_url.rstrip("/") + self._owns_client = http_client is None + self._http = http_client or httpx.Client(timeout=timeout) + # Bearer PAT, applied per-request to DDS calls only (not the S3 upload). + self._auth_headers = {"Authorization": f"Bearer {creds.password}"} + + # -- lifecycle -------------------------------------------------------- + + def close(self) -> None: + if self._owns_client: + self._http.close() + + def __enter__(self) -> IngestClient: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + self.close() + + # -- API endpoints ---------------------------------------------------- + + def begin( + self, + *, + target_catalog_path: str, + intent: Intent, + data_format: DataFormat, + conflict_mode: str, + files: list[FileSpec], + table_name: str | None = None, + idempotency_key: str | None = None, + multipart: bool | None = None, + ) -> BeginResult: + body: dict[str, Any] = { + "target_catalog_path": target_catalog_path, + "intent": intent, + "format": data_format, + "conflict_mode": conflict_mode, + "runner": "notebook", + "files": [f.as_payload() for f in files], + } + if table_name is not None: + body["table_name"] = table_name + if idempotency_key is not None: + body["idempotency_key"] = idempotency_key + if multipart is not None: + body["multipart"] = multipart + data = self._post("/api/v1/ingest/begin", body) + return BeginResult.from_json(data) + + def commit( + self, + *, + session_id: str, + definition: dict[str, Any] | None = None, + multipart_etags: list[dict[str, Any]] | None = None, + ) -> CommitResult: + body: dict[str, Any] = {"session_id": session_id} + if definition is not None: + body["definition"] = definition + if multipart_etags: + # DI-7.2 commit shape: [{rel_path, upload_id, parts:[{part_number, etag}]}] + body["multipart_etags"] = multipart_etags + data = self._post("/api/v1/ingest/commit", body) + return CommitResult.from_json(data) + + def get_status(self, session_id: str) -> StatusResult: + resp = self._http.get( + f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers + ) + self._raise_for_status(resp) + return StatusResult.from_json(resp.json()) + + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: + """Upload one file's bytes to its presigned target. + + For a pre-signed **POST policy** (the default, ``method == "POST"``) this + submits the form ``fields`` plus a ``file`` part as multipart/form-data; + for a legacy pre-signed PUT it streams the body. Returns the ETag header + if supplied, else ``None``; raises :class:`IngestApiError` on non-2xx. + + Multipart targets (DI-7) are handled by :meth:`upload_file_multipart`, + which returns the per-part ETags; this method covers the small-file path. + + This is the *only* path that touches object storage, and it uses the + presigned URL/fields exactly as issued — no S3 SDK or credentials. + """ + + if target.max_bytes is not None and len(data) > target.max_bytes: + raise IngestApiError( + 413, + f"file {target.rel_path!r} ({len(data)} bytes) exceeds " + f"max_bytes ({target.max_bytes})", + ) + if target.method.upper() == "POST": + # Pre-signed POST policy: form fields first, then the `file` part + # (file MUST be last so S3 honours the policy conditions). + resp = self._http.post( + target.url, + data=target.fields, + files={"file": (target.rel_path, data)}, + ) + else: # legacy pre-signed PUT + resp = self._http.request( + target.method, target.url, content=data, headers=target.headers + ) + self._raise_for_status(resp) + return resp.headers.get("ETag") + + def upload_file_multipart( + self, target: UploadTarget, data: bytes + ) -> list[dict[str, Any]]: + """Upload one large file as S3 multipart parts (DI-7.3). + + Splits ``data`` into ``len(target.parts)`` chunks and ``PUT``s each chunk + to its presigned part URL, collecting the ETag S3 returns per part. The + chunk size is derived from the part count so the parts tile the bytes + exactly (the last part takes the remainder). Returns the commit-shaped + ``[{"part_number", "etag"}, ...]`` for the file's ``multipart_etags``. + + As with :meth:`upload_file`, this is the only object-storage path and it + carries no DDS credentials — each part URL is self-authenticating. + """ + if not target.is_multipart: + raise IngestApiError( + 400, f"target {target.rel_path!r} is not a multipart upload" + ) + if target.max_bytes is not None and len(data) > target.max_bytes: + raise IngestApiError( + 413, + f"file {target.rel_path!r} ({len(data)} bytes) exceeds " + f"max_bytes ({target.max_bytes})", + ) + num_parts = len(target.parts) + # Ceil division so num_parts chunks cover all bytes; the final chunk is + # whatever remains. An empty file still yields one (empty) part. + chunk = max(1, -(-len(data) // num_parts)) if data else 0 + etags: list[dict[str, Any]] = [] + for index, part in enumerate( + sorted(target.parts, key=lambda p: p.part_number) + ): + start = index * chunk + body = data[start : start + chunk] if chunk else b"" + resp = self._http.put(part.url, content=body) + self._raise_for_status(resp) + etag = resp.headers.get("ETag") + if etag is None: + raise IngestApiError( + 502, + f"S3 returned no ETag for part {part.part_number} " + f"of {target.rel_path!r}", + ) + etags.append({"part_number": part.part_number, "etag": etag}) + return etags + + # -- internals -------------------------------------------------------- + + def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: + resp = self._http.post( + f"{self._base_url}{path}", json=body, headers=self._auth_headers + ) + self._raise_for_status(resp) + result: dict[str, Any] = resp.json() + return result + + @staticmethod + def _raise_for_status(resp: httpx.Response) -> None: + if resp.is_success: + return + message = resp.text + try: + payload = resp.json() + if isinstance(payload, dict): + message = str(payload.get("message") or payload.get("error") or message) + except ValueError: + pass + raise IngestApiError(resp.status_code, message) diff --git a/src/eea_datalakehouse/dds_ingestion/credentials.py b/src/eea_datalakehouse/dds_ingestion/credentials.py new file mode 100644 index 0000000..dcc5f57 --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/credentials.py @@ -0,0 +1,76 @@ +"""Credential resolution for the ingest client. + +Dremio credentials are read from the kernel environment variables injected by +the JupyterLab extension (``_DREMIO_USER`` / ``_DREMIO_PWD``). The DDS base URL +is read from ``DDS_BASE_URL``. + +The :class:`DremioCreds` object deliberately hides its secret from ``repr``, +``str`` and logging so credentials never leak into notebook output or logs. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + +ENV_USER = "_DREMIO_USER" +ENV_PWD = "_DREMIO_PWD" # noqa: S105 — env var name, not a secret value +ENV_BASE_URL = "DDS_BASE_URL" + + +class MissingCredentialsError(RuntimeError): + """Raised when the required credentials or base URL are absent.""" + + +@dataclass(frozen=True) +class DremioCreds: + """Dremio username + PAT. + + ``_password`` is the Dremio PAT (the kernel's ``_DREMIO_PWD``); it is sent as + the ``Authorization: Bearer`` token. It is stored but never rendered: + ``repr``/``str`` redact it so it cannot reach notebook output, tracebacks or + log handlers. + """ + + username: str + _password: str + + @property + def password(self) -> str: + return self._password + + def __repr__(self) -> str: + return f"DremioCreds(username={self.username!r}, password=***)" + + __str__ = __repr__ + + +def load_creds(env: dict[str, str] | None = None) -> DremioCreds: + """Load Dremio credentials from the kernel environment. + + Parameters + ---------- + env: + Mapping to read from; defaults to :data:`os.environ`. Passing an + explicit mapping keeps this testable without mutating process state. + """ + + source = os.environ if env is None else env + user = source.get(ENV_USER) + pwd = source.get(ENV_PWD) + if not user or not pwd: + raise MissingCredentialsError( + f"missing Dremio credentials in environment " + f"({ENV_USER}/{ENV_PWD} must both be set)" + ) + return DremioCreds(username=user, _password=pwd) + + +def load_base_url(env: dict[str, str] | None = None) -> str: + """Load the DDS base URL from the environment.""" + + source = os.environ if env is None else env + base = source.get(ENV_BASE_URL) + if not base: + raise MissingCredentialsError(f"missing DDS base URL ({ENV_BASE_URL} must be set)") + return base.rstrip("/") diff --git a/src/eea_datalakehouse/dds_ingestion/folder.py b/src/eea_datalakehouse/dds_ingestion/folder.py new file mode 100644 index 0000000..2f71dad --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/folder.py @@ -0,0 +1,252 @@ +"""Folder-level ingest orchestration (DI-8.4 / DI-8.5). + +:class:`FolderIngest` drives the full ``begin → upload(all files) → commit`` +flow for a local folder: + +* recurse the folder and keep exactly one data format (single-format scan); +* upload all files to their presigned URLs with configurable parallelism + (default 4); +* show a progress bar (tqdm, degrading gracefully if absent); +* resume — re-running skips files already uploaded for the session. + +The class depends only on :class:`~eea_datalakehouse.dds_ingestion.client.IngestClient`, +so the HTTP layer can be mocked or swapped in tests. +""" + +from __future__ import annotations + +import logging +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path + +from .client import IngestClient +from .credentials import DremioCreds, load_base_url, load_creds +from .models import ( + BeginResult, + CommitResult, + DataFormat, + FileSpec, + Intent, + UploadTarget, +) +from .progress import make_progress_bar + +logger = logging.getLogger("eea_datalakehouse.dds_ingestion") + +DEFAULT_PARALLELISM = 4 +_FORMAT_EXTENSIONS: dict[DataFormat, tuple[str, ...]] = { + "parquet": (".parquet",), + "csv": (".csv",), + "json": (".json", ".ndjson"), +} + + +@dataclass(slots=True) +class IngestOutcome: + """Summary returned by :meth:`FolderIngest.run`.""" + + begin: BeginResult + commit: CommitResult + files_uploaded: int + files_skipped: int + etags: dict[str, str] = field(default_factory=dict) + + +def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: + """Recurse ``folder`` and return files matching ``data_format`` only. + + Other formats present in the folder are ignored (single-format scan). Paths + are returned folder-relative with forward slashes so they are stable across + platforms. Results are sorted for deterministic ordering. + """ + + exts = _FORMAT_EXTENSIONS[data_format] + specs: list[FileSpec] = [] + for path in sorted(folder.rglob("*")): + if not path.is_file(): + continue + if path.suffix.lower() not in exts: + continue + rel = path.relative_to(folder).as_posix() + specs.append(FileSpec(rel_path=rel, size=path.stat().st_size)) + return specs + + +class FolderIngest: + """Orchestrate ingest of a local folder into a Dremio table via DDS.""" + + def __init__( + self, + folder: str | Path, + target_catalog_path: str, + *, + data_format: DataFormat, + intent: Intent = "read_only", + conflict_mode: str = "fail", + table_name: str | None = None, + parallelism: int = DEFAULT_PARALLELISM, + idempotency_key: str | None = None, + multipart: bool | None = None, + show_progress: bool = True, + client: IngestClient | None = None, + base_url: str | None = None, + creds: DremioCreds | None = None, + ) -> None: + if parallelism < 1: + raise ValueError("parallelism must be >= 1") + self.folder = Path(folder) + if not self.folder.is_dir(): + raise NotADirectoryError(f"not a directory: {self.folder}") + self.target_catalog_path = target_catalog_path + self.data_format = data_format + self.intent = intent + self.conflict_mode = conflict_mode + self.table_name = table_name + self.parallelism = parallelism + self.idempotency_key = idempotency_key + self.multipart = multipart + self.show_progress = show_progress + + # The client owns the credentials; if the caller did not inject one we + # build it from the kernel environment. Creds never leave the client. + if client is not None: + self._client = client + self._owns_client = False + else: + resolved_url = base_url or load_base_url() + resolved_creds = creds or load_creds() + self._client = IngestClient(resolved_url, resolved_creds) + self._owns_client = True + + # -- orchestration ---------------------------------------------------- + + def run(self) -> IngestOutcome: + """Execute the full begin → upload → commit flow.""" + + try: + files = scan_folder(self.folder, self.data_format) + if not files: + raise FileNotFoundError( + f"no {self.data_format} files found under {self.folder}" + ) + begin = self._client.begin( + target_catalog_path=self.target_catalog_path, + intent=self.intent, + data_format=self.data_format, + conflict_mode=self.conflict_mode, + files=files, + table_name=self.table_name, + idempotency_key=self.idempotency_key, + multipart=self.multipart, + ) + etags, multipart_etags, uploaded, skipped = self._upload_all(begin) + commit = self._client.commit( + session_id=begin.session_id, + multipart_etags=multipart_etags or None, + ) + return IngestOutcome( + begin=begin, + commit=commit, + files_uploaded=uploaded, + files_skipped=skipped, + etags=etags, + ) + finally: + if self._owns_client: + self._client.close() + + # -- upload phase ----------------------------------------------------- + + def _already_done(self, session_id: str) -> set[str]: + """Return rel_paths already uploaded for this session (resume support). + + Queries the session status; the server reports ``files_done`` and may + list completed rel_paths under ``raw["uploaded"]``. We only skip files + the server explicitly names, so resume never wrongly drops a file. + """ + + try: + status = self._client.get_status(session_id) + except Exception: # noqa: BLE001 — status is best-effort for resume + logger.debug("could not fetch status for resume; uploading all files") + return set() + done = status.raw.get("uploaded") + if isinstance(done, list): + return {str(p) for p in done} + return set() + + def _upload_all( + self, begin: BeginResult + ) -> tuple[dict[str, str], list[dict[str, object]], int, int]: + targets = list(begin.s3.uploads) + done = self._already_done(begin.session_id) + pending = [t for t in targets if t.rel_path not in done] + skipped = len(targets) - len(pending) + + # Single-shot ETags (rel_path -> ETag) and multipart commit entries + # ({rel_path, upload_id, parts:[{part_number, etag}]}) are collected + # separately: only the latter is sent to commit as ``multipart_etags``. + etags: dict[str, str] = {} + multipart_etags: list[dict[str, object]] = [] + bar = make_progress_bar(len(targets), desc="Uploading") if self.show_progress else None + if bar is not None and skipped: + bar.update(skipped) + + try: + if not pending: + return etags, multipart_etags, 0, skipped + with ThreadPoolExecutor(max_workers=self.parallelism) as pool: + futures = {pool.submit(self._upload_one, t): t for t in pending} + for future in as_completed(futures): + target = futures[future] + etag, parts = future.result() + if parts is not None: + multipart_etags.append( + { + "rel_path": target.rel_path, + "upload_id": target.upload_id, + "parts": parts, + } + ) + elif etag is not None: + etags[target.rel_path] = etag + if bar is not None: + bar.update(1) + finally: + if bar is not None: + bar.close() + + return etags, multipart_etags, len(pending), skipped + + def _upload_one( + self, target: UploadTarget + ) -> tuple[str | None, list[dict[str, object]] | None]: + """Upload one file; return ``(single_etag, multipart_parts)``. + + Exactly one element is non-``None``: a single-shot upload yields the ETag + (or ``None`` if S3 omitted it); a multipart upload yields the per-part + ETag list and the single ETag is ``None``. + """ + data = (self.folder / target.rel_path).read_bytes() + if target.is_multipart: + parts = self._client.upload_file_multipart(target, data) + return None, [dict(p) for p in parts] + return self._client.upload_file(target, data), None + + +def ingest_folder( + folder: str | Path, + target_catalog_path: str, + *, + data_format: DataFormat, + **kwargs: object, +) -> IngestOutcome: + """Convenience wrapper: build a :class:`FolderIngest` and run it.""" + + return FolderIngest( + folder, + target_catalog_path, + data_format=data_format, + **kwargs, # type: ignore[arg-type] + ).run() diff --git a/src/eea_datalakehouse/dds_ingestion/models.py b/src/eea_datalakehouse/dds_ingestion/models.py new file mode 100644 index 0000000..1ed1da9 --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/models.py @@ -0,0 +1,162 @@ +"""Typed models for the DDS Ingest API contract (v0.1). + +These mirror the request/response shapes of the ``/api/v1/ingest/*`` endpoints. +They are intentionally permissive on parsing (extra fields are ignored) so the +client keeps working if the server adds new keys. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Literal + +Intent = Literal["read_only", "editable"] +DataFormat = Literal["parquet", "csv", "json"] + + +@dataclass(frozen=True, slots=True) +class FileSpec: + """A local file scheduled for upload, identified by its folder-relative path.""" + + rel_path: str + size: int + + def as_payload(self) -> dict[str, Any]: + return {"rel_path": self.rel_path, "size": self.size} + + +@dataclass(frozen=True, slots=True) +class UploadPart: + """One pre-signed part URL of a multipart upload target (DI-7.3).""" + + part_number: int + url: str + + @classmethod + def from_json(cls, data: dict[str, Any]) -> UploadPart: + return cls(part_number=int(data["part_number"]), url=data["url"]) + + +@dataclass(frozen=True, slots=True) +class UploadTarget: + """A presigned-upload instruction returned by ``begin`` for one file. + + A *single-shot* target carries a POST ``url`` + ``fields`` (DI-2). A + *multipart* target instead carries an ``upload_id`` and per-part presigned + URLs in ``parts`` (DI-7); ``method`` is then ``"PUT"`` and ``url`` is empty. + """ + + rel_path: str + url: str = "" + method: str = "POST" + # Pre-signed POST policy form fields, submitted alongside the file part. + fields: dict[str, str] = field(default_factory=dict) + headers: dict[str, str] = field(default_factory=dict) + max_bytes: int | None = None + # Multipart (DI-7): present only when the server chose a multipart upload. + upload_id: str | None = None + parts: tuple[UploadPart, ...] = () + + @property + def is_multipart(self) -> bool: + return self.upload_id is not None and bool(self.parts) + + @classmethod + def from_json(cls, data: dict[str, Any]) -> UploadTarget: + return cls( + rel_path=data["rel_path"], + url=data.get("url", ""), + method=data.get("method", "POST"), + fields=dict(data.get("fields") or {}), + headers=dict(data.get("headers") or {}), + max_bytes=data.get("max_bytes"), + upload_id=data.get("upload_id"), + parts=tuple(UploadPart.from_json(p) for p in data.get("parts") or ()), + ) + + +@dataclass(frozen=True, slots=True) +class S3Plan: + """The S3 upload plan: bucket, key prefix and per-file presigned targets.""" + + bucket: str + key_prefix: str + uploads: tuple[UploadTarget, ...] + + @classmethod + def from_json(cls, data: dict[str, Any]) -> S3Plan: + return cls( + bucket=data["bucket"], + key_prefix=data["key_prefix"], + uploads=tuple(UploadTarget.from_json(u) for u in data.get("uploads", [])), + ) + + +@dataclass(frozen=True, slots=True) +class BeginResult: + """Response from ``POST /api/v1/ingest/begin``.""" + + session_id: str + status: str + s3: S3Plan + collision: Any = None + + @classmethod + def from_json(cls, data: dict[str, Any]) -> BeginResult: + return cls( + session_id=data["session_id"], + status=data["status"], + s3=S3Plan.from_json(data["s3"]), + collision=data.get("collision"), + ) + + +@dataclass(frozen=True, slots=True) +class CommitResult: + """Response from ``POST /api/v1/ingest/commit``.""" + + session_id: str + status: str + table_path: str | None = None + record_count: int | None = None + + @classmethod + def from_json(cls, data: dict[str, Any]) -> CommitResult: + return cls( + session_id=data["session_id"], + status=data["status"], + table_path=data.get("table_path"), + record_count=data.get("record_count"), + ) + + +@dataclass(frozen=True, slots=True) +class Progress: + files_done: int + files_total: int + step: str + + @classmethod + def from_json(cls, data: dict[str, Any]) -> Progress: + return cls( + files_done=int(data.get("files_done", 0)), + files_total=int(data.get("files_total", 0)), + step=str(data.get("step", "")), + ) + + +@dataclass(frozen=True, slots=True) +class StatusResult: + """Response from ``GET /api/v1/ingest/{session_id}``.""" + + status: str + progress: Progress + raw: dict[str, Any] + + @classmethod + def from_json(cls, data: dict[str, Any]) -> StatusResult: + return cls( + status=data["status"], + progress=Progress.from_json(data.get("progress") or {}), + raw=data, + ) diff --git a/src/eea_datalakehouse/dds_ingestion/progress.py b/src/eea_datalakehouse/dds_ingestion/progress.py new file mode 100644 index 0000000..a1535cd --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/progress.py @@ -0,0 +1,44 @@ +"""Progress-bar abstraction that degrades gracefully without ``tqdm``. + +If ``tqdm`` is installed a real bar is shown; otherwise we fall back to a tiny +no-frills reporter that prints occasional line updates. Either way the public +surface is the same: ``update(n)`` and ``close()``. +""" + +from __future__ import annotations + +from typing import Protocol + + +class ProgressBar(Protocol): + def update(self, n: int = 1) -> object: ... + + def close(self) -> None: ... + + +class _NullProgress: + """Fallback used when tqdm is unavailable; prints sparse text updates.""" + + def __init__(self, total: int, desc: str) -> None: + self._total = total + self._desc = desc + self._done = 0 + print(f"{desc}: 0/{total}") + + def update(self, n: int = 1) -> object: + self._done += n + print(f"{self._desc}: {self._done}/{self._total}") + return None + + def close(self) -> None: + return None + + +def make_progress_bar(total: int, desc: str = "Uploading") -> ProgressBar: + """Return a tqdm bar if available, else a minimal text reporter.""" + + try: + from tqdm.auto import tqdm + except ImportError: + return _NullProgress(total, desc) + return tqdm(total=total, desc=desc, unit="file") diff --git a/src/eea_datalakehouse/dds_ingestion/py.typed b/src/eea_datalakehouse/dds_ingestion/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/eea_datalakehouse_ingestion/__init__.py b/src/eea_datalakehouse_ingestion/__init__.py deleted file mode 100644 index ac6525e..0000000 --- a/src/eea_datalakehouse_ingestion/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -from .pipeline import IngestionPipeline, IngestionValidationError - -__version__ = "0.1.0" - -__all__ = ["IngestionPipeline", "IngestionValidationError", "__version__"] diff --git a/tests/data_preparation/test_stage_bases.py b/tests/data_preparation/test_stage_bases.py new file mode 100644 index 0000000..e08f59c --- /dev/null +++ b/tests/data_preparation/test_stage_bases.py @@ -0,0 +1,9 @@ +import pytest + +from eea_datalakehouse.data_preparation import DataAcquirer, Explorer, VocabularyLoader + + +@pytest.mark.parametrize("base", [DataAcquirer, VocabularyLoader, Explorer]) +def test_stage_base_cannot_be_instantiated_directly(base): + with pytest.raises(TypeError): + base() diff --git a/tests/data_preparation/test_transformation.py b/tests/data_preparation/test_transformation.py new file mode 100644 index 0000000..d05950f --- /dev/null +++ b/tests/data_preparation/test_transformation.py @@ -0,0 +1,40 @@ +from pathlib import Path + +import pytest + +from eea_datalakehouse.data_preparation import DataValidationError, ParquetTransformer + + +class _StubTransformer(ParquetTransformer): + def to_parquet(self, records, destination: Path) -> Path: + return destination + + +def test_prepare_normalizes_records_into_plain_dicts(): + transformer = _StubTransformer() + source = ({"id": 1, "value": "a"}, {"id": 2, "value": "b"}) + + result = transformer.prepare(source) + + assert result == [{"id": 1, "value": "a"}, {"id": 2, "value": "b"}] + assert all(type(record) is dict for record in result) + + +def test_validate_passes_when_required_fields_present(): + transformer = _StubTransformer(required_fields=["id", "value"]) + records = transformer.prepare([{"id": 1, "value": "a"}]) + + assert transformer.validate(records) is True + + +def test_validate_raises_on_missing_required_field(): + transformer = _StubTransformer(required_fields=["id", "value"]) + records = transformer.prepare([{"id": 1}]) + + with pytest.raises(DataValidationError, match="record 0"): + transformer.validate(records) + + +def test_parquet_transformer_cannot_be_instantiated_directly(): + with pytest.raises(TypeError): + ParquetTransformer() diff --git a/tests/dds_ingestion/__init__.py b/tests/dds_ingestion/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/dds_ingestion/conftest.py b/tests/dds_ingestion/conftest.py new file mode 100644 index 0000000..be2a09f --- /dev/null +++ b/tests/dds_ingestion/conftest.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +from eea_datalakehouse.dds_ingestion.credentials import DremioCreds + +BASE_URL = "https://dds.example.test" + + +@pytest.fixture +def creds() -> DremioCreds: + return DremioCreds(username="alice", _password="s3cr3t-pwd") + + +@pytest.fixture +def data_folder(tmp_path: Path) -> Path: + """A folder with two parquet files plus noise of other formats.""" + + (tmp_path / "a.parquet").write_bytes(b"PAR1-a") + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.parquet").write_bytes(b"PAR1-bb") + # noise that the single-format scan must ignore: + (tmp_path / "notes.csv").write_text("x,y\n1,2\n") + (tmp_path / "meta.json").write_text("{}") + (tmp_path / "readme.txt").write_text("ignore me") + return tmp_path diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py new file mode 100644 index 0000000..a17c367 --- /dev/null +++ b/tests/dds_ingestion/test_client.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import httpx +import pytest +import respx +from eea_datalakehouse.dds_ingestion.client import IngestApiError, IngestClient +from eea_datalakehouse.dds_ingestion.credentials import DremioCreds +from eea_datalakehouse.dds_ingestion.models import FileSpec, UploadPart, UploadTarget + +from .conftest import BASE_URL + + +@respx.mock +def test_begin_sends_contract_body_and_bearer_auth(creds: DremioCreds) -> None: + route = respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response( + 200, + json={ + "session_id": "sess-1", + "status": "open", + "s3": {"bucket": "b", "key_prefix": "p/", "uploads": []}, + "collision": None, + }, + ) + ) + with IngestClient(BASE_URL, creds) as client: + result = client.begin( + target_catalog_path="bio.uploads", + intent="read_only", + data_format="parquet", + conflict_mode="fail", + files=[FileSpec("a.parquet", 6)], + table_name="t", + idempotency_key="idem-1", + ) + + assert result.session_id == "sess-1" + request = route.calls.last.request + sent = request.read() + import json + + body = json.loads(sent) + assert body["target_catalog_path"] == "bio.uploads" + assert body["runner"] == "notebook" + assert body["format"] == "parquet" + assert body["files"] == [{"rel_path": "a.parquet", "size": 6}] + assert body["idempotency_key"] == "idem-1" + # Bearer header carries the Dremio PAT (the kernel's _DREMIO_PWD) + assert request.headers["authorization"] == "Bearer s3cr3t-pwd" + + +@respx.mock +def test_upload_file_posts_to_presigned_policy(creds: DremioCreds) -> None: + route = respx.post("https://s3.example/bucket").mock( + return_value=httpx.Response(204, headers={"ETag": '"abc123"'}) + ) + with IngestClient(BASE_URL, creds) as client: + target = UploadTarget( + rel_path="a.parquet", + url="https://s3.example/bucket", + method="POST", + fields={"key": "p/a.parquet", "policy": "x", "x-amz-signature": "sig"}, + ) + etag = client.upload_file(target, b"PAR1-a") + + assert etag == '"abc123"' + body = route.calls.last.request.read() + # multipart/form-data carries the POST-policy fields and the file part + assert b"PAR1-a" in body + assert b"x-amz-signature" in body + assert b'name="file"' in body + # The DDS Bearer token must NOT leak onto the presigned S3 upload. + assert "authorization" not in route.calls.last.request.headers + + +@respx.mock +def test_upload_rejects_oversize(creds: DremioCreds) -> None: + with IngestClient(BASE_URL, creds) as client: + target = UploadTarget(rel_path="a", url="https://s3.example/a", max_bytes=2) + with pytest.raises(IngestApiError) as exc: + client.upload_file(target, b"too long") + assert exc.value.status_code == 413 + + +@respx.mock +def test_upload_file_multipart_splits_and_puts_parts(creds: DremioCreds) -> None: + p1 = respx.put("https://s3.example/part1").mock( + return_value=httpx.Response(200, headers={"ETag": '"etag-1"'}) + ) + p2 = respx.put("https://s3.example/part2").mock( + return_value=httpx.Response(200, headers={"ETag": '"etag-2"'}) + ) + target = UploadTarget( + rel_path="big.parquet", + method="PUT", + upload_id="up-123", + parts=( + UploadPart(part_number=1, url="https://s3.example/part1"), + UploadPart(part_number=2, url="https://s3.example/part2"), + ), + ) + with IngestClient(BASE_URL, creds) as client: + parts = client.upload_file_multipart(target, b"AAAABBBB") # 8 bytes → 4+4 + + assert parts == [ + {"part_number": 1, "etag": '"etag-1"'}, + {"part_number": 2, "etag": '"etag-2"'}, + ] + # The bytes were split: part 1 got the first half, part 2 the second. + assert p1.calls.last.request.read() == b"AAAA" + assert p2.calls.last.request.read() == b"BBBB" + # No DDS bearer token leaks onto the presigned part PUTs. + assert "authorization" not in p1.calls.last.request.headers + + +@respx.mock +def test_upload_file_multipart_missing_etag_raises(creds: DremioCreds) -> None: + respx.put("https://s3.example/part1").mock(return_value=httpx.Response(200)) + target = UploadTarget( + rel_path="big.parquet", + method="PUT", + upload_id="up-1", + parts=(UploadPart(part_number=1, url="https://s3.example/part1"),), + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.upload_file_multipart(target, b"data") + assert exc.value.status_code == 502 + + +def test_upload_file_multipart_rejects_non_multipart(creds: DremioCreds) -> None: + target = UploadTarget(rel_path="a", url="https://s3.example/a") + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError): + client.upload_file_multipart(target, b"x") + + +@respx.mock +def test_commit_returns_table_path(creds: DremioCreds) -> None: + respx.post(f"{BASE_URL}/api/v1/ingest/commit").mock( + return_value=httpx.Response( + 200, + json={ + "session_id": "sess-1", + "status": "committed", + "table_path": "bio.uploads.t", + "record_count": 42, + }, + ) + ) + with IngestClient(BASE_URL, creds) as client: + result = client.commit(session_id="sess-1") + assert result.table_path == "bio.uploads.t" + assert result.record_count == 42 + + +@respx.mock +def test_error_response_raises_with_message(creds: DremioCreds) -> None: + respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response(409, json={"error": "conflict", "message": "exists"}) + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.begin( + target_catalog_path="x", + intent="read_only", + data_format="csv", + conflict_mode="fail", + files=[], + ) + assert exc.value.status_code == 409 + assert "exists" in str(exc.value) diff --git a/tests/dds_ingestion/test_credentials.py b/tests/dds_ingestion/test_credentials.py new file mode 100644 index 0000000..4c6241c --- /dev/null +++ b/tests/dds_ingestion/test_credentials.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import pytest +from eea_datalakehouse.dds_ingestion.credentials import ( + ENV_BASE_URL, + ENV_PWD, + ENV_USER, + DremioCreds, + MissingCredentialsError, + load_base_url, + load_creds, +) + + +def test_load_creds_from_env() -> None: + creds = load_creds({ENV_USER: "bob", ENV_PWD: "pw"}) + assert creds.username == "bob" + # _DREMIO_PWD is the PAT, sent as the Bearer token. + assert creds.password == "pw" + + +def test_load_creds_missing_raises() -> None: + with pytest.raises(MissingCredentialsError): + load_creds({ENV_USER: "bob"}) + + +def test_load_base_url_strips_trailing_slash() -> None: + assert load_base_url({ENV_BASE_URL: "https://dds.test/"}) == "https://dds.test" + + +def test_load_base_url_missing_raises() -> None: + with pytest.raises(MissingCredentialsError): + load_base_url({}) + + +def test_password_redacted_in_repr_and_str() -> None: + creds = DremioCreds(username="alice", _password="TOP-SECRET") + assert "TOP-SECRET" not in repr(creds) + assert "TOP-SECRET" not in str(creds) + assert "***" in repr(creds) + # the secret is still retrievable via the explicit accessor + assert creds.password == "TOP-SECRET" diff --git a/tests/dds_ingestion/test_folder.py b/tests/dds_ingestion/test_folder.py new file mode 100644 index 0000000..99fcacd --- /dev/null +++ b/tests/dds_ingestion/test_folder.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import logging +import threading +import time +from pathlib import Path + +import pytest +from eea_datalakehouse.dds_ingestion.folder import FolderIngest, scan_folder +from eea_datalakehouse.dds_ingestion.models import ( + BeginResult, + CommitResult, + DataFormat, + FileSpec, + S3Plan, + StatusResult, + UploadPart, + UploadTarget, +) + + +class FakeClient: + """In-memory stand-in for IngestClient that records calls.""" + + def __init__(self, *, already_uploaded: list[str] | None = None) -> None: + self.begin_calls: list[dict[str, object]] = [] + self.uploaded: list[str] = [] + self.commit_calls: list[dict[str, object]] = [] + self.closed = False + self._already_uploaded = already_uploaded or [] + # concurrency instrumentation + self._lock = threading.Lock() + self._active = 0 + self.max_concurrent = 0 + + def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult: + self.begin_calls.append({"files": files, **kwargs}) + uploads = tuple( + UploadTarget(rel_path=f.rel_path, url=f"https://s3.test/{f.rel_path}") + for f in files + ) + return BeginResult( + session_id="sess-X", + status="open", + s3=S3Plan(bucket="b", key_prefix="p/", uploads=uploads), + ) + + def get_status(self, session_id: str) -> StatusResult: + return StatusResult.from_json( + { + "status": "open", + "progress": {"files_done": len(self._already_uploaded), "files_total": 0}, + "uploaded": self._already_uploaded, + } + ) + + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: + with self._lock: + self._active += 1 + self.max_concurrent = max(self.max_concurrent, self._active) + time.sleep(0.02) # widen the window so concurrency is observable + with self._lock: + self._active -= 1 + self.uploaded.append(target.rel_path) + return f'"etag-{target.rel_path}"' + + def commit(self, *, session_id: str, **kwargs: object) -> CommitResult: + self.commit_calls.append({"session_id": session_id, **kwargs}) + return CommitResult( + session_id=session_id, + status="committed", + table_path="bio.uploads.t", + record_count=2, + ) + + def close(self) -> None: + self.closed = True + + +def _make_ingest(folder: Path, client: FakeClient, **kwargs: object) -> FolderIngest: + return FolderIngest( + folder, + "bio.uploads", + data_format="parquet", + show_progress=False, + client=client, # type: ignore[arg-type] + **kwargs, # type: ignore[arg-type] + ) + + +def test_scan_keeps_single_format_only(data_folder: Path) -> None: + specs = scan_folder(data_folder, "parquet") + paths = sorted(s.rel_path for s in specs) + assert paths == ["a.parquet", "sub/b.parquet"] + # csv/json/txt noise is ignored + assert all(p.endswith(".parquet") for p in paths) + + +@pytest.mark.parametrize("fmt", ["csv", "json"]) +def test_scan_other_formats(data_folder: Path, fmt: DataFormat) -> None: + specs = scan_folder(data_folder, fmt) + assert len(specs) == 1 + + +def test_happy_path_begin_upload_commit(data_folder: Path) -> None: + client = FakeClient() + outcome = _make_ingest(data_folder, client).run() + + assert len(client.begin_calls) == 1 + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert len(client.commit_calls) == 1 + assert outcome.commit.table_path == "bio.uploads.t" + assert outcome.files_uploaded == 2 + assert outcome.files_skipped == 0 + # An *injected* client is caller-owned, so FolderIngest must NOT close it. + assert client.closed is False + + +def test_parallel_uploads_run_concurrently(tmp_path: Path) -> None: + for i in range(8): + (tmp_path / f"f{i}.parquet").write_bytes(b"x") + client = FakeClient() + _make_ingest(tmp_path, client, parallelism=4).run() + + assert len(client.uploaded) == 8 + # with 8 files and 4 workers we must observe more than one concurrent upload + assert client.max_concurrent > 1 + assert client.max_concurrent <= 4 + + +def test_resume_skips_already_uploaded(data_folder: Path) -> None: + client = FakeClient(already_uploaded=["a.parquet"]) + outcome = _make_ingest(data_folder, client).run() + + # only the not-yet-done file is uploaded + assert client.uploaded == ["sub/b.parquet"] + assert outcome.files_uploaded == 1 + assert outcome.files_skipped == 1 + + +def test_no_matching_files_raises(tmp_path: Path) -> None: + (tmp_path / "only.csv").write_text("a\n") + client = FakeClient() + with pytest.raises(FileNotFoundError): + _make_ingest(tmp_path, client).run() + + +def test_credentials_never_logged( + data_folder: Path, caplog: pytest.LogCaptureFixture +) -> None: + client = FakeClient() + with caplog.at_level(logging.DEBUG, logger="eea_datalakehouse.dds_ingestion"): + _make_ingest(data_folder, client).run() + blob = "\n".join(r.getMessage() for r in caplog.records) + assert "s3cr3t" not in blob + assert "_password" not in blob + + +def test_invalid_parallelism_rejected(data_folder: Path) -> None: + client = FakeClient() + with pytest.raises(ValueError): + _make_ingest(data_folder, client, parallelism=0) + + +class MultipartFakeClient(FakeClient): + """FakeClient whose begin returns a multipart target per file (DI-7.3).""" + + def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult: + self.begin_calls.append({"files": files, **kwargs}) + uploads = tuple( + UploadTarget( + rel_path=f.rel_path, + method="PUT", + upload_id=f"up-{f.rel_path}", + parts=(UploadPart(part_number=1, url=f"https://s3.test/{f.rel_path}/1"),), + ) + for f in files + ) + return BeginResult( + session_id="sess-MP", + status="open", + s3=S3Plan(bucket="b", key_prefix="p/", uploads=uploads), + ) + + def upload_file_multipart( + self, target: UploadTarget, data: bytes + ) -> list[dict[str, object]]: + with self._lock: + self.uploaded.append(target.rel_path) + return [{"part_number": 1, "etag": f'"etag-{target.rel_path}"'}] + + +def test_multipart_etags_threaded_into_commit(data_folder: Path) -> None: + client = MultipartFakeClient() + outcome = _make_ingest(data_folder, client, multipart=True).run() + + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert outcome.files_uploaded == 2 + # commit received the contract-shaped multipart_etags list. + sent = client.commit_calls[0]["multipart_etags"] + assert isinstance(sent, list) + by_rel = {e["rel_path"]: e for e in sent} + assert set(by_rel) == {"a.parquet", "sub/b.parquet"} + entry = by_rel["a.parquet"] + assert entry["upload_id"] == "up-a.parquet" + assert entry["parts"] == [{"part_number": 1, "etag": '"etag-a.parquet"'}] diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py deleted file mode 100644 index eb54ade..0000000 --- a/tests/test_pipeline.py +++ /dev/null @@ -1,32 +0,0 @@ -import pytest - -from eea_datalakehouse_ingestion import IngestionPipeline, IngestionValidationError - - -def test_hello_world_returns_greeting(): - assert IngestionPipeline().hello_world() == "Hello, world!" - - -def test_prepare_normalizes_records_into_plain_dicts(): - pipeline = IngestionPipeline() - source = ({"id": 1, "value": "a"}, {"id": 2, "value": "b"}) - - result = pipeline.prepare(source) - - assert result == [{"id": 1, "value": "a"}, {"id": 2, "value": "b"}] - assert all(type(record) is dict for record in result) - - -def test_validate_passes_when_required_fields_present(): - pipeline = IngestionPipeline(required_fields=["id", "value"]) - records = pipeline.prepare([{"id": 1, "value": "a"}]) - - assert pipeline.validate(records) is True - - -def test_validate_raises_on_missing_required_field(): - pipeline = IngestionPipeline(required_fields=["id", "value"]) - records = pipeline.prepare([{"id": 1}]) - - with pytest.raises(IngestionValidationError, match="record 0"): - pipeline.validate(records)