feat!: v3 add middlewares

.windsurf/rules/wurzel-dev-rule.md

@@ -0,0 +1,117 @@
+---
+trigger: always_on
+---
+
+# Wurzel Project Rules & Context
+
+## Core Architecture
+
+### TypedStep System (`wurzel/core/`)
+- **TypedStep[SETTINGS, INCONTRACT, OUTCONTRACT]**: Base class for all steps
+- Chain with `>>` operator (type-safe): `source >> splitter >> embedder`
+- Input can be `None` for leaf steps, output always requiredAfter ANY Code Change:
+
+### Data Contracts (`wurzel/datacontract/`)
+- **PydanticModel**: Structured objects (JSON), **PanderaDataFrameModel**: Tabular data (CSV)
+- Both implement `save_to_path()` and `load_from_path()`
+
+### Executors (`wurzel/step_executor/`)
+- **BaseStepExecutor**: Core engine with env encapsulation
+- **PrometheusStepExecutor**: Adds metrics
+- Middleware support via `MiddlewareRegistry`
+
+### Backends (`wurzel/backend/`)
+- **DvcBackend**: Generates `dvc.yaml`, **ArgoBackend**: Argo Workflows YAML
+
+### CLI (`wurzel/cli/`)
+- `wurzel run/inspect/generate/middlewares`
+
+### Built-in Steps (`wurzel/steps/`)
+- ManualMarkdown, SimpleSplitter, Embedding, Qdrant/Milvus connectors, Docling, ScraperAPI
+
+## Development Workflow (CRITICAL - Always Follow)
+
+### After ANY Code Change:
+0. **Check and update documentation** (`docs/` - mkdocs format)
+1. **Lint**: `make lint` (runs pre-commit: ruff, pylint, reuse)
+2. **Test**: `make test` (pytest with 90% coverage requirement)
+
+### Before Committing:
+- Ensure all tests pass
+- Verify linting passes
+- Update relevant documentation in `docs/`
+- Follow conventional commit format (feat/fix/docs/refactor/test/chore)
+## Tech Stack & Standards
+
+### Core Technologies
+- **Pydantic v2**: For data validation and settings
+  - Use `pydantic.BaseModel` for data models
+  - Use `wurzel.step.Settings` for step settings (NOT `pydantic_settings.BaseSettings`)
+  - Custom implementations in `wurzel.datacontract` and `wurzel.step.settings`
+- **Pandera**: For DataFrame validation
+- **uv**: Package manager (NOT pip directly)
+- **mkdocs**: Documentation (with material theme, mermaid, typer integration)
+- **DVC**: Pipeline versioning and execution
+- **typer**: CLI framework
+
+
+## Environment Variables
+
+```bash
+# Step settings: <STEP_NAME_UPPERCASE>__<SETTING_NAME>
+export MANUALMARKDOWNSTEP__FOLDER_PATH=/path/to/docs
+export ALLOW_EXTRA_SETTINGS=True
+export MIDDLEWARES=prometheus
+export DVCBACKEND__DATA_DIR=./data
+```
+
+## Implementation Patterns
+
+### Step
+```python
+from wurzel.core import TypedStep, Settings
+from wurzel.datacontract import MarkdownDataContract
+
+class MyStepSettings(Settings):
+    my_param: str
+
+class MyStep(TypedStep[MyStepSettings, InputType, OutputType]):
+    def run(self, inpt: InputType) -> OutputType:
+        return result
+```
+
+### Pipeline
+```python
+from wurzel.utils import WZ
+source = WZ(SourceStep)
+processor = WZ(ProcessStep)
+source >> processor
+pipeline = processor
+```
+
+### Execution
+```python
+with BaseStepExecutor(middlewares=["prometheus"]) as ex:
+    ex(MyStep, {Path("./input")}, Path("./output"))
+```
+
+## Documentation Updates
+- New step → `docs/developer-guide/creating-steps.md`
+- New backend → `docs/backends/index.md`
+- New middleware → `docs/executor/middlewares.md`
+
+## Key Notes
+- TypedStep enforces type compatibility at definition time
+- Steps run in isolated env (settings from env vars with `STEPNAME__` prefix)
+- **Settings**: Use `wurzel.step.Settings` (custom wrapper around pydantic_settings)
+  - Supports nested settings with `__` delimiter
+  - Auto-loads from env vars with step name prefix
+  - Use `NoSettings` type alias for steps without settings
+- History tracking: `[source].[step1].[step2]...`
+- Optional deps: `wurzel[qdrant,milvus,argo]`, check `wurzel.utils.HAS_*`
+
+## Troubleshooting
+- Import errors: Use full module path
+- Type errors: Check INCONTRACT/OUTCONTRACT compatibility
+- Settings errors: Verify `STEPNAME__SETTING` format
+- `wurzel inspect module.path.StepName` for details

