from stringdale import V,E,Define,ScopeStringdale
Defining Diagrams
Define
Signature:
Define(diagram_name, type: str = 'flow', state: pydantic.main.BaseModel = <class 'stringdale.base.BaseModelExtra'>, validate: bool = True)
Define a new diagram using a context manager.
| Parameter | Type | Default | Description |
|---|---|---|---|
| diagram_name | None | None | Name for the new diagram |
| type | <class ‘str’> | flow | The type of diagram to create, either ‘flow’ or ‘decision’. Defaults to ‘flow’ |
| state | <class ‘pydantic.main.BaseModel’> | <class ‘stringdale.base.BaseModelExtra’> | The state class to use for the diagram. Defaults to BaseModelExtra |
| validate | <class ‘bool’> | True | If True, validates diagram structure after definition. Defaults to True |
| :Returns: | None | - | The created diagram object |
V
Signature:
V(name: str, func: Callable = None, inputs: Any = None, outputs=None, is_break: bool = False, for_each: Optional[List[str]] = None, filter: bool = False, flat: bool = False, as_start: bool = False, as_end: bool = False) -> None
Add a vertex (node) to the current diagram.
| Parameter | Type | Default | Description |
|---|---|---|---|
| name | <class ‘str’> | None | Name of the node |
| func | typing.Callable | None | Function to execute at this node. If None, node acts as a passthrough |
| inputs | typing.Any | None | List of input edge descriptors. Each descriptor can be either:- A string in format “source_node.source_port->target_port” - A tuple (edge_descriptor, condition_func) for conditional edges |
| outputs | List[str] | None | List of output edge descriptors. Each descriptor can be either:- A string in format “source_port->target_node.target_port”- A tuple (edge_descriptor, condition_func) for conditional edges |
| is_break | <class ‘bool’> | False | If True, execution will pause at this node. Not allowed in flow scopes |
| for_each | typing.Optional[typing.List[str]] | None | List of input keys to iterate over. Used for batching operations in Flow diagrams.If provided, the node will be executed once for each product of items in the input list.This for each keys must get iteratbles from the input edges. |
| filter | <class ‘bool’> | False | Used for batching operations in Flow diagrams. If True, falsy node outputs will be filtered out. Cannot be used with flat=True |
| flat | <class ‘bool’> | False | Used for batching operations in Flow diagrams.If True, node output lists will be flattened into a single list.Cannot be used with filter=True |
| as_start | <class ‘bool’> | False | If True, marks this node as the diagram’s start node |
| as_end | <class ‘bool’> | False | If True, marks this node as the diagram’s end node |
| :Returns: | None | - | Name of the created node |
E
Signature:
E(edge_string: str, cond: Callable = None, type: str = None) -> None
Add an edge to the current diagram
| Parameter | Type | Default | Description |
|---|---|---|---|
| edge_string | <class ‘str’> | None | The edge descriptor string |
| cond | typing.Callable | None | The condition of the edge |
| type | <class ‘str’> | None | The type of the edge, either ‘flow’ or ‘decision’, by default, the type is determined by the current scope |
| :Returns: | None | - |
Scope
Signature:
Scope(scope: str)
Context manager for defining a flow/decision scope in a diagram of a different type
| Parameter | Type | Default | Description |
|---|---|---|---|
| scope | <class ‘str’> | None | The scope type to start. Can be either a string (‘flow’ or ‘decision’). |
DiagramSchema
from stringdale import DiagramSchemaDiagramSchema
Signature:
DiagramSchema(graph=None, state_class=<class 'stringdale.base.BaseModelExtra'>, factored_graph=None, start_node=None, end_node=None, name=None, type: stringdale.base.DiagramType = None, anon=False, derive_state=False)
None
| Parameter | Type | Default | Description |
|---|
__call__
None > Signature: DiagramSchema.__call__(self: stringdale.base.DiagramSchema, **kwargs)
| Parameter | Type | Default | Description |
|---|
draw
Draw a DiagramSchema using graphviz.
Signature:
DiagramSchema.draw(self: stringdale.base.DiagramSchema, return_dot=False, direction='LR', recursive: Union[bool, List[str]] = False, factored=False, **kwargs)
| Parameter | Type | Default | Description |
|---|---|---|---|
| diagram | None | None | Either a diagram object or a diagram scheme object |
| name | None | None | If provided, uses this name for the diagram in the Mermaid title |
| return_dot | None | False | If True, returns the graphviz dot object |
| direction | None | LR | direction to draw, either TB (top to bottom) or LR (left to right), defaults to TB |
| recursive | typing.Union[bool, typing.List[str]] | False | Whether to draw subdiagrams as well. If False, only the top level diagram is drawn.If True, all subdiagrams are drawn.If a list of strings, only the subdiagrams with whose names the regex strings are drawn. |
| factored | None | False | If True, draws the factored graph, used for debugging |
Running diagrams
from stringdale import DiagramDiagram
Signature:
Diagram(graph, funcs, type, schema: stringdale.base.DiagramSchema, state: pydantic.main.BaseModel = None, anon=False, root=None)
An instance of a stringdale diagram. Instantiated by calling the Schema()
Has the following public attributes: output - the output of the last run finished - whether the diagram has reached the End node state - the current state of the diagram
| Parameter | Type | Default | Description |
|---|
run_all
Run the diagram to completion and return the final output.
Signature:
Diagram.run_all(self: stringdale.base.Diagram, input: Any, state: Union[pydantic.main.BaseModel, Dict] = None, progress_bars: bool = True, trace_nested: bool = True)
| Parameter | Type | Default | Description |
|---|---|---|---|
| input | typing.Any | None | The input data to process through the diagram |
| state | typing.Union[pydantic.main.BaseModel, typing.Dict] | None | Optional state to initialize the diagram with |
| progress_bars | <class ‘bool’> | True | Whether to display progress bars during execution (default True). Deprecated. |
| trace_nested | <class ‘bool’> | True | Whether to trace nested diagram execution (default True) |
run
Run the diagram with the given input and state.
Signature:
Diagram.run(self: stringdale.base.Diagram, input: Any, state: Union[pydantic.main.BaseModel, Dict] = None, progress_bars: bool = True, trace_nested: bool = True)
| Parameter | Type | Default | Description |
|---|---|---|---|
| input | typing.Any | None | The input data to process through the diagram |
| state | typing.Union[pydantic.main.BaseModel, typing.Dict] | None | Optional state to initialize the diagram with |
| progress_bars | <class ‘bool’> | True | Whether to display progress bars during execution (default True). Deprecated. |
| trace_nested | <class ‘bool’> | True | Whether to trace nested diagram execution (default True) |
arun
Asynchronously run the diagram with the given input and state.
Signature:
Diagram.arun(self: stringdale.base.Diagram, input: Any, state: Union[pydantic.main.BaseModel, Dict] = None, progress_bars: bool = True, trace_nested: bool = True)
| Parameter | Type | Default | Description |
|---|---|---|---|
| input | typing.Any | None | The input data to process through the diagram |
| state | typing.Union[pydantic.main.BaseModel, typing.Dict] | None | Optional state to initialize the diagram with |
| progress_bars | <class ‘bool’> | True | Whether to display progress bars during execution (default True). Deprecated. |
| trace_nested | <class ‘bool’> | True | Whether to trace nested diagram execution (default True) |
__getitem__
None > Signature: Diagram.__getitem__(self, key)
| Parameter | Type | Default | Description |
|---|
Condition
from stringdale import ConditionCondition
Signature:
Condition(func: Union[Callable, str], mapping: Optional[str] = None, name: Optional[str] = None)
A utility function for creating condition functions using stringdale’s port mapping logic.
| Parameter | Type | Default | Description |
|---|---|---|---|
| func | typing.Union[typing.Callable, str] | None | A function or string to match against the input. If string, it will be interpreted as a regex pattern to match the input against. |
| mapping | typing.Optional[str] | None | A mapping of the input to the function. by default, the input is the first argument of the function. |
| name | typing.Optional[str] | None | A name for the condition. by default, is func_name->mapping_string |
Example
def is_module_2(x):
return x%2==0
is_input_even = Condition(is_module_2,'(x=input)',name='input is even')
assert is_input_even({"input": 2})
assert not is_input_even({"input": 3})def is_sum_42(x,y):
return x+y==42
is_io_42 = Condition(is_sum_42,'(x=input,y=output)',name='i+o=42?')
assert is_io_42({"input": 2,'output':40})
# missing inputs results in False
assert not is_io_42({"input": 2,})
assert not is_io_42({"input": 2,'output':39})C = Condition
complex_cond = (
C("^3",'(0=input)',name='starts with 3') &
C("^5",'(0=output)',name='starts with 5') |
C("^2",'(0=input)',name='starts with 2'))
complex_cond((starts with 3 & starts with 5) | starts with 2)assert complex_cond({"input": "345", "output": "567"})
assert complex_cond({"input": "244", "output": "567"})
assert not complex_cond({"input": "145", "soutput": "567"})
assert not complex_cond({"input": "345", "output": "67"})
assert not complex_cond({"not_input": "145"})
assert complex_cond({ "input": "244"})StructureJson
from stringdale import StructureJsonStructureJson
Signature:
StructureJson(*assignments)
A class for restructuring JSON objects by nested paths This class allows restructuring JSON objects by specifying path assignments in the form ‘target_path=source_path’. It parses these assignments and creates a new JSON object with the specified structure.
| Parameter | Type | Default | Description |
|---|---|---|---|
| assignments | None | None | Variable number of strings specifying path assignments in the format’target_path=source_path’ |
| :Returns: | None | - | A new JSON object with the specified structure. |
__call__
None > Signature: StructureJson.__call__(self, **kwargs)
| Parameter | Type | Default | Description |
|---|---|---|---|
| :Returns: | None | - | A new JSON object with the specified structure. |
Example
source_obj = {'d':{'e':{'f':'foo'},'x':'bar'}}
StructureJson('a.b.c=father.d.e.f','a.b.d=father.d.x')(father=source_obj){'a': {'b': {'c': 'foo', 'd': 'bar'}}}JsonRenderer
from stringdale import JsonRendererJsonRenderer
Signature:
JsonRenderer(json_obj, **kwargs)
A class for rendering JSON objects with nested jinja2 templates.
Allows setting template variables both during init and when calling the object.
| Parameter | Type | Default | Description |
|---|---|---|---|
| json_obj | None | None | A JSON object to render. |
| **kwargs | None | None | Context variables to use in the rendering. |
| :Returns: | None | - | A rendered JSON object. |
__call__
None > Signature: JsonRenderer.__call__(self, **kwargs)
| Parameter | Type | Default | Description |
|---|---|---|---|
| :Returns: | None | - | A rendered JSON object. |
Example
template_json = {
'name': '{{name}}',
'age': '{{age}}',
'city_path': '{%for city in cities%} {{city}} - {%endfor%}'
}
rend = JsonRenderer(template_json,age=30,cities=['SF','LA','NY'])
rendJsonRenderer(missing_keys={'name'})rend(name='Bob'){'name': 'Bob', 'age': '30', 'city_path': 'SF - LA - NY -'}