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.
`_MARKER_TO_CATEGORY`

Classes¶

Info

Functions¶

`_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_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`(→ str \| None)	Collect IO markers and record trigger from an argument collection.
`_collect_ios_from_bound_value`(→ str \| None)	Collect IO markers and record trigger from a bound argument value.
`_collect_ios_from_bound_arguments`(→ 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.
`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._MARKER_TO_CATEGORY: dict[str, str]¶

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_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) → 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.

Returns:

Updated record trigger 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, prefix: str = '') → 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.
prefix – Optional field prefix for nested models.

Returns:

Updated record trigger 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) → 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.

Returns:

Updated record trigger 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][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).

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¶

argv: list[str] | None = None¶

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

The argument names passed in IOArgumentNames should match keys in the bound arguments (typically from app.parse_args() or the decorated function arguments). For example def myfunc(input: Path, output: Path) would correspond to parameter names input and output.

Parameters:

app – Root Cyclopts App instance.
tokens – Optional command arguments to use in action id.
dataset_license – Optional dataset license string. If absent ro-crate will be invalid.
crate_dir – Optional path to RO-Crate directory.
software_version – Optional software version override. Otherwise extracted from App instance.
current_user – Optional user override. Uses current system user if None.