README.md

@@ -50,7 +50,7 @@ Run a step using the snippet below:
 import os
 from pathlib import Path
 
-from wurzel.step_executor import BaseStepExecutor
+from wurzel.executors import BaseStepExecutor
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 
 # Create input dir and set folder (required by ManualMarkdownStep)

REUSE.toml

@@ -29,6 +29,8 @@ path = [
     "**/requirements.txt",
     "**/*.pyc",
     "**/__pycache__/**",
+    "**/ub/**/*",
+    ".windsurf/**/*",
     ".python-version",
     ]
 precedence = "override"

docs/backends/argoworkflows.md

@@ -123,9 +123,17 @@ workflows:
 
       # Runtime environment variables (step settings)
       env:
+        # Step-specific settings
         MANUALMARKDOWNSTEP__FOLDER_PATH: "examples/pipeline/demo-data"
         SIMPLESPLITTERSTEP__BATCH_SIZE: "100"
 
+        # Middleware configuration (optional)
+        # Enable Prometheus middleware for metrics collection
+        MIDDLEWARES: "prometheus"
+        PROMETHEUS__PROMETHEUS_GATEWAY: "prometheus-pushgateway.monitoring.svc.cluster.local:9091"
+        PROMETHEUS__PROMETHEUS_JOB: "wurzel-pipeline"  # optional
+        PROMETHEUS__PROMETHEUS_DISABLE_CREATED_METRIC: "true"  # optional
+
       # Environment from Kubernetes Secrets/ConfigMaps
       envFrom:
         - kind: secret
@@ -227,22 +235,6 @@ workflows:
 - `"0 0 * * 0"` - Weekly on Sundays at midnight
 - `"0 0 1 * *"` - Monthly on the 1st at midnight
 
-**Monitoring:**
-```bash
-# List all CronWorkflows
-argo cron list
-
-# View CronWorkflow details
-argo cron get my-scheduled-pipeline
-
-# List workflow runs from CronWorkflow
-argo list --label workflows.argoproj.io/cron-workflow=my-scheduled-pipeline
-```
-
-!!! tip "Choosing the Right Type"
-    - Use **Workflow** (schedule: null) when you need explicit control over when pipelines run
-    - Use **CronWorkflow** (with schedule) for automated, time-based execution
-    - You can have both: a CronWorkflow for regular execution and a Workflow template for manual reruns
 
 ### Configuration Reference
 
@@ -325,6 +317,43 @@ When enabled, the `HF_HOME` environment variable is automatically set to the `mo
 | `endpoint` | string | `s3.amazonaws.com` | S3 endpoint URL |
 | `defaultMode` | int | `null` | File permissions (decimal) |
 
+### Middleware Configuration
+
+Middlewares (like Prometheus for metrics collection) are configured via environment variables in the `container.env` section. Middlewares must be enabled and configured at **generate-time** in your `values.yaml` file.
+
+#### Enabling Prometheus Middleware
+
+To enable Prometheus middleware for metrics collection, add the following to your `container.env` section:
+
+```yaml
+container:
+  env:
+    # Enable Prometheus middleware
+    MIDDLEWARES: "prometheus"
+    PROMETHEUS__PROMETHEUS_GATEWAY: "prometheus-pushgateway.monitoring.svc.cluster.local:9091"
+
+    # Optional Prometheus settings
+    PROMETHEUS__PROMETHEUS_JOB: "wurzel-pipeline"
+    PROMETHEUS__PROMETHEUS_DISABLE_CREATED_METRIC: "true"
+```
+
+**Available Prometheus Settings:**
+
+| Environment Variable | Default | Description |
+|---------------------|---------|-------------|
+| `MIDDLEWARES` | - | Comma-separated list of middleware names (e.g., `"prometheus"`) |
+| `PROMETHEUS__PROMETHEUS_GATEWAY` | `localhost:9091` | Prometheus Pushgateway endpoint (host:port) |
+| `PROMETHEUS__PROMETHEUS_JOB` | `default-job-name` | Job name for Prometheus metrics |
+| `PROMETHEUS__PROMETHEUS_DISABLE_CREATED_METRIC` | `true` | Disable `*_created` metrics |
+
+**Metrics Collected:**
+
+- `step_duration_seconds` - Histogram of step execution duration
+- `step_executions_total` - Counter of step executions
+- Labels: `step_name`, `run_id` (from `WURZEL_RUN_ID`)
+
+For more details on middlewares, see the [Middleware Documentation](../executor/middlewares.md).
+
 ### Runtime Environment Variables
 
 Step settings are configured via environment variables at **runtime** (when the workflow executes). These can be set in three ways:
@@ -364,7 +393,7 @@ container:
 Use the Argo backend directly in Python:
 
 ```python
