flowrep.compiler.function module

class flowrep.compiler.function.Emitter(namespace: 'dict[str, Any]' = <factory>, nested_defs: 'list[str]' = <factory>, module_imports: 'set[str]' = <factory>, module_names: 'NameAllocator' = <factory>, workflow_decorator: 'tuple[str, str]' = ('not', 'useful'))[source]

Bases: object

property decorator_string

module_imports: set[str]

module_names: NameAllocator

namespace: dict[str, Any]

nested_defs: list[str]

workflow_decorator: tuple[str, str] = ('not', 'useful')

class flowrep.compiler.function.FunctionBuilder(name: 'str', decorator: 'str', params: 'list[str]' = <factory>, body_lines: 'list[str]' = <factory>, return_annotation: 'str | None' = None, return_symbols: 'list[str]' = <factory>, output_labels: 'list[str]' = <factory>)[source]

Bases: object

body_lines: list[str]

decorator: str

name: str

output_labels: list[str]

params: list[str]

render() → str[source]

return_annotation: str | None = None

return_symbols: list[str]

class flowrep.compiler.function.NameAllocator[source]

Bases: object

Mints unique, valid Python identifiers within a single function namespace.

fresh(hint: str) → str[source]

reserve(name: str) → str[source]

flowrep.compiler.function.build_signature(inputs: MutableMapping[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)], InputDataPort], outputs: MutableMapping[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)], OutputDataPort]) → Signature[source]

flowrep.compiler.function.emit_nested_workflow_node(node: WorkflowRecipe, label: str, emitter: Emitter, alloc: NameAllocator) → str[source]

Emit a nested @flowrep.workflow-decorated def for a reference-free workflow node.

Returns the local function name (the call path for the enclosing body). The nested function is appended to emitter.nested_defs.

The function is named after the label’s base (pre-suffix) form so that workflow_parser.parse_workflow re-derives the same node label on round-trip via label_helpers.unique_suffix.

flowrep.compiler.function.emit_workflow_function(recipe: WorkflowRecipe, name: str, emitter: Emitter, signature: Signature | None = None) → FunctionBuilder[source]

Yield the top-level module binding of every import the recipe will emit.

Mirrors the structure emission walks: referenced nodes (atomic or referenced workflow) emit a dotted call and are not recursed; reference-free workflows and flow controls are recursed via .nodes / .prospective_nodes. Builtin exception types are skipped, matching _exception_name.