Metadata-Version: 2.4
Name: orcapod
Version: 0.1.4
Summary: Intuitive and powerful library for highly reproducible scientific data pipeline
Project-URL: Homepage, https://github.com/nauticalab/orcapod-python
License: MIT License
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.11.0
Requires-Dist: basedpyright>=1.38.1
Requires-Dist: beartype>=0.21.0
Requires-Dist: deltalake>=1.0.2
Requires-Dist: gitpython>=3.1.45
Requires-Dist: graphviz>=0.21
Requires-Dist: matplotlib>=3.10.3
Requires-Dist: networkx
Requires-Dist: numpy>=1.26.4
Requires-Dist: pandas>=2.2.3
Requires-Dist: polars>=1.36.0
Requires-Dist: pyarrow>=20.0.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pygraphviz>=1.14
Requires-Dist: pymongo>=4.15.5
Requires-Dist: pyyaml>=6.0.2
Requires-Dist: s3fs>=2025.12.0
Requires-Dist: starfix~=0.4.0
Requires-Dist: typer>=0.12
Requires-Dist: typing-extensions
Requires-Dist: tzdata>=2024.1
Requires-Dist: universal-pathlib>=0.3.8
Requires-Dist: uuid-utils>=0.11.1
Requires-Dist: xxhash
Provides-Extra: all
Requires-Dist: ipywidgets>=8.1.7; extra == 'all'
Requires-Dist: psycopg[binary]>=3.0; extra == 'all'
Requires-Dist: pyspiral>=0.14.0; extra == 'all'
Requires-Dist: ray[default]==2.48.0; extra == 'all'
Requires-Dist: redis>=6.2.0; extra == 'all'
Provides-Extra: postgresql
Requires-Dist: psycopg[binary]>=3.0; extra == 'postgresql'
Provides-Extra: ray
Requires-Dist: ipywidgets>=8.1.7; extra == 'ray'
Requires-Dist: ray[default]==2.48.0; extra == 'ray'
Provides-Extra: redis
Requires-Dist: redis>=6.2.0; extra == 'redis'
Provides-Extra: spiraldb
Requires-Dist: pyspiral>=0.14.0; extra == 'spiraldb'
Description-Content-Type: text/markdown

# Orcapod Python
Orcapod's Python library for developing reproducbile scientific pipelines.

## Releasing

To cut a release, tag a commit on `main` — `hatch-vcs` derives the version
automatically and CI publishes to PyPI. See [RELEASING.md](RELEASING.md) for the
full workflow.

## Continuous Integration

This project uses GitHub Actions for continuous integration:

- **Run Tests**: Runs the full unit test suite on Ubuntu with Python 3.11 and 3.12. PostgreSQL integration tests are excluded.
- **Run PostgreSQL Tests**: Runs only the `@pytest.mark.postgres` integration tests against a PostgreSQL 16 service container.

### Running Tests Locally

