flowrep.parsers.case_helpers module

class flowrep.parsers.case_helpers.WalkedBranch(label: 'str', walker: 'parser_protocol.BodyWalker', assigned: 'list[str]')

Bases: object

assigned: list[str]

label: str

to_labeled_node() → LabeledRecipe

walker: BodyWalker

flowrep.parsers.case_helpers.parse_case(test: expr, scope: ScopeProxy, symbol_map: SymbolScope, info_factory: VersionInfoFactory, label: str, 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')]], reserved_ports: set[str] | None = None) → tuple[LabeledRecipe, dict[TargetHandle, InputSource], dict[str, ConstantRecipe | SourceHandle]]

Parse a conditional expression.

Validates that the statement is a function call returning exactly one value. Returns the labeled condition node, and the input edges neeeded to feed it.

flowrep.parsers.case_helpers.walk_branch(walker: BodyWalker, label: str, stmts: list[stmt]) → WalkedBranch

Fork a walker and walk a conditional branch body.

Both the SymbolScope and the ScopeProxy are forked so that symbol assignments and import-based scope extensions in one branch do not leak into sibling or parent branches.

flowrep.parsers.case_helpers.wire_inputs(branches: list[WalkedBranch]) → tuple[list[str], dict[TargetHandle, InputSource]]

Collect input edges from the condition and body branches.

flowrep.parsers.case_helpers.wire_outputs(branches: list[WalkedBranch]) → tuple[list[str], dict[OutputTarget, list[SourceHandle]]]

Collect outputs and prospective output edges from try and except bodies.