flowrep.parsers.parser_helpers module

flowrep.parsers.parser_helpers.FlowControlBindings

Peers the enclosing walker must wire into a flow-control node’s generated ports.

A flow-control recipe has no room to host a peer node inside it, so a literal or an attribute chain in a condition becomes a peer of the flow-control node one level up, reaching it through a generated input port.

A ConstantRecipe value means “create this peer node and wire it to the port”; a SourceHandle means “the peer already exists in your nodes – just wire it”.

alias of dict[str, ConstantRecipe | SourceHandle]

class flowrep.parsers.parser_helpers.SignatureInfo(names: 'list[str]', have_defaults: 'list[str]', have_restricted_kinds: 'dict[str, base_models.RestrictedParamKind]')

Bases: object

have_defaults: list[str]

have_restricted_kinds: dict[str, RestrictedParamKind]

names: list[str]

classmethod of(func: LambdaType | type) → SignatureInfo

exception flowrep.parsers.parser_helpers.SourceCodeUnavailableError

Bases: ValueError

flowrep.parsers.parser_helpers.apply_label_decorator(func: _D | str | None, output_labels: tuple[str, ...], *, wrap: Callable[[_D, tuple[str, ...]], _D], decorator_name: str, allowed_types: tuple[type, ...]) → _D | Callable[[_D], _D]

flowrep.parsers.parser_helpers.consume_call_arguments(scope: SymbolScope, ast_call: Call, child: LabeledRecipe, nodes: dict[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)], Annotated[AtomicRecipe | ConstantRecipe | ForEachRecipe | IfRecipe | TryRecipe | WhileRecipe | WorkflowRecipe, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, condition_bindings: dict[str, ConstantRecipe | SourceHandle] | None = None, reserved_ports: set[str] | None = None, hoisted: dict[expr, SourceHandle] | None = None) → None

Record all argument->port consumptions for a node-creating call.

ast.Name arguments consume an existing symbol; any other argument must be a Python literal. In the default (non-condition) mode a literal is injected as a ConstantRecipe source node into nodes and wired to the consuming port. In condition mode – selected by passing a condition_bindings dict – a literal is instead routed to a synthetic flow-control input port and recorded in condition_bindings ({synthetic_port: ConstantRecipe}) for the enclosing walker to satisfy with a constant peer. A flow-control node has no room to host a constant peer inside it, so the peer lives one level up. reserved_ports carries synthetic port names already allocated for this flow-control chain so that multiple literals – across an if/elif chain – get distinct, deterministic ports.

Data-attribute arguments are injected ahead of the call by chain_parser.hoist_call_arguments and arrive here in hoisted, mapping the argument’s AST node to the SourceHandle of its outermost getattr node. Outside condition mode the consuming node is a peer of that getattr and simply reads from it. In condition mode the consumer lives inside a flow-control recipe, which has no room for a peer, so the handle is routed through a generated input port and recorded in condition_bindings – exactly as a literal is.

flowrep.parsers.parser_helpers.get_ast_function_node(func: LambdaType) → FunctionDef

flowrep.parsers.parser_helpers.get_available_source_code(func: LambdaType) → str | None

flowrep.parsers.parser_helpers.get_function_definition(tree: Module) → FunctionDef

flowrep.parsers.parser_helpers.get_source_code(func: LambdaType) → str

flowrep.parsers.parser_helpers.reject_input_alias_outputs(body_symbol_map: SymbolScope, candidate_outputs: Iterable[str], control_kind: str) → None

Raise if a flow-control body tries to surface an input-alias as an output.

A symbol aliased directly to a workflow input keeps an InputSource rather than a node SourceHandle. Such a symbol cannot be represented as a {control_kind} output, so we reject it with a clear error instead of emitting an invalid recipe.

flowrep.parsers.parser_helpers.resolve_symbols_to_strings(node: expr | None) → list[str]