rocrate_action_recorder.adapters.cyclopts¶

Adapter for Cyclopts CLI framework.

Attributes¶

`logger`
`INPUT_FILE`	Marker for Annotated argument that represent input file.
`INPUT_FILES`	Marker for Annotated argument that represent multiple input files.
`INPUT_DIR`	Marker for Annotated argument that represent input directory.
`INPUT_DIRS`	Marker for Annotated argument that represent multiple input directories.
`OUTPUT_FILE`	Marker for Annotated argument that represent output file.
`OUTPUT_FILES`	Marker for Annotated argument that represent multiple output files.
`OUTPUT_DIR`	Marker for Annotated argument that represent output directory.
`OUTPUT_DIRS`	Marker for Annotated argument that represent multiple output directories.
`RECORD_TRIGGER`	Marker for Annotated boolean argument that trigger recording.
`CRATE_DIR`	Marker for Annotated Path argument that specifies the RO-Crate directory.
`_MARKER_TO_CATEGORY`

Classes¶

Info

Functions¶

`_annotation_markers`() → list[Any])	Collect Annotated-style metadata from an annotation tree.
`_collect_subcommands`(→ dict[str, ...)	Recursively collect all subcommands from a Cyclopts App.
`_plaintext_doc`(→ str)	Extract plaintext description from a Cyclopts App's docstring.
`program_from_app`(→ rocrate_action_recorder.core.Program)	Extract Program information from a Cyclopts App.
`_is_pydantic_model`(→ bool)	Check if a value is a Pydantic BaseModel instance.
`_is_attrs_instance`(→ bool)	Check if a value is an attrs instance.
`_search_argument_value`(→ Any \| None)	Search recursively for a field value within parsed Cyclopts arguments.
`_lookup_argument_value`(→ Any \| None)	Resolve an argument value from Cyclopts bound arguments.
`_lookup_nested_field`(→ Any \| None)	Lookup a nested field value using dot notation.
`_resolve_crate_dir_value`(→ pathlib.Path \| None)	Resolve a CRATE_DIR-marked value to a concrete path.
`_resolve_io_argument_paths`(...)	Resolve named Cyclopts arguments to path-bearing IOArgumentPath entries.
`_collect_ioargs`(...)	Collect IOArgumentPaths from Cyclopts inputs.
`_extract_markers_from_pydantic_fields`(...)	Extract markers from Pydantic model fields.
`_extract_markers_from_attrs_fields`(→ list[tuple[str, str]])	Extract markers from attrs class fields.
`_is_pydantic_model_type`(→ bool)	Check if an annotation is a Pydantic BaseModel class.
`_is_attrs_type`(→ bool)	Check if an annotation is an attrs class.
`_collect_ios_from_argument_collection`(...)	Collect IO markers and record trigger from an argument collection.
`_collect_ios_from_bound_value`(→ tuple[str \| None, ...)	Collect IO markers and record trigger from a bound argument value.
`_collect_ios_from_bound_arguments`(→ tuple[str \| None, ...)	Collect IO markers and record trigger from parsed bound arguments.
`_detect_ios_and_trigger`(...)	Auto-detect input/output arguments and optional record trigger from Annotated metadata.
`_should_record`(→ bool)	Evaluate whether recording should occur for this invocation.
`_resolve_executed_subapp`(→ cyclopts.App \| None)	Find the app node that owns the executed default command.
`_parse_tokens`(→ list[str])
`_is_successful_system_exit`(→ bool)	Return whether a SystemExit represents a successful exit.
`_flatten_keys`(→ set[str])	Takes {"mysub": {"something": "valuefromconfig"}} and returns {"mysub.something"}.
`collect_info`(→ Info)
`record_cyclopts`(→ collections.abc.Generator[cyclopts.App])	Context manager to record a Cyclopts CLI invocation in an RO-Crate.

Module Contents¶

rocrate_action_recorder.adapters.cyclopts.logger¶

rocrate_action_recorder.adapters.cyclopts.INPUT_FILE: LiteralString = 'INPUT_FILE'¶: Marker for Annotated argument that represent input file.

rocrate_action_recorder.adapters.cyclopts.INPUT_FILES: LiteralString = 'INPUT_FILES'¶: Marker for Annotated argument that represent multiple input files.

rocrate_action_recorder.adapters.cyclopts.INPUT_DIR: LiteralString = 'INPUT_DIR'¶: Marker for Annotated argument that represent input directory.

rocrate_action_recorder.adapters.cyclopts.INPUT_DIRS: LiteralString = 'INPUT_DIRS'¶: Marker for Annotated argument that represent multiple input directories.

rocrate_action_recorder.adapters.cyclopts.OUTPUT_FILE: LiteralString = 'OUTPUT_FILE'¶: Marker for Annotated argument that represent output file.

rocrate_action_recorder.adapters.cyclopts.OUTPUT_FILES: LiteralString = 'OUTPUT_FILES'¶: Marker for Annotated argument that represent multiple output files.

rocrate_action_recorder.adapters.cyclopts.OUTPUT_DIR: LiteralString = 'OUTPUT_DIR'¶: Marker for Annotated argument that represent output directory.

rocrate_action_recorder.adapters.cyclopts.OUTPUT_DIRS: LiteralString = 'OUTPUT_DIRS'¶: Marker for Annotated argument that represent multiple output directories.

rocrate_action_recorder.adapters.cyclopts.RECORD_TRIGGER: LiteralString = 'RECORD_TRIGGER'¶: Marker for Annotated boolean argument that trigger recording.

rocrate_action_recorder.adapters.cyclopts.CRATE_DIR: LiteralString = 'CRATE_DIR'¶: Marker for Annotated Path argument that specifies the RO-Crate directory.

rocrate_action_recorder.adapters.cyclopts._MARKER_TO_CATEGORY: dict[str, str]¶

rocrate_action_recorder.adapters.cyclopts._annotation_markers(annotation: Any, extra_metadata: collections.abc.Iterable[Any] = ()) → list[Any][source]¶

Collect Annotated-style metadata from an annotation tree.

This unwraps nested Annotated values that may be wrapped inside Optional/union annotations.

rocrate_action_recorder.adapters.cyclopts._collect_subcommands(app: cyclopts.App, parent_name: str, seen: set[str] | None = None) → dict[str, rocrate_action_recorder.core.Program][source]¶

Recursively collect all subcommands from a Cyclopts App.

Parameters:

app – The Cyclopts App instance to collect from.
parent_name – The accumulated name path of parent commands.
seen – Set of app name strings to track visited apps and avoid infinite recursion.

Returns:

Dictionary mapping command names to Program objects.

rocrate_action_recorder.adapters.cyclopts._plaintext_doc(app: cyclopts.App) → str[source]¶: Extract plaintext description from a Cyclopts App’s docstring.

rocrate_action_recorder.adapters.cyclopts.program_from_app(app: cyclopts.App) → rocrate_action_recorder.core.Program[source]¶

Extract Program information from a Cyclopts App.

Parameters:: app – The Cyclopts App instance.
Returns:: Program with command and selected subcommand information.

rocrate_action_recorder.adapters.cyclopts._is_pydantic_model(value: Any) → bool[source]¶: Check if a value is a Pydantic BaseModel instance.

rocrate_action_recorder.adapters.cyclopts._is_attrs_instance(value: Any) → bool[source]¶: Check if a value is an attrs instance.

rocrate_action_recorder.adapters.cyclopts._search_argument_value(value: Any, field_name: str) → Any | None[source]¶

Search recursively for a field value within parsed Cyclopts arguments.

Parameters:

value – Value to inspect.
field_name – Field name to resolve.

Returns:

The resolved value, or None if the field cannot be found.

rocrate_action_recorder.adapters.cyclopts._lookup_argument_value(arguments: dict[str, Any], field_name: str) → Any | None[source]¶

Resolve an argument value from Cyclopts bound arguments.

Cyclopts stores grouped parameters, such as dataclass-backed option groups, under their parent parameter name in bound_args.arguments. This helper resolves both top-level and nested field values.

Parameters:

arguments – Parsed Cyclopts bound arguments.
field_name – Field name to resolve. Supports dot notation for nested fields.

Returns:

The resolved value, or None if the field cannot be found.

rocrate_action_recorder.adapters.cyclopts._lookup_nested_field(value: Any, field_path: str) → Any | None[source]¶

Lookup a nested field value using dot notation.

Parameters:

value – The value to search in (Pydantic model, dataclass, attrs, dict, etc.).
field_path – Dot-separated field path (e.g., “io.input”).

Returns:

The resolved value, or None if the field cannot be found.

rocrate_action_recorder.adapters.cyclopts._resolve_crate_dir_value(value: Any, field_name: str) → pathlib.Path | None[source]¶

Resolve a CRATE_DIR-marked value to a concrete path.

Parameters:

value – Parsed argument value.
field_name – Argument name used for developer-facing errors.

Returns:

A resolved crate directory path, or None when the value is intentionally unset.

Raises:

ValueError – If the CRATE_DIR marker is applied to a non-path-like value.

rocrate_action_recorder.adapters.cyclopts._resolve_io_argument_paths(names: list[str], name_info: dict[str, tuple[str, str]], arguments: dict[str, Any]) → list[rocrate_action_recorder.core.IOArgumentPath][source]¶

Resolve named Cyclopts arguments to path-bearing IOArgumentPath entries.

Parameters:

names – IO argument names to resolve.
name_info – Mapping from CLI argument name to bound field name and help text.
arguments – Parsed Cyclopts bound arguments.

Returns:

Resolved IO argument path entries.

rocrate_action_recorder.adapters.cyclopts._collect_ioargs(bound_args: inspect.BoundArguments, ios: rocrate_action_recorder.adapters.shared.IOArgumentNames, argument_collection: cyclopts.argument.ArgumentCollection) → rocrate_action_recorder.core.IOArgumentPaths[source]¶

Collect IOArgumentPaths from Cyclopts inputs.

Parameters:

bound_args – Parsed Cyclopts bound arguments.
ios – Parameter names that map to input/output files and directories.
argument_collection – Assembled Cyclopts argument collection (from the executed subapp).

Returns:

A IOArgumentPaths instance

rocrate_action_recorder.adapters.cyclopts._extract_markers_from_pydantic_fields(model_cls: Any, prefix: str = '') → list[tuple[str, str]][source]¶

Extract markers from Pydantic model fields.

Parameters:

model_cls – Pydantic model class.
prefix – Field name prefix for nested fields.

Returns:

List of (field_path, marker) tuples.

rocrate_action_recorder.adapters.cyclopts._extract_markers_from_attrs_fields(attrs_cls: Any, prefix: str = '') → list[tuple[str, str]][source]¶

Extract markers from attrs class fields.

Parameters:

attrs_cls – attrs class.
prefix – Field name prefix for nested fields.

Returns:

List of (field_path, marker) tuples.

rocrate_action_recorder.adapters.cyclopts._is_pydantic_model_type(ann: Any) → bool[source]¶: Check if an annotation is a Pydantic BaseModel class.

rocrate_action_recorder.adapters.cyclopts._is_attrs_type(ann: Any) → bool[source]¶: Check if an annotation is an attrs class.

rocrate_action_recorder.adapters.cyclopts._collect_ios_from_argument_collection(collection: cyclopts.argument.ArgumentCollection, ios: rocrate_action_recorder.adapters.shared.IOArgumentNames, record_trigger_name: str | None, crate_dir_name: str | None) → tuple[str | None, str | None][source]¶

Collect IO markers and record trigger from an argument collection.

Parameters:

collection – Argument collection to inspect.
ios – IO marker accumulator.
record_trigger_name – Current record trigger name, if any.
crate_dir_name – Current crate dir argument name, if any.

Returns:

Updated record trigger name and crate dir argument name.

rocrate_action_recorder.adapters.cyclopts._collect_ios_from_bound_value(value: Any, ios: rocrate_action_recorder.adapters.shared.IOArgumentNames, record_trigger_name: str | None, crate_dir_name: str | None, prefix: str = '') → tuple[str | None, str | None][source]¶

Collect IO markers and record trigger from a bound argument value.

Parameters:

value – Bound argument value to inspect.
ios – IO marker accumulator.
record_trigger_name – Current record trigger name, if any.
crate_dir_name – Current crate dir argument name, if any.
prefix – Optional field prefix for nested models.

Returns:

Updated record trigger name and crate dir argument name.

rocrate_action_recorder.adapters.cyclopts._collect_ios_from_bound_arguments(bound_args: inspect.BoundArguments, ios: rocrate_action_recorder.adapters.shared.IOArgumentNames, record_trigger_name: str | None, crate_dir_name: str | None) → tuple[str | None, str | None][source]¶

Collect IO markers and record trigger from parsed bound arguments.

Parameters:

bound_args – Parsed Cyclopts bound arguments.
ios – IO marker accumulator.
record_trigger_name – Current record trigger name, if any.
crate_dir_name – Current crate dir argument name, if any.

Returns:

Updated record trigger name and crate dir argument name.

rocrate_action_recorder.adapters.cyclopts._detect_ios_and_trigger(argument_collection: cyclopts.argument.ArgumentCollection, command: Any | None = None, meta_argument_collection: cyclopts.argument.ArgumentCollection | None = None, bound_args: inspect.BoundArguments | None = None) → tuple[rocrate_action_recorder.adapters.shared.IOArgumentNames, str | None, str | None][source]¶

Auto-detect input/output arguments and optional record trigger from Annotated metadata.

Parameters:

argument_collection – Assembled Cyclopts argument collection.
command – The command function to check for *args.
meta_argument_collection – Optional meta app argument collection to check for triggers.

Returns:

A tuple of (IOArgumentNames, trigger_arg_name_or_None, crate_dir_arg_name_or_None).

rocrate_action_recorder.adapters.cyclopts._should_record(bound_args: inspect.BoundArguments, record_trigger_name: str | None) → bool[source]¶

Evaluate whether recording should occur for this invocation.

Parameters:

bound_args – Parsed Cyclopts bound arguments.
record_trigger_name – Optional trigger argument name.

Returns:

True when recording is enabled, otherwise False.

rocrate_action_recorder.adapters.cyclopts._resolve_executed_subapp(app: cyclopts.App, command: Any) → cyclopts.App | None[source]¶

Find the app node that owns the executed default command.

Parameters:

app – App node to inspect.
command – Resolved callable returned by parse_args.

Returns:

Matching app instance, or None when no match is found.

rocrate_action_recorder.adapters.cyclopts._parse_tokens(tokens: str | collections.abc.Iterable[str] | None = None) → list[str][source]¶

rocrate_action_recorder.adapters.cyclopts._is_successful_system_exit(error: SystemExit) → bool[source]¶: Return whether a SystemExit represents a successful exit.

class rocrate_action_recorder.adapters.cyclopts.Info[source]¶

program: rocrate_action_recorder.core.Program¶

ioargs: rocrate_action_recorder.core.IOArgumentPaths¶

should_record: bool¶

crate_dir: pathlib.Path | None = None¶

argv: list[str] | None = None¶

rocrate_action_recorder.adapters.cyclopts._flatten_keys(d: dict, separator='.') → set[str][source]¶: Takes {“mysub”: {“something”: “valuefromconfig”}} and returns {“mysub.something”}.

rocrate_action_recorder.adapters.cyclopts.collect_info(app: cyclopts.App, tokens: str | collections.abc.Iterable[str] | None = None, software_version: str | None = None) → Info[source]¶

rocrate_action_recorder.adapters.cyclopts.record_cyclopts(app: cyclopts.App, tokens: str | collections.abc.Iterable[str] | None = None, dataset_license: str | None = None, crate_dir: pathlib.Path | None = None, software_version: str | None = None, current_user: str | None = None) → collections.abc.Generator[cyclopts.App]¶

Context manager to record a Cyclopts CLI invocation in an RO-Crate.

Hint

Marker metadata (for example INPUT_FILE and OUTPUT_FILE) is auto-detected from typing.Annotated parameters and nested models. Recording writes ro-crate-metadata.json only when execution finishes successfully and recording is enabled.

Examples

Full runnable example: https://github.com/i-VRESSE/rocrate-action-recorder/tree/main/example/cyclopts

Basic positional input/output tracking:

from pathlib import Path
from typing import Annotated

from cyclopts import App
from rocrate_action_recorder.adapters.cyclopts import (
    INPUT_FILE,
    OUTPUT_FILE,
    record_cyclopts,
)

app = App(version="1.2.3")

@app.default
def main(
    input: Annotated[Path, INPUT_FILE],
    output: Annotated[Path, OUTPUT_FILE],
    /,
):
    output.write_text(input.read_text().upper())

# Call as: myscript.py input.txt output.txt
with record_cyclopts(app, dataset_license="CC-BY-4.0"):
    app()

Toggle recording with a boolean trigger flag:

from typing import Annotated

from cyclopts import App, Parameter
from rocrate_action_recorder.adapters.cyclopts import RECORD_TRIGGER, record_cyclopts

app = App()

@app.default
def main(*, prov: Annotated[bool, Parameter(negative=""), RECORD_TRIGGER] = False):
    pass

# Records only when --prov is passed.
# Call as: myscript.py --prov
with record_cyclopts(app):
    app()

Track nested configuration fields (for example dataclass or Pydantic models):

from dataclasses import dataclass
from pathlib import Path
from typing import Annotated

from cyclopts import App
from rocrate_action_recorder.adapters.cyclopts import INPUT_FILE, OUTPUT_FILE, record_cyclopts

@dataclass
class IO:
    input: Annotated[Path, INPUT_FILE]
    output: Annotated[Path, OUTPUT_FILE]

app = App()

@app.default
def main(io: IO):
    io.output.write_text(io.input.read_text().upper())

# Call as: myscript.py input.txt output.txt
with record_cyclopts(app):
    app()

Record multiple inputs/outputs from list markers:

from pathlib import Path
from typing import Annotated

from cyclopts import App
from rocrate_action_recorder.adapters.cyclopts import (
    INPUT_FILES,
    OUTPUT_FILES,
    record_cyclopts,
)

app = App()

@app.default
def main(
    *,
    inputs: Annotated[list[Path], INPUT_FILES],
    outputs: Annotated[list[Path], OUTPUT_FILES],
):
    for src, dst in zip(inputs, outputs, strict=True):
        dst.write_text(src.read_text().upper())

# Call as: myscript.py --inputs in1.txt --inputs in2.txt --outputs out1.txt --outputs out2.txt
with record_cyclopts(app):
    app()

Let CLI arguments choose the crate destination using CRATE_DIR:

from pathlib import Path
from typing import Annotated

from cyclopts import App
from rocrate_action_recorder.adapters.cyclopts import CRATE_DIR, record_cyclopts

app = App()

@app.default
def main(*, session_dir: Annotated[Path, CRATE_DIR]):
    pass

# The CRATE_DIR-marked value overrides record_cyclopts(crate_dir=...).
# Call as: myscript.py --session-dir ./runs/session-001
with record_cyclopts(app, crate_dir=Path("fallback-crate")):
    app()

Subcommands are supported; wrap the root app once and invoke normally:

app = App(name="tool")
process = App(name="process")

@process.default
def run(...):
    ...

app.command(process)
# Call as: tool process
with record_cyclopts(app):
    app()

Use config file to override default prov trigger value:

from typing import Annotated

from cyclopts import App, Parameter
from cyclopts.config import Toml
from rocrate_action_recorder.adapters.cyclopts import RECORD_TRIGGER, record_cyclopts

app = App(config=Toml("myconfig.toml"))

@app.default
def main(*, prov: Annotated[bool, Parameter(negative=""), RECORD_TRIGGER] = False):
    pass

with record_cyclopts(app):
    app()

Create a myconfig.toml file containing:

prov = true

When script is called then provenance is always recorded.

Parameters:

app – Root Cyclopts App instance.
tokens – Optional command arguments used for parsing and action id generation.
dataset_license – Optional dataset license string. If absent ro-crate will be invalid.
crate_dir – Optional path to RO-Crate directory. If a parsed argument is annotated with CRATE_DIR, that value takes precedence.
software_version – Optional software version override. Otherwise extracted from App instance.
current_user – Optional user override. Uses current system user if None.