feat(elt-pipelines): Add initial project with example pipeline

.pre-commit-config.yaml

@@ -12,10 +12,10 @@ repos:
     hooks:
       # Run the linter.
       - id: ruff-check
-        files: (elt-common|warehouses)
+        files: (elt-common|elt-pipelines|warehouses)
       # Run the formatter.
       - id: ruff-format
-        files: (elt-common|warehouses)
+        files: (elt-common|elt-pipelines|warehouses)
   - repo: https://github.com/shellcheck-py/shellcheck-py
     rev: v0.11.0.1
     hooks:

elt-pipelines/.gitignore

@@ -0,0 +1,7 @@
+# ignore basic python artifacts
+.env
+**/__pycache__/
+**/*.py[cod]
+**/*$py.class
+**/build/
+**/*.egg-info/

elt-pipelines/README.md

@@ -0,0 +1,34 @@
+# elt-pipelines
+
+Pipelines for ingesting data from various sources into Iceberg catalogs using elt-common.
+
+**Under construction** - this project is being developed as a replacement for the existing DLT pipelines.
+[See here](https://github.com/ISISNeutronMuon/analytics-data-platform/issues/321) for details.
+
+## Development setup
+
+Development requires the following tools:
+
+- [uv](https://docs.astral.uv/uv/): Used to manage both Python installations and dependencies
+
+### Setting up a Python virtual environment
+
+Once `uv` is installed, create an environment, activate it, and install dependencies with:
+
+```bash
+> uv venv
+> source .venv/bin/activate
+> uv sync
+```
+
+Pipelines can declare optional dependencies in `pyproject.toml` - for example, `statusdisplay` uses `requests` for
+fetching data. To install any additional dependencies for that specific pipeline, use:
+
+```bash
+> uv sync --extra statusdisplay
+```
+
+## Running a pipeline
+
+Pipelines are run using the `elt` CLI tool. As an example, with the package as current working directory,
+`elt run pipelines statusdisplay` will run the statusdisplay pipeline. See `elt -h` for full usage.

elt-pipelines/pipelines/ingest/accelerator/statusdisplay/statusdisplay.py

@@ -0,0 +1,57 @@
+from datetime import datetime
+import io
+import json
+import pyarrow.json
+import requests
+
+from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties
+
+CYCLES_URL = "https://status.isis.stfc.ac.uk/api/cycles"
+
+
+class Extract(BaseExtract):
+    def extract_resource_properties(self):
+        yield (
+            "elt_cycles",
+            ResourceProperties(
+                extractor=extract_cycles,
+                write_properties=ResourceWriteProperties(write_mode="replace"),
+                watermark_column=None,
+            ),
+        )
+
+
+def extract_cycles(_):
+    data = clean(fetch())
+    newline_delimited = "\n".join(json.dumps(row) for row in data)
+
+    with io.BytesIO(newline_delimited.encode()) as f:
+        yield pyarrow.json.read_json(f)
+
+
+def fetch():
+    try:
+        response = requests.get(CYCLES_URL, timeout=20)
+    except requests.Timeout as ex:
+        raise RuntimeError("Timed out when fetching cycles") from ex
+
+    if not response.ok:
+        raise RuntimeError(f"Failed to fetch cycles - {response.reason}")
+
+    return response.json()
+
+
+def reformat(date_string):
+    """Convert a date from ISO format into one that pyarrow will convert into a timestamp"""
+    return datetime.fromisoformat(date_string).strftime("%Y-%m-%d %H:%M:%S")
+
+
+def clean(data):
+    for cycle in data:
+        for phase in cycle["phases"]:
+            if "start" in phase:
+                phase["start"] = reformat(phase["start"])
+            if "end" in phase:
+                phase["end"] = reformat(phase["end"])
+
+    return data

elt-pipelines/pyproject.toml

@@ -0,0 +1,24 @@
+[project]
+name = "elt-pipelines"
+version = "0.1.0"
+description = "Pipelines for ingesting data into Iceberg catalogs"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "elt-common",
+    "pydantic-settings>=2.14.2",
+]
+
+[project.optional-dependencies]
+statusdisplay = [
+    "pyarrow>=24.0.0",
+    "requests>=2.34.2",
+]
+
+[tool.uv.sources]
+elt-common = { path = "../elt-common", editable = true }
+
+[dependency-groups]
+dev = [
+    "prek>=0.4.5",
+]