This project uses [uv](https://docs.astral.sh/uv/) for dependency management.

```bash
# Install all dependencies
uv sync --all-extras --dev

# Run the standard test suite (no PostgreSQL required)
uv run pytest -m "not postgres"

# Run with coverage
uv run pytest -m "not postgres" --cov=src --cov-report=term-missing
```

### Running PostgreSQL Integration Tests Locally

The PostgreSQL integration tests require a running PostgreSQL instance. The easiest way is Docker:

```bash
docker run --rm -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:16
```

Then in another terminal:

```bash
PGPASSWORD=postgres uv run pytest -m postgres -v
```

The tests connect via standard `PG*` environment variables — override any of these defaults as needed:

| Variable     | Default     |
|--------------|-------------|
| `PGHOST`     | `localhost`  |
| `PGPORT`     | `5432`       |
| `PGDATABASE` | `testdb`     |
| `PGUSER`     | `postgres`   |
| `PGPASSWORD` | `postgres`   |

## Lower-level details on pipeline workflow (tentative)

While the following is subject to change based on future development, it represents a nice high-level overview of some of the lower-level details of how Orcapod pipelines work and how the various objects interact with one another.

### Pipeline setup

  Instantiate:

  - OBSERVER with a log database
    - Two kinds: NoOpObserver, LoggingObserver
  - ORCHESTRATOR with this observer
    - Two kinds: SyncPipelineOrchestrator, AsyncPipelineOrchestrator
  - EXECUTOR
    - Two kinds: LocalExecutor, RayExecutor
    - Executors have a slim base protocol (DataFunctionExecutorProtocol — identity, compatibility, lifecycle) and a Python-specific subtype (PythonFunctionExecutorProtocol — adds
   execute_callable / async_execute_callable). This leaves room for non-Python executor types in the future.
  - PIPELINE (name, DB)

### Construct PIPELINE using with pipeline

  "Compile" PIPELINE
  - auto_compile=False is default
  - Walk the DAG, wrap each element in a node that owns execution logic, fixates input stream, attaches caching
  - Nodes (function, operator) are associated with storage/DB
  - If a function node has no executor, Pipeline.compile() assigns LocalExecutor() by default — so capture/logging works out of the box without explicit executor setup

  Run PIPELINE by passing several objects to pipeline.run():
  - The EXECUTOR (optional) — node.executor = ray_executor.with_options(**opts) called on every function node, replacing the compile-time default
    - Per-node override also possible: pipeline.transform_a.executor = RayExecutor(...) after compile
  - The ORCHESTRATOR
    - Calls obs.on_run_start(run_id)
    - Walks nodes in topological order (or launches all concurrently for async)
    - Wires nodes together (edges in DAG) with channels (async) or buffers (sync)
    - For each node, calls node.execute(input_stream, observer=obs) or node.async_execute(input_channel, output_channel, observer=obs)
    - on_run_end(run_id) is in a try/finally — always fires, even on failure
  - Engine opts dict — forwarded to executor via with_options(**opts)

### Function node processing

  Capture happens at the bottom of the chain (inside the executor). A logger is injected DOWN the call chain — FunctionNode creates it from the observer and passes it through to
  the executor. The executor captures I/O, calls logger.record(**captured_fields), and re-raises on failure. No CapturedLogs appears in any return type or protocol — it's an
  internal convenience struct inside the executor.

  The observer is contextualized per-node: FunctionNode.execute() calls observer.contextualize(node_hash, node_label) to get a lightweight wrapper stamped with node identity. If no
   observer is provided, a NoOpObserver is used as default, eliminating all is not None checks.

  The logger protocol is generic: DataExecutionLoggerProtocol.record(**kwargs) accepts arbitrary keyword arguments, so different executor types can log different fields without
  the protocol being tied to any specific data structure.

  If no executor is set and a logger is passed, a UserWarning is emitted: "A logger was passed but no executor is set — capture will not occur."

  Down the chain (FunctionNode → user function)

  FunctionNode.execute(input_stream, observer=obs)
      │
      │  obs = observer or NoOpObserver()
      │  ctx_observer = obs.contextualize(node_hash, node_label)
      │  ctx_observer.on_node_start(node_label, node_hash)
      │
      │  for each non-cached (tag, data):
      │
      │  ctx_observer.on_data_start(node_label, tag, data)
      │
      ├─► pkt_logger = ctx_observer.create_data_logger(tag, data, pipeline_path=...)
      │       │
      │       └─► _ContextualizedLoggingObserver creates a DataLogger bound to
      │           (run_id, node_label, node_hash, tag_data, log_path)
      │
      ├─► FunctionNode._process_data_internal(tag, data, logger=pkt_logger)
      │       │
      │       ├─► CachedFunctionPod.process_data(tag, data, logger=pkt_logger)
      │       │       │
      │       │       │  checks pod-level cache (ResultCache.lookup)
      │       │       │  cache hit? → return (tag, cached_data)
      │       │       │  cache miss ↓
      │       │       │
      │       │       ├─► _FunctionPodBase.process_data(tag, data, logger=pkt_logger)
      │       │       │       │
      │       │       │       ├─► PythonDataFunction.call(data, logger=pkt_logger)
      │       │       │       │       │
      │       │       │       │       │  executor is set? (LocalExecutor from compile,
      │       │       │       │       │  or RayExecutor from pipeline.run)
      │       │       │       │       │
      │       │       │       │       ├─► executor.execute_callable(fn, kwargs, logger=pkt_logger)
      │       │       │       │       │       │
      │       │       │       │       │       │  ── LocalExecutor path ──
      │       │       │       │       │       │  ctx = LocalCaptureContext()
      │       │       │       │       │       │  with ctx:
      │       │       │       │       │       │    fn(**kwargs)               ← USER FUNCTION RUNS HERE
      │       │       │       │       │       │  on success: logger.record(**captured.as_dict())
      │       │       │       │       │       │  on exception: logger.record(**captured.as_dict())
      │       │       │       │       │       │                then RE-RAISE the original exception
      │       │       │       │       │       │
      │       │       │       │       │       │  ── RayExecutor path ──
      │       │       │       │       │       │  dispatches _make_capture_wrapper() to Ray
      │       │       │       │       │       │  Ray worker runs the wrapper which:
      │       │       │       │       │       │    - redirects fd 1/2 to temp files
      │       │       │       │       │       │    - installs a logging handler
      │       │       │       │       │       │    - calls fn(**kwargs)       ← USER FUNCTION RUNS HERE
      │       │       │       │       │       │    - on success: returns (raw, stdout, stderr, python_logs)
      │       │       │       │       │       │    - on failure: RAISES _CapturedTaskError(cause, stdout, ...)
      │       │       │       │       │       │      → Ray propagates via RayTaskError (retries etc. work)
      │       │       │       │       │       │  driver side:
      │       │       │       │       │       │    success → logger.record(stdout=..., stderr=..., success=True, ...)
      │       │       │       │       │       │    failure → logger.record(stdout=..., success=False, ...)
      │       │       │       │       │       │              then RE-RAISE the original user exception
      │       │       │       │       │       │
      │       │       │       │       │       └─► returns raw_result (or raises)
      │       │       │       │       │
      │       │       │       │       │  no executor? warns if logger was passed, calls direct_call()
      │       │       │       │       │
      │       │       │       │       │  builds output Data from raw_result
      │       │       │       │       │
      │       │       │       │       └─► returns Data | None (or raises)
      │       │       │       │
      │       │       │       └─► returns (tag, Data | None)
      │       │       │
      │       │       │  stores result in pod-level cache (on success)
      │       │       │
      │       │       └─► returns (tag, Data | None)
      │       │
      │       │  writes pipeline provenance record (on success)
      │       │  caches result internally
      │       │
      │       └─► returns (tag, Data | None)
      │
      │  ← back in FunctionNode.execute() with (tag_out, result)
      │
      │  (logger.record already called inside the executor — nothing to do here)
      │
      ├─► try/except around _process_data_internal:
      │   on success:
      │       ctx_observer.on_data_end(node_label, tag, data, result, cached=False)
      │       emit (tag_out, result) downstream
      │   on exception:
      │       ctx_observer.on_data_crash(node_label, tag, data, exc)
      │       if error_policy == "fail_fast": raise
      │       otherwise: skip this data, continue
      │
      ctx_observer.on_node_end(node_label, node_hash)

  DataLogger.record(**kwargs)

  DataLogger.record(**kwargs)
      │
      └─► builds a pyarrow table row:
          Context columns (always present, prefixed with "__"):
            `__log_id`, `__run_id`, `__node_label`, `__node_hash`, `__timestamp`
          Execution output columns (from **kwargs, prefixed with "`__`"):
            __stdout, __stderr, __python_logs, __traceback, __success
            (or any other fields the executor passes — protocol is generic)
          Tag columns (unprefixed, from tag_data baked in at creation):
            e.g. "idx" → "0", "key" → "a"

Writes the row to the database at the mirrored log path.

### Key design points

  - CapturedLogs never appears in any return type or protocol — it's an internal convenience struct inside executors. The protocol uses record(**kwargs) so different executor types
   can log different fields.
  - Exceptions propagate — direct_call() and executors re-raise user exceptions after recording logs. Error handling happens at the FunctionNode.execute() boundary via try/except.
  - Ray exceptions propagate through Ray — the worker raises _CapturedTaskError (carrying captured I/O), Ray wraps it in RayTaskError (retries etc. work normally), the driver
  extracts the captured data, records to logger, and re-raises the original user exception type.
  - Observer is contextualized per-node — contextualize(node_hash, node_label) returns a lightweight wrapper. No GraphNode reference crosses the protocol boundary — only strings.
  NoOpObserver is used as default when no observer is provided, eliminating null checks.
  - pipeline_path is a storage detail — passed to create_data_logger() for routing log table location, but NOT stored as a column in log rows (only node_hash + node_label
  identify the node).
  - No auto-executor in the class — DataFunctionBase.__init__ does not assign a default executor. Pipeline.compile() assigns LocalExecutor to function nodes that have none. Users
   can override per-node (pipeline.node.executor = ...) or globally via pipeline.run(execution_engine=...).
  - Log columns use __ prefix — fixed columns (__log_id, __stdout, __success, etc.) are prefixed to avoid collision with user-defined tag column names.