pycapio — high-level API

The top-level package exposes the CapioContext() decorator and the CAPIO-aware proxy functions that replace Python’s built-in I/O entry points.

Package

Top-level package for PyCAPIO.

PyCAPIO brings transparent data streaming to file-based Python workflows by monkey patching Python’s built-in I/O entry points (open(), os.mkdir(), os.scandir(), …) and routing any access that targets the configured CAPIO directory through the native CAPIO server instead of the operating system.

The public surface of this module is intentionally small:

• CapioContext() – a decorator that initialises CAPIO and patches the built-in I/O functions for the duration of the decorated call.

• the *_proxy callables (open_proxy(), mkdir_proxy(), …) – drop-in replacements for the corresponding built-ins that dispatch to CAPIO when a path lives inside the CAPIO directory and fall back to the original implementation otherwise.

Everything else re-exported here (pycapio_open, PyCapioPath, FILE_MODES, the IO wrappers, …) comes from the compiled extension pycapio._pycapio and is documented in the Native API section.

Example

Intercept only the I/O performed inside a single function:

from pycapio import CapioContext

@CapioContext(capio_dir=".", app_name="reader",
              workflow_name="example_workflow")
def read(path):
    with open(path, "r") as f:
        return f.read()

pycapio.wraps(wrapped, assigned=('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__'), updated=('__dict__',))

Decorator factory to apply update_wrapper() to a wrapper function

Returns a decorator that invokes update_wrapper() with the decorated function as the wrapper argument and the arguments to wraps() as the remaining arguments. Default arguments are as for update_wrapper(). This is a convenience function to simplify applying partial() to update_wrapper().

class pycapio.Any(*args, **kwargs)

Bases: object

Special type indicating an unconstrained type.

• Any is compatible with every type.

• Any assumed to have all methods.

• All values assumed to be instances of Any.

Note that all the above statements are true from the point of view of static type checkers. At runtime, Any should not be used with instance checks.

class pycapio.PyCAPIOException

Bases: object

pycapio.CapioContext(*, capio_dir='.', app_name='writer', workflow_name='workflow', silent=True, server_exec_path='capio_server', capio_cl_configuration_file='', await_server_timeout_seconds=2, teardown_server=True)

Decorator factory that runs a function with CAPIO interception enabled.

On first use within a process this initialises CAPIO (starting/attaching to the CAPIO server) and registers teardown via atexit. For every call of the decorated function it swaps the built-in I/O callables for their CAPIO-aware proxies, runs the function, and restores the originals afterwards – so interception is scoped to the decorated call only.

All arguments are keyword-only.

Parameters:

• capio_dir – Root directory managed by CAPIO. I/O on paths inside this directory is intercepted; everything else falls back to the standard library. Defaults to ".".

• app_name – Logical application name reported to CAPIO. Defaults to CAPIO_DEFAULT_APP_NAME.

• workflow_name – Logical workflow name reported to CAPIO. Defaults to CAPIO_DEFAULT_WORKFLOW_NAME.

• silent – When True (default) sets the SILENT environment variable to "ON" to suppress CAPIO server chatter.

• server_exec_path – Path or name of the CAPIO server executable. Defaults to "capio_server".

• capio_cl_configuration_file – Optional path to a CAPIO-CL configuration file. Empty string means no configuration file.

• await_server_timeout_seconds – How long to wait for the CAPIO server to become ready before giving up. Defaults to 2.

• teardown_server – When True (default) the CAPIO server is torn down at process exit.

Returns:

A decorator that wraps the target function with CAPIO setup, I/O patching and cleanup.

Example

@CapioContext(capio_dir="./data", app_name="writer")
def produce(path, payload):
    with open(path, "w") as f:
        f.write(payload)

Command-line launcher

Command-line entry point for PyCAPIO.

This module wraps an arbitrary Python script or module in a pycapio.CapioContext() so that every I/O operation performed by that program is transparently intercepted by CAPIO. No edits to the target program are required.

Usage:

pycapio --capio-dir ./dir --app-name app \
    --workflow-name my_workflow my_step.py [script args...]

pycapio.__main__.main()

Parse CLI arguments and run the target script under CAPIO interception.

The launcher accepts the following command-line arguments:

--capio-dir

CAPIO directory to manage (default ".").

--workflow-name

CAPIO workflow name (default CAPIO_DEFAULT_WORKFLOW_NAME).

--app-name

CAPIO application name (default CAPIO_DEFAULT_APP_NAME).

--capio-cl

Path to a CAPIO-CL configuration file (default "").

script_path

Path to the Python script, package directory, or importable module to execute.

script_args

Remaining arguments forwarded verbatim to the target program via sys.argv.–

The target is launched inside a pycapio.CapioContext(). Execution is attempted first as a filesystem path with runpy.run_path() (handling .py files and package directories that expose __main__.py or main.py); if that fails it is retried as an importable module with runpy.run_module(). The process exit code mirrors the target’s exit status.