-from wurzel.backend.backend_argo import ArgoBackend
+from wurzel.executors.backend.backend_argo import ArgoBackend
 from wurzel.steps.embedding import EmbeddingStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.steps.qdrant.step import QdrantConnectorStep
@@ -386,54 +415,6 @@ argo_yaml = backend.generate_artifact(pipeline)
 # backend = ArgoBackend.from_values(files=[Path("values.yaml")], workflow_name="pipelinedemo")
 ```
 
-## Deploying Argo Workflows
-
-Once you've generated your Workflow or CronWorkflow YAML, deploy it to your Kubernetes cluster:
-
-### Deploying a Normal Workflow
-
-```bash
-# Apply the Workflow to your cluster
-kubectl apply -f workflow.yaml
-
-# Submit it for execution
-argo submit workflow.yaml
-
-# Or create and submit in one command
-kubectl create -f workflow.yaml
-```
-
-### Deploying a CronWorkflow
-
-```bash
-# Apply the CronWorkflow to your cluster (starts the cron schedule)
-kubectl apply -f cronworkflow.yaml
-
-# View CronWorkflow status
-argo cron get wurzel-pipeline
-
-# List CronWorkflows
-argo cron list
-```
-
-### Monitoring Workflow Executions
-
-```bash
-# List all workflow executions
-argo list
-
-# Get detailed workflow status
-argo get <workflow-name>
-
-# View workflow logs
-argo logs <workflow-name>
-
-# Follow logs in real-time
-argo logs <workflow-name> -f
-
-# View logs for specific step
-argo logs <workflow-name> -c <container-name>
-```
 
 ## Benefits for Cloud-Native Pipelines
 

docs/backends/dvc.md

@@ -98,7 +98,7 @@ dvc repro
 Use the DVC backend directly in Python:
 
 ```python
-from wurzel.backend.backend_dvc import DvcBackend
+from wurzel.executors.backend.backend_dvc import DvcBackend
 from wurzel.steps.embedding import EmbeddingStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.steps.qdrant.step import QdrantConnectorStep

docs/datacontract/common.md

@@ -2,6 +2,13 @@
 
 Data contracts are the primarily inputs and outputs of pipeline steps, e.g., Markdown documents.
 
+## Metrics
+
+Data contracts can optionally expose numeric metrics (for example, counts or sizes).
+For `PydanticModel`-based contracts, implement a `metrics()` method on the instance.
+For `PanderaDataFrameModel` contracts, override `get_metrics(cls, obj)` on the class.
+Middlewares (like Prometheus) can use these metrics if provided.
+
 ## MarkdownDataContract
 
 ::: wurzel.datacontract.common.MarkdownDataContract

docs/developer-guide/building-pipelines.md

@@ -37,7 +37,7 @@ source >> embedding >> storage
 Define a function that builds the chain and returns the last step. Wurzel runs upstream steps in order:
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.utils import WZ
@@ -60,7 +60,7 @@ Execution order: ManualMarkdownStep → EmbeddingStep → QdrantConnectorStep.
 One source can feed multiple downstream steps:
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.steps.splitter import SimpleSplitterStep
@@ -82,7 +82,7 @@ def branching_pipeline() -> TypedStep:
 Choose steps at build time:
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.embedding import TruncatedEmbeddingStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
@@ -110,7 +110,7 @@ Use environment variables to choose steps:
 ```python
 import os
 
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.embedding import TruncatedEmbeddingStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
@@ -129,7 +129,7 @@ def configurable_pipeline() -> TypedStep:
 ## Testing Pipelines
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.utils import WZ
@@ -161,7 +161,7 @@ def test_complete_pipeline():
 Independent branches can run in parallel (backend-dependent):
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.utils import WZ
@@ -199,7 +199,7 @@ Steps cache outputs based on input changes; backends handle persistence.
 ### ETL-style (extract → transform → load)
 
 ```python
-from wurzel.step import TypedStep
+from wurzel.core import TypedStep
 from wurzel.steps import EmbeddingStep, QdrantConnectorStep
 from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.steps.splitter import SimpleSplitterStep

docs/developer-guide/cli.md

@@ -48,11 +48,11 @@ wurzel run wurzel.steps.manual_markdown.ManualMarkdownStep \
     --inputs ./markdown-files \
     --output ./processed-output
 
-# With custom executor
+# With middlewares (e.g., prometheus metrics)
 wurzel run wurzel.steps.manual_markdown.ManualMarkdownStep \
     --inputs ./markdown-files \
     --output ./processed-output \
-    --executor PrometheusStepExecutor
+    --middlewares prometheus
 
 # Multiple input folders
 wurzel run wurzel.steps.splitter.SimpleSplitterStep \