- [Home](https://flows-sdk.hyperscience.ai/index.html)
- Source Documentation

* * *

# Source Documentation [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#source-documentation "Link to this heading")

## blocks.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.blocks "Link to this heading")

_class_ flows\_sdk.blocks.Block( _identifier_, _reference\_name=None_, _input=None_, _title=None_, _description=None_, _error\_handling=None_, _compatibility\_spec=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "Link to this definition")

Bases: `object`

Base structure representing an executable within a Hyperscience deployment.
Available blocks depend on the version of the underlying Hyperscience platform,
but additional blocks may have also been manually installed. Check out [IDP Library](https://flows-sdk.hyperscience.ai/pages/idp.html) for
example blocks used for the Document Processing flow that comes with V32.

Parameters:

- **reference\_name** (`Optional`\[`str`\]) – unique identifier on a per-Flow basis, used to identify the concrete
block within a Flow (e.g., in order to reference the outputs of a concrete block)

- **identifier** (`str`) – block implementation that will be used at runtime
(e.g., MACHINE\_CLASSIFICATION). Multiple blocks with same identifier
can be present in a Flow.

- **input** (`Optional`\[`Dict`\[`str`, `Any`\]\]) –

key-value dict being passed during execution of the Block.
Can both have static values or various dynamic values (e.g., output from a previously
executed block, see [`flows_sdk.blocks.Block.output()`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block.output "flows_sdk.blocks.Block.output")). Most blocks have mandatory
inputs, depending on their identifier. For example:

```
{
      'a': 42,
      'b': 'foo',
      'c': some_previous_block.output(),
      'd': another_previous_block.output('nested.path')
}
```

- **title** (`Optional`\[`str`\]) – UI-visible title of the block.

- **description** (`Optional`\[`str`\]) – Description of the block. Both useful for documentation purpose and visible
to users in the Flow studio.

- **error\_handling** (`Optional`\[ [`ErrorHandling`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorHandling "flows_sdk.error_handling.ErrorHandling")\]) – The error handling policy that will be applied to this block.

- **compatibility\_spec** (`Optional`\[ [`CompatibilitySpec`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec "flows_sdk.types.CompatibilitySpec")\]) – What flows or blocks can this block be changed with on the UI.
“filter\_roles” param takes a list of strings that represent roles.
“filter\_schema” param is a JSON schema, that allows for more intricate filtering.
“label” represents the UI label in FlowStudio

output( _key=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block.output "Link to this definition")

Used to reference the output of a block that has already been executed.
For example, if we have a Flow with two blocks, the second can reference
the output of the first as part of its inputs.

Parameters:

**key** ( _Optional_ _\[_ _str_ _\]_) – When not provided, the entire output of the referenced block will be
passed along. When provided, will pass a concrete key from the output
(e.g. for an output like `{"a": 42}`, calling `output()` will pass it
as a dictionary, while calling `output("a")` will pass `42`.)

Returns:

a string-formatted reference that will be unpacked to a value during runtime.

Return type:

str

For example:

```
a = Block(...)
b = Block(
    ...,
    input = {
        'foo': a.output('bar')
    }
)
```

with\_error\_handling( _error\_handling_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block.with_error_handling "Link to this definition")

Adds the provided error handling section to this block.

Available in v36 and later.

Return type:

[`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

_class_ flows\_sdk.blocks.BaseCodeBlock( _code_, _identifier_, _reference\_name=None_, _input=None_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

TEMPLATE\_FUNCTION _='fromtypingimportAny,Dict,List,Tuple\\nfromblocks.base\_python\_blockimportWFEngineTaskResult\\nfromblocks.typesimportBlockInputs\\n{imports}\\n\\n#UsedtodeserializejsoninputsbytheCCB\\nclassCustomCodeBlockProxyInputs(BlockInputs):\\n{block\_inputs\_schema}\\n\\n{source}\\n\\n{function\_invocation}\\n'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock.TEMPLATE_FUNCTION "Link to this definition")TEMPLATE\_MODULE _='{source}\\n\\nfromblocks.base\_python\_blockimportWFEngineTaskResult\\nfromblocks.typesimportBlockInputs\\n{imports}\\n\\n#UsedtodeserializejsoninputsbytheCCB\\nclassCustomCodeBlockProxyInputs(BlockInputs):\\n{block\_inputs\_schema}\\n\\n{function\_invocation}\\n'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock.TEMPLATE_MODULE "Link to this definition")output( _key=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock.output "Link to this definition")

CodeBlock nests the actual result returned from the provided function under
the ‘result’ key. This method overrides the default output, removing the need
to always prepend ‘result’ when calling CodeBlock.output(…)

Parameters:

**key** (`Optional`\[`str`\]) –

Optionally provide a key to directly get a nested property.
When not provided, the entire ‘result’ will be returned.

For example, we have:

```
{
    "result": {
        "a": {
            "b": 42
        }
    }
}
```

- skipping the key ```output()` will return ``{"a": {"b": 42}}```

- calling with `output("a")` will result in `{"b": 42}`

- calling with `output("a.b")` will return `42`

Return type:

`str`

Returns:

the value under the provided key

_class_ flows\_sdk.blocks.PythonBlock( _code_, _reference\_name=None_, _include\_module=False_, _code\_input=None_, _input=None_, _title=None_, _description=None_, _python\_version=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock "Link to this definition")

Bases: [`BaseCodeBlock`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock "flows_sdk.blocks.BaseCodeBlock")

Python Blocks are a special type of block that can run custom Python code on
the Hyperscience Platform. Code will be serialized, then, as part of a flow run,
executed on an isolated environment. By default, users can only utilize packages that
are part of the Python Standard Library. To access third-party Python packages,
users need to follow these
[instructions](https://flows-sdk.hyperscience.ai/pages/python_packages.html#managing-third-party-python-packages)
to have them installed on the Hyperscience Platform.

Lambda functions are executed in the context of the execution engine itself
(less overhead, but cannot be scaled)
Code functions are executed in a dedicated container (more overhead, can scale horizontally)

**Module-Backed Code Blocks:**

By setting `include_module=True`, the entire module containing the code function will be
serialized and made available at runtime. This allows you to:

- Organize complex logic into multiple classes and helper functions

- Improve code maintainability and testability by separating concerns

- Reuse code across your CCB implementation

When testing module-backed CCBs, use the mock objects from [`flows_sdk.mocks`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#module-flows_sdk.mocks "flows_sdk.mocks") to simulate
the Hyperscience runtime environment.
See the [Unit Testing Custom Code Blocks](https://flows-sdk.hyperscience.ai/pages/testing.html#unit-testing-custom-code-blocks)
guide for comprehensive testing strategies.

Caution

When using `include_module=True` you need to make sure that the module imports
everything it needs to run. Boilerplate imports will NOT be automatically included
the way they are when the code is only a function.

**Python Versions:**

Caution

When using HsBlockInstance.log(), its behaviour will vary based on the version of
the python block:

PYTHON\_3\_12 and below will not support formatting arguments
and using brackets {} in the log message will result in an error.

PYTHON\_3\_13 and above will support logging arguments.
For more details, see [`flows_sdk.types.HsBlockInstance.log()`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.log "flows_sdk.types.HsBlockInstance.log")

The PythonBlock can be compiled to blocks that run on different Python versions. There are
several ways to modify the behavior.

By default, the Python version of the PythonBlock will be set based on the Python version of
the current Python virtual environment. For example, if the current virtual environment uses
Python 3.11, the PythonBlock will be compiled to a Python 3.11 block. If the version is not
supported, it would default to Python 3.9 block.

To override the default behavior controlled by the virtual environment, an environment
variable - HS\_PYTHON\_CODE\_BLOCK\_PYTHON\_VERSION \- can be set. All blocks would be compiled to
use the Python version dictated by that variable. It can be set to 3.9, 3.11, 3.12 or
3.13

To enable more fine-grained control over the Python version of the block, python\_version
parameter can be explicitly set to each PythonBlock instance using one of the available values
in the [`flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion "flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion") enum.

- **Python Block version modifier precedence**

> virtual env version < HS\_PYTHON\_CODE\_BLOCK\_PYTHON\_VERSION env var <
> specific python\_version argument to PythonBlock

Python version examples:

Default behaviour, given a virtual environment based off Python 3.9:

```
PythonBlock(
    reference_name='ref_name', code=lambda x, y: x + y, code_input={'x': 0, 'y': 1}
)

produces:

{
    "identifier": "PYTHON_CODE",
    "reference_name": "ref_name",
    "input": {"data": {"x": 0, "y": 1}, "code": "lambda x, y: x + y"}
}
```

Default behaviour, given a virtual environment based off Python 3.11:

```
PythonBlock(
    reference_name='ref_name', code=lambda x, y: x + y, code_input={'x': 0, 'y': 1}
)

produces:

{
    "identifier": "PYTHON_3_11_CODE",
    "reference_name": "ref_name",
    "input": {"data": {"x": 0, "y": 1}, "code": "lambda x, y: x + y"}
}
```

HS\_PYTHON\_CODE\_BLOCK\_PYTHON\_VERSION is set to 3.11. Virtual environment Python version
is ignored:

```
PythonBlock(
    reference_name='ref_name', code=lambda x, y: x + y, code_input={'x': 0, 'y': 1}
)

produces:

{
    "identifier": "PYTHON_3_11_CODE",
    "reference_name": "ref_name",
    "input": {"data": {"x": 0, "y": 1}, "code": "lambda x, y: x + y"}
}
```

HS\_PYTHON\_CODE\_BLOCK\_PYTHON\_VERSION is set to 3.9, but python\_version is set
to 3.11 in the block instance. Virtual environment Python version is ignored:

```
PythonBlock(
    reference_name='ref_name',
    code=lambda x, y: x + y,
    code_input={'x': 0, 'y': 1},
    python_version=PythonBlock.SupportedPythonBlockPythonVersion.PYTHON_3_11
)

produces:

{
    "identifier": "PYTHON_3_11_CODE",
    "reference_name": "ref_name",
    "input": {"data": {"x": 0, "y": 1}, "code": "lambda x, y: x + y"}
}
```

Example usage with a code function:

```
def code_fn(a_static_input: int, a_dynamic_input: str) -> str:
    import regex # third party python package
    return 'found' if regex.search(r'Hello|Hi', 'HelloWorld') else 'not found'

function_ccb = PythonBlock(
    reference_name='example_function_ccb',
    code=code_fn,
    code_input={
        'a_static_input': 42,
        'a_dynamic_input': some_previous_block.output('path.to.value')
    },
)
```

Example usage with a lambda:

```
lambda_ccb = PythonBlock(
    reference_name='example_lambda_ccb',
    code=lambda a_static_input, a_dynamic_input: {
        'foo': f'prefix_{a_dynamic_input}',
        'bar': 5 + a_static_input
    },
    code_input={
        'a_static_input': 42,
        'a_dynamic_input': some_previous_block.output('path.to.value')
    },
)
```

Example usage with a module:

```
# In my_ccb_module.py:
class DataProcessor:
    def __init__(self, prefix: str):
        self.prefix = prefix

def process(self, value: str) -> str:
        return f"{self.prefix}_{value}"

def entrypoint(input_value: str, prefix: str) -> dict:
    processor = DataProcessor(prefix)
    result = processor.process(input_value)
    return {"processed_value": result}

# In your flow definition:
from my_ccb_module import entrypoint

module_ccb = PythonBlock(
    reference_name='example_module_ccb',
    code=entrypoint,
    code_input={
        'input_value': 'hello',
        'prefix': 'output'
    },
    include_module=True  # This includes the entire module
)
```

For testing your custom code blocks, see [Unit Testing Custom Code Blocks](https://flows-sdk.hyperscience.ai/pages/testing.html#unit-testing-custom-code-blocks).

_class_ SupportedPythonBlockPythonVersion( _\*values_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion "Link to this definition")

Bases: `str`, `Enum`

Describes which versions of Pyhon can be passed to the python\_version
argument of PythonCode.

PYTHON\_3\_9 _='PYTHON\_CODE'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion.PYTHON_3_9 "Link to this definition")

**DEPRECATED** from version 40+ due to end-of-life of Python 3.9.

**NOT SUPPORTED** in version 43+.

PYTHON\_3\_11 _='PYTHON\_3\_11\_CODE'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion.PYTHON_3_11 "Link to this definition")

Available since version 39.1.

**DEPRECATED** from version 43+.

PYTHON\_3\_12 _='PYTHON\_3\_12\_CODE'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion.PYTHON_3_12 "Link to this definition")

Available since version 41.0

PYTHON\_3\_13 _='PYTHON\_3\_13\_CODE'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.SupportedPythonBlockPythonVersion.PYTHON_3_13 "Link to this definition")

Available since version 42.2

PYTHON\_BLOCK\_IDENTIFIERS _=\['PYTHON\_CODE','PYTHON\_3\_11\_CODE','PYTHON\_3\_12\_CODE','PYTHON\_3\_13\_CODE'\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock.PYTHON_BLOCK_IDENTIFIERS "Link to this definition")_class_ flows\_sdk.blocks.CodeBlock( _code_, _reference\_name=None_, _include\_module=False_, _code\_input=None_, _input=None_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.CodeBlock "Link to this definition")

Bases: [`BaseCodeBlock`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.BaseCodeBlock "flows_sdk.blocks.BaseCodeBlock")

DEPRECATED from 34.0.1+ going forward, please use [`flows_sdk.blocks.PythonBlock`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock "flows_sdk.blocks.PythonBlock")
instead

Code Blocks are a special type of block that can run custom Python code on
the Hyperscience Platform. Code will be serialized, then, as part of a flow run,
executed on an isolated environment. It will not have access
to dependencies outside of basic ones like the Python Standard Library.

**Module Support:**

The `include_module=True` parameter allows serializing the entire module containing
your code function. See [`flows_sdk.blocks.PythonBlock`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.PythonBlock "flows_sdk.blocks.PythonBlock") for detailed documentation.

Example usage with a code function:

```
def code_fn(a_static_input: int, a_dynamic_input: str) -> str:
    return f'Hello {code_block_input_param}'

function_ccb = CodeBlock(
    reference_name='example_function_ccb',
    code=code_fn,
    code_input={
        'a_static_input': 42,
        'a_dynamic_input': some_previous_block.output('path.to.value')
    },
)
```

Example usage with a lambda:

```
lambda_ccb = CodeBlock(
    reference_name='example_lambda_ccb',
    code=lambda a_static_input, a_dynamic_input: {
        'foo': f'prefix_{a_dynamic_input}',
        'bar': 5 + a_static_input
    },
    code_input={
        'a_static_input': 42,
        'a_dynamic_input': some_previous_block.output('path.to.value')
    },
)
```

Example usage with a module:

```
# In my_ccb_module.py:
class DataProcessor:
    def __init__(self, prefix: str):
        self.prefix = prefix

def process(self, value: str) -> str:
        return f"{self.prefix}_{value}"

# In your flow definition:
from my_ccb_module import entrypoint

module_ccb = CodeBlock(
    reference_name='example_module_ccb',
    code=entrypoint,
    code_input={
        'input_value': 'hello',
        'prefix': 'output'
    },
    include_module=True  # This includes the entire module
)
```

IDENTIFIER _='CUSTOM\_CODE'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.CodeBlock.IDENTIFIER "Link to this definition")_class_ flows\_sdk.blocks.Fork( _reference\_name_, _branches_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Fork "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

Fork is a system block that is used to schedule other
blocks for parallel execution.

While all branches of a Fork will be scheduled for parallel execution,
the tasks within a branch itself will be executed serially.

The output of a Fork is a dictionary with key: identifier of the output block from each branch
and value: the output from that block

Parameters:

- **reference\_name** ( _str_) – unique identifier on a per-Flow basis

- **branches** ( _Sequence_ _\[_ [_Fork.Branch_](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Fork.Branch "flows_sdk.blocks.Fork.Branch") _\]_) – A sequence of branches, that will be scheduled for parallel execution.

- **title** ( _Optional_ _\[_ _str_ _\]_) – UI-visible title of the block.

- **description** ( _Optional_ _\[_ _str_ _\]_) – Description of the block. Both useful for documentation purpose and visible
to users in the Flow studio.

```
from typing import Any
from uuid import UUID

from flows_sdk.blocks import CodeBlock, Fork
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow

def entry_point_flow() -> Flow:
    return example_flow_with_fork()

def example_flow_with_fork() -> Flow:
    ccb_before_fork = CodeBlock(
        reference_name='ccb_before_fork',
        code=lambda _: {'pre_fork': 'response CCB before fork'},
        code_input={'_': None},
    )

ccb_A_1 = CodeBlock(
        reference_name='ccb_A_1', code=lambda _: {'a': 'response from A 1'}, code_input={'_': None}
    )

ccb_A_2 = CodeBlock(
        reference_name='ccb_A_2', code=lambda _: {'a': 'response from A 2'}, code_input={'_': None}
    )

# Note that both ccb_A_1 and ccb_A_2 will be executed in this branch sequentially,
    # but because of output=ccb_A_1._reference_name, only it will "ccb_A_1" will have its output
    # as a key under "fork_output" (check sample output in the comment below)
    branch_A = Fork.Branch(blocks=[ccb_A_1, ccb_A_2], label='first', output=ccb_A_1._reference_name)

ccb_B = CodeBlock(
        reference_name='ccb_B', code=lambda _: {'b': 'response from B'}, code_input={'_': None}
    )
    branch_B = Fork.Branch(blocks=[ccb_B], label='second', output=ccb_B._reference_name)

branch_C = Fork.Branch(blocks=[], label='third', output=ccb_before_fork._reference_name)

fork = Fork(reference_name='a_fork', branches=[branch_A, branch_B, branch_C])

def print_function(fork_output: Any) -> None:
        print(fork_output)
        return

#  {
    #     "ccb_A_1": {
    #        "result": {
    #           "a": "response from A 1"
    #        }
    #     },
    #     "ccb_B": {
    #        "result": {
    #           "b": "response from B"
    #        }
    #     }
    #     "ccb_before_fork": {
    #        "result": {
    #           "pre_fork": "response CCB before fork"
    #        }
    #     }
    #  }
    print_ccb = CodeBlock(
        reference_name='print_ccb', code=print_function, code_input={'fork_output': fork.output()}
    )

return Flow(
        title='Fork sample flow',
        description='A simple Flow showcasing how a Fork is used',
        blocks=[ccb_before_fork, fork, print_ccb],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(identifier='fork_example', input=[]),
        uuid=UUID('3e3ab564-fcf5-41fb-a573-4bc2fd153b6d'),
        input={},
    )

if __name__ == '__main__':
    export_flow(flow=entry_point_flow())
```

_class_ Branch( _blocks_, _label=None_, _output=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Fork.Branch "Link to this definition")

Bases: `_Branch`

A collection of Blocks that will be serially scheduled for execution.

Parameters:

- **blocks** ( _Sequence_ _\[_ [_Block_](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block") _\]_) – A sequence of Blocks that will be executed as part of that Fork.Branch.
Should contain at least one.

- **label** ( _Optional_ _\[_ _str_ _\]_) – UI-visible text lable of the branch.

- **output** ( _Optional_ _\[_ _str_ _\]_) – Reference name of the Block which will be used as the output of the entire
branch. When not provided, the output of the last block of the branch will be treated
as the branch output.

_class_ flows\_sdk.blocks.Routing( _decision_, _branches_, _reference\_name=None_, _default\_branch=None_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

Routing is a system Block that executes only
the blocks in one of its branches that meets a condition.
Similar in concept to a switch..case or an if/else construction.

Parameters:

- **reference\_name** (`Optional`\[`str`\]) – \[description\]

- **decision** (`str`) – \[description\]

- **branches** (`Sequence`\[ [`Branch`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing.Branch "flows_sdk.blocks.Routing.Branch")\]) – \[description\]

- **default\_branch** (`Optional`\[ [`DefaultBranch`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing.DefaultBranch "flows_sdk.blocks.Routing.DefaultBranch")\]) – \[description\], defaults to None

- **title** (`Optional`\[`str`\]) – \[description\], defaults to None

- **description** (`Optional`\[`str`\]) – \[description\], defaults to None

```
from typing import Any
from uuid import UUID

from flows_sdk.blocks import CodeBlock, Routing
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow

def entry_point_flow() -> Flow:
    return example_flow_with_routing()

def example_flow_with_routing() -> Flow:

ccb_before_routing = CodeBlock(
        reference_name='ccb_before_routing',
        code=lambda _: {'pre_routing': 'response CCB before routing'},
        code_input={'_': None},
    )

decision_ccb = CodeBlock(
        reference_name='decision_ccb',
        code=lambda a_char: {'some_field': a_char},
        code_input={'a_char': 'A'},
    )

ccb_A = CodeBlock(
        reference_name='ccb_A', code=lambda _: {'a': 'response from A'}, code_input={'_': None}
    )
    branch_A = Routing.Branch(case='A', blocks=[ccb_A], label='first', output=ccb_A._reference_name)

branch_B = Routing.Branch(
        case='B', blocks=[], label='second', output=ccb_before_routing._reference_name
    )

ccb_C = CodeBlock(
        reference_name='ccb_C', code=lambda _: {'c': 'response from C'}, code_input={'_': None}
    )
    default_branch = Routing.DefaultBranch(
        blocks=[ccb_C], label='default', output=ccb_C._reference_name
    )

# 'decision' is like a switch..case statement - when a Branch 'case' is matched, it will be
    # scheduled for execution. If no case matches, a 'default_branch' can be defined as fallback.
    routing = Routing(
        reference_name='a_routing',
        decision=decision_ccb.output('some_field'),
        branches=[branch_A, branch_B],
        default_branch=default_branch,
    )

def print_function(routing_output: Any) -> None:
        print(routing_output)
        return

#   "routing_output": {
    #       "result": {
    #          "a": "response from A"
    #       }
    #    }
    print_ccb = CodeBlock(
        reference_name='print_ccb',
        code=print_function,
        code_input={'routing_output': routing.output()},
    )

return Flow(
        title='Routing sample flow',
        description='A simple Flow showcasing how Routing is used',
        blocks=[ccb_before_routing, decision_ccb, routing, print_ccb],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(identifier='routing_example', input=[]),
        uuid=UUID('4e3ab564-fcf5-41fb-a573-4bc2fd153b6d'),
        input={},
    )

if __name__ == '__main__':
    export_flow(flow=entry_point_flow())
```

_class_ Branch( _case_, _blocks_, _label=None_, _output=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing.Branch "Link to this definition")

Bases: `_Branch`

\[summary\]

Parameters:

- **case** ( _str_) – case that must be matched in order for this branch to be scheduled for
execution.

- **blocks** ( _Sequence_ _\[_ [_Block_](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block") _\]_) – sequence of blocks to be executed as part of this branch

- **label** ( _Optional_ _\[_ _str_ _\]_) – UI-visible text lable of the branch

_class_ DefaultBranch( _blocks_, _label=None_, _output=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing.DefaultBranch "Link to this definition")

Bases: `_Branch`

Same as [`flows_sdk.blocks.Routing.Branch`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Routing.Branch "flows_sdk.blocks.Routing.Branch"), but without a case.
DefaultBranch is optionally provided and executed as a fallback when there is no
matching case from other branches.

_class_ flows\_sdk.blocks.IOBlock( _identifier_, _enabled_, _reference\_name=None_, _input=None_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.IOBlock "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

A Block that has the additional boolean property enabled.
Used for Triggers and Outputs where enabled can be triggered via the Flow Studio UI
and is considered during processing (e.g. only enabled Triggers can initiate Flow execution).

See [`flows_sdk.blocks.Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block") for description of the inherited parameters.

Parameters:

**enabled** ( _bool_) – on-off state of the Block.
Visualized as a checkbox and taken into account during processing.

_class_ flows\_sdk.blocks.Outputs( _role\_filter_, _input\_template_, _blocks_, _reference\_name=None_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Outputs "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

Outputs is a special block-section meant to provide an easy way for users to configure
Flow output connections through the UI. They contain a list of output blocks, that can
optionally be filtered by role and define an input template that will wire these blocks’
inputs automatically after the user adds them.

Parameters:

- **role\_filter** (`List`\[`str`\]) – Serves as an instruction to the UI for what blocks are allowed in
this output section, since not all output blocks will be suitable for all flows.
UI will allow adding only blocks that have all listed roles.
If empty, blocks with any role will be allowed.

- **input\_template** (`Dict`\[`str`, `Any`\]) – Used by the UI to automatically pre-populate the input of
newly added blocks. Thus the Flow designer can define the wiring of output blocks
to the results from previous blocks in the Flow, while allowing the user
to add or remove output blocks via the UI.

- **blocks** (`Sequence`\[ [`IOBlock`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.IOBlock "flows_sdk.blocks.IOBlock")\]) – List of output IOBlocks for sending results to external systems.
Can be empty, since the outputs within an Outputs section are editable in Flow studio.

- **title** (`Optional`\[`str`\]) – UI-visible title of the block.

- **description** (`Optional`\[`str`\]) – Description of the block. Both useful for documentation purpose and visible
to users in the Flow studio.

_class_ flows\_sdk.blocks.Foreach( _items_, _template_, _reference\_name_, _title=None_, _description=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Foreach "Link to this definition")

Bases: [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")

**Available in v35 and later.**

Foreach is a system block that can be used to run a given task for each item in a collection.
The task is represented by a block and the collection is dynamically generated at runtime - can
be a flow input or the output of a previous block.
The block that will run for each item in the collection is called a “template” and supports
special syntax for referencing the basic foreach loop concepts:

- `${foreach_reference_name.item}` \- to access the current item. Syntax like
${foreach\_reference\_name.item.nested\_prop} is also supported.

- `${foreach_reference_name.index}` \- to access the index of the current item, 0-based

At runtime Foreach dynamically generates blocks by applying the template to each item in the
collection. The generated blocks are executed in parallel.
The output of the Foreach block is a list that contains the outputs of the dynamically created
blocks in the same order as the input collection.

Parameters:

- **items** (`str`) – reference to the list of items to which the template will be applied

- **template** ([`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")) – the block to run for each item in the provided collection

- **reference\_name** (`str`) – unique identifier of the block in the flow. Note that unlike in other
blocks here the reference\_name is not optional as it is needed to refer to the items
of the collection in the template

- **title** (`Optional`\[`str`\]) – UI-visible title of the block.

- **description** (`Optional`\[`str`\]) – Description of the block. Both useful for documentation purposes and
visible to users in the Flow studio.

```
import sys
from typing import List
from uuid import UUID

from flows_sdk.blocks import CodeBlock, Foreach
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow

def entry_point_flow() -> Flow:
    return example_flow_with_foreach()

def example_flow_with_foreach() -> Flow:
    def _create_items() -> List[int]:
        return [1, 2, 3, 4]

create_items = CodeBlock(reference_name='create_items', code=_create_items, code_input={})

def multiply_by_2(n: int) -> int:
        return n * 2

foreach = Foreach(
        reference_name='multiply_by_2',
        items=create_items.output(),
        template=CodeBlock(
            reference_name='multiply_ccb_${multiply_by_2.index}',
            code=multiply_by_2,
            code_input={'n': '${multiply_by_2.item}'},
        ),
    )

def sum_numbers(multiplied_numbers: list) -> int:
        # the outputs of the code blocks are wrapped in a dict like {'result': <actual_number>}
        return sum([n['result'] for n in multiplied_numbers])

use_foreach_output = CodeBlock(
        reference_name='sum_numbers',
        code=sum_numbers,
        code_input={'multiplied_numbers': foreach.output()},
    )

return Flow(
        title='Foreach sample flow',
        description='A simple Flow showcasing how FOREACH is used',
        blocks=[create_items, foreach, use_foreach_output],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(identifier='FOREACH_EXAMPLE', input=[]),
        uuid=UUID('a1edebd9-e96f-4c10-beeb-6cce3e92d4f1'),
        input={},
    )

if __name__ == '__main__':
    export_filename = None
    if len(sys.argv) > 1:
        export_filename = sys.argv[1]

export_flow(flow=entry_point_flow(), filename=export_filename)
```

output( _key=None_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Foreach.output "Link to this definition")

Parameters:

Returns:

a string-formatted reference that will be unpacked to a value during runtime.

Return type:

str

For example:

```
a = Block(...)
b = Block(
    ...,
    input = {
        'foo': a.output('bar')
    }
)
```

## flows.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.flows "Link to this heading")

_pydanticmodel_ flows\_sdk.flows.Parameter [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter "Link to this definition")

Bases: `_BaseModel`

Defines an input parameter of a block or a flow.

Fields:

- [`dependencies (List[flows_sdk.flows.Dependency] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.dependencies "flows_sdk.flows.Parameter.dependencies")

- [`description (str | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.description "flows_sdk.flows.Parameter.description")

- [`json_schema (Dict[str, Any] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.json_schema "flows_sdk.flows.Parameter.json_schema")

- [`name (str)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.name "flows_sdk.flows.Parameter.name")

- [`optional (bool)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.optional "flows_sdk.flows.Parameter.optional")

- [`secret (bool | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.secret "flows_sdk.flows.Parameter.secret")

- [`title (str | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.title "flows_sdk.flows.Parameter.title")

- [`type (str)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.type "flows_sdk.flows.Parameter.type")

- [`type_spec (Dict[str, Any] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.type_spec "flows_sdk.flows.Parameter.type_spec")

- [`ui (Dict[str, Any] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.ui "flows_sdk.flows.Parameter.ui")

- [`value (Any)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.value "flows_sdk.flows.Parameter.value")

_field_ dependencies _:`Optional`\[`List`\[`Dependency`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.dependencies "Link to this definition")

Internal mechanism for describing dependencies between fields (e.g. show A when B is checked,
hide A when B is unchecked)

_field_ description _:`Optional`\[`str`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.description "Link to this definition")

UI-visible description of the field

_field_ json\_schema _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ _(alias'schema')_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.json_schema "Link to this definition")

Defines a format for the field using the JSON Schema specification
( [https://json-schema.org/understanding-json-schema/](https://json-schema.org/understanding-json-schema/)).
Field values will be validated against this format when a flow is imported or edited.

_field_ name _:`str`_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.name "Link to this definition")

Must be a valid Python identifier

_field_ optional _:`bool`_ _=False_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.optional "Link to this definition")

True when the field can be omitted, False when it is mandatory.
The default value is False, i.e. if the optional flag is not specified,
the field will be considered mandatory.

_field_ secret _:`Optional`\[`bool`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.secret "Link to this definition")

UI-hint that the field should be presented as a secret (hidden symbols)
rather than plain text. Secrets are explained in depth in
[Secure Handling of Secrets](https://flows-sdk.hyperscience.ai/pages/secrets.html#secure-handling-of-secrets).

_field_ title _:`Optional`\[`str`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.title "Link to this definition")

UI-visible representation of the field name

_field_ type _:`str`_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.type "Link to this definition")

Json-schema-like type (e.g. string, integer, number, array, object…)
Between Hyperscience platform versions there are also extensions like:

- Percentage (number between 0 and 1)

- UUID (string in UUID format or null)

- MultilineText (string, but represented as multiline text box in Flow studio)

Depending on the Hyperscience version, other supported types may also exist,
some of which allow for specific rendering in UI.

_field_ type\_spec _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.type_spec "Link to this definition")

An optional object property that further specifies (or “specializes”) the type, in ways that
cannot be achieved via the schema property. Its exact format is specific to the specific type
used. If specified must conform to whatever the type\_spec format specific to the chosen
type is.

File input flow that uses type\_spec

This flow uses type\_spec to limit the types of files that can be selected for the file
input.

```
from uuid import UUID

from flows_sdk.blocks import CodeBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

FILE_INPUT_FLOW_IDENTIFIER = 'FILE_INPUT_SHOWCASE'

FILE_INPUT_FLOW_UUID = UUID('2869449d-db9e-485b-b285-f954346793c6')

class FlowInputs:
    FILE = 'file'

def entry_point_flow() -> Flow:
    return file_input_showcase_flow()

def file_input_showcase_flow() -> Flow:
    def _read_and_log_file(file_uuid: str, _hs_block_instance: HsBlockInstance) -> str:
        if not file_uuid:
            _hs_block_instance.log('No file configured')
            return ''
        blob = _hs_block_instance.fetch_blob(file_uuid)
        file_text = blob.content.decode(encoding='utf-8')
        _hs_block_instance.log(file_text)
        return file_text

read_and_log_file = CodeBlock(
        reference_name='read_and_log_file',
        code=_read_and_log_file,
        code_input={'file_uuid': workflow_input(FlowInputs.FILE)},
    )

return Flow(
        title='Flow with a file input',
        description='Accepts a text file as an input and logs and outputs its contents',
        blocks=[read_and_log_file],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(
            identifier=FILE_INPUT_FLOW_IDENTIFIER,
            input=[\
                Parameter(\
                    name=FlowInputs.FILE,\
                    title='File to read',\
                    type='File',\
                    # restricts the type of file that can be selected\
                    type_spec={'allowed_extensions': '.txt'},\
                    optional=True,\
                )\
            ],
        ),
        uuid=FILE_INPUT_FLOW_UUID,
        input={FlowInputs.FILE: 'some_text.txt'},
    )

if __name__ == '__main__':
    export_flow(flow=entry_point_flow())
```

_field_ ui _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.ui "Link to this definition")

Presentation-related customizations for the field

_field_ value _:`Any`_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter.value "Link to this definition")

Optional default value of the field,
used when its value is not specified explicitly in the inputs

_pydanticmodel_ flows\_sdk.flows.InputDefinition [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition "Link to this definition")

Bases: `_BaseModel`

Declaratively defines what parameters are expected, usually part of a Manifest.

Fields:

- [`input (List[flows_sdk.flows.Parameter])`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition.input "flows_sdk.flows.InputDefinition.input")

- [`ui (Dict[str, Any] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition.ui "flows_sdk.flows.InputDefinition.ui")

_field_ input _:`List`\[ [`Parameter`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter "flows_sdk.flows.Parameter")\]_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition.input "Link to this definition")

All parameters that are part of this definition

_field_ ui _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition.ui "Link to this definition")

Presentation-related customizations for how fields are grouped together.
ui\[‘groups’\] defines how fields are grouped together in the UI and specifies the group titles.
groups are optional - if omitted, every input field is listed in the “default” group.
If any fields are not included in any group, then they are shown as part of the “default”
group of non-nested fields.

example:

```
ui: {
    groups: [\
    {\
        title: 'Group 1',\
        fields: [\
            'layout_release_uuid',\
            'policy',\
        ],\
    },\
    {\
        title: 'Group 2',\
        fields: ['accuracy'],\
    },\
    ]
}
```

_pydanticmodel_ flows\_sdk.flows.Manifest [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest "Link to this definition")

Bases: [`InputDefinition`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.InputDefinition "flows_sdk.flows.InputDefinition")

Describes how the Flow studio should present the Flow/Block.
The parameters are displayed in the Flow studio and can be futher customized into sections via.
the ui section.

Inputs are used both for visualization and for validation - for example, in the
[`flows_sdk.flows.Flow`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow "flows_sdk.flows.Flow") manifest `.input`, if a string is passed for a parameter
defined as number in the manifest, the user will be prompted to

Here is an example of how the manifest is visualized on the Flow level with
[`flows_sdk.flows.Parameter`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter "flows_sdk.flows.Parameter") groupped under _Classification_ and _Identification_

example:

```
Manifest(
    identifier='CUSTOM_FLOW',
    input=[\
        Parameter(\
            name='setting_a',\
            type='string',\
            title='Setting A',\
            value=''\
        ),\
        Parameter(\
            name='setting_b',\
            type='number',\
            title='Setting B',\
            value=42\
        ),\
        Parameter(\
            name='setting_c',\
            type='string',\
            title='Setting C',\
            value=''\
        ),\
    ],
    ui={
        'groups': [\
            {\
                'title': 'Groupped Settings',\
                'fields': [\
                    'setting_a',\
                    'setting_b',\
                ],\
            },\
        ]
    }
)
```

Fields:

- [`enable_overrides (bool | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.enable_overrides "flows_sdk.flows.Manifest.enable_overrides")

- [`identifier (str)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.identifier "flows_sdk.flows.Manifest.identifier")

- [`output (List[flows_sdk.flows.Parameter] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.output "flows_sdk.flows.Manifest.output")

- [`roles (List[str])`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.roles "flows_sdk.flows.Manifest.roles")

_field_ enable\_overrides _:`Optional`\[`bool`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.enable_overrides "Link to this definition")

UI representation of overrides if present.
This requires additional API calls and should generally be set to True only when overrides
are defined
Advanced functionality, contact Hyperscience if you think you need it!

_field_ identifier _:`str`_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.identifier "Link to this definition")

Globally-unique identifier of the block/flow. By convention - all capital snake-case witn
an optional numeric suffix (e.g., `HELLO_FLOW_2`)

_field_ output _:`Optional`\[`List`\[ [`Parameter`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Parameter "flows_sdk.flows.Parameter")\]\]_ _=\[\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.output "Link to this definition")

Documents the outputs of the block/flow. For the time being, used for documentation
purposes only.

_field_ roles _:`List`\[`str`\]_ _=\[\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest.roles "Link to this definition")

Allows for “tagging” blocks/flows, used by internal features.

_pydanticmodel_ flows\_sdk.flows.Flow [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow "Link to this definition")

Bases: `_BaseModel`

Flow is the top-level construct describing the type and order of various components.

Sample usage:

```
Flow(
    title='Flow title', # UI visible and editable title
    description='Flow description' # UI visible and editable description
    blocks=[ccb], # sequence of blocks in order
    owner_email='flows.sdk@hyperscience.com',
    manifest=Manifest(identifier='HELLO_FLOW', input=[hello_input_param]),
    uuid=UUID('2e3ab564-fcf5-41fb-a573-4bc2fd153b6d'),
    input={'hello_input': 'World'},
)
```

Config:

- **arbitrary\_types\_allowed**: _bool = True_

Fields:

- [`blocks (Sequence[flows_sdk.blocks.Block])`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.blocks "flows_sdk.flows.Flow.blocks")

- [`description (str | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.description "flows_sdk.flows.Flow.description")

- [`error_handling (flows_sdk.error_handling.ErrorHandling | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.error_handling "flows_sdk.flows.Flow.error_handling")

- [`input (Dict[str, Any])`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.input "flows_sdk.flows.Flow.input")

- [`manifest (flows_sdk.flows.Manifest)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.manifest "flows_sdk.flows.Flow.manifest")

- [`output (Dict[str, str] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.output "flows_sdk.flows.Flow.output")

- [`owner_email (str)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.owner_email "flows_sdk.flows.Flow.owner_email")

- [`title (str)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.title "flows_sdk.flows.Flow.title")

- [`triggers (flows_sdk.flows.Triggers | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.triggers "flows_sdk.flows.Flow.triggers")

- [`uuid (uuid.UUID)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.uuid "flows_sdk.flows.Flow.uuid")

- [`variables (Dict[str, Any] | None)`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.variables "flows_sdk.flows.Flow.variables")

_field_ blocks _:`Sequence`\[ [`Block`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.blocks.Block "flows_sdk.blocks.Block")\]_ _=\[\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.blocks "Link to this definition")

Sequence of Blocks to be scheduled for execution when the Flow is triggered.

_field_ description _:`Optional`\[`str`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.description "Link to this definition")

User-editable description of what the Flow is used for.

_field_ error\_handling _:`Optional`\[ [`ErrorHandling`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorHandling "flows_sdk.error_handling.ErrorHandling")\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.error_handling "Link to this definition")

Error handling policies for the flow.

_field_ input _:`Dict`\[`str`,`Any`\]_ _={}_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.input "Link to this definition")

Key-value pairs for top-level inputs of the flow. Their types / UI representations are part
of the Manifest.

_field_ manifest _: [`Manifest`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Manifest "flows_sdk.flows.Manifest")_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.manifest "Link to this definition")_field_ output _:`Optional`\[`Dict`\[`str`,`str`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.output "Link to this definition")_field_ owner\_email _:`str`_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.owner_email "Link to this definition")

An email address where the creator of the Flow can be reached.
This field is purely informative metadata, it does not control any permissions or
behaviours in the system. The intended use is for the flow developer to define it in
the Flows SDK-based definition of the flow. It is read-only in the UI because business users
will mostly be using this for information purposes and will not usually need to change it.
If necessary, they may still change this by exporting, editing the json of the flow,
and reimporting.

_field_ title _:`str`_ _\[Required\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.title "Link to this definition")

User-visible and editable title, does not have to be unique.

_field_ triggers _:`Optional`\[`Triggers`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.triggers "Link to this definition")

Trigger blocks that initiate the Flow.

_field_ uuid _:`UUID`_ _=UUID('00000000-0000-0000-0000-000000000000')_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.uuid "Link to this definition")

Unique identifier of the Flow. If not provided, one will be deterministically
generated based on the value of manifest.identifier. Note that if manually setting the uuid,
a new one should be used every time manifest.identifier changes.

_field_ variables _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.variables "Link to this definition")_class_ Config [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.Config "Link to this definition")

Bases: `object`

arbitrary\_types\_allowed _=True_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.Config.arbitrary_types_allowed "Link to this definition")_classmethod_ generate\_uuid\_when\_not\_provided( _values_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.flows.Flow.generate_uuid_when_not_provided "Link to this definition")

Generate a deterministic UUID based on manifest.identifier if one is not provided.
We use \_ZERO\_UUID as an anchor value to ensure that the type of uuid is kept as UUID
instead of Optional\[UUID\], as some pieces of code might rely on the value being a valid
UUID instead of Null that gets rendered to a valid UUID.

Return type:

`Dict`\[`str`, `Any`\]

## utils.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.utils "Link to this heading")

This module provides utility functions and references for working with flow inputs
and runtime metadata in flows.

flows\_sdk.utils.workflow\_input( _\*params_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.workflow_input "Link to this definition")

Generate a reference string to access flow input parameters.

This function creates a reference that will be resolved at runtime to access
values from the flow’s input parameters. It supports accessing nested
parameters using dot notation.

Parameters:

**params** ( _str_) – One or more parameter names to create a reference path.
Multiple parameters will be joined with dots to access nested values.

Returns:

A reference string in the format `${workflow.input.param_path}`
that will be resolved at runtime.

Return type:

str

Raises:

**AssertionError** – If no parameters are provided.

```
from uuid import UUID

from flows_sdk.blocks import CodeBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.utils import workflow_input

def entry_point_flow() -> Flow:
    return workflow_input_example_flow()

def workflow_input_example_flow() -> Flow:
    """Example showing how to use workflow_input to access simple and nested input parameters."""

# Simple parameter access - single argument
    def process_text(text: str) -> str:
        return text.upper()

simple_block = CodeBlock(
        reference_name='process_text',
        code=process_text,
        code_input={'text': workflow_input('text_to_process')},
    )

# Nested parameter access - multiple arguments joined with dots
    def extract_submission_id(submission_id: str) -> str:
        return f'Processing submission: {submission_id}'

nested_block = CodeBlock(
        reference_name='extract_id',
        code=extract_submission_id,
        code_input={'submission_id': workflow_input('submission', 'id')},
    )

return Flow(
        title='Workflow Input Example',
        description='Demonstrates using workflow_input() with simple and nested parameters',
        blocks=[simple_block, nested_block],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(
            identifier='WORKFLOW_INPUT_EXAMPLE',
            input=[\
                Parameter(name='text_to_process', type='string'),\
                Parameter(name='submission', type='object'),\
            ],
        ),
        uuid=UUID('c2d3e4f5-a6b7-8c9d-0e1f-2a3b4c5d6e7f'),
        input={
            'text_to_process': 'hello world',
            'submission': {'id': 'sub_12345', 'status': 'pending'},
        },
    )
```

_class_ flows\_sdk.utils.FlowDefinitionReferences [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.FlowDefinitionReferences "Link to this definition")

Bases: `object`

Container for flow definition metadata references.

This class provides constant reference strings that will be resolved at runtime
to access metadata about the flow definition itself (as opposed to a specific
flow run).

These references can be used in block inputs to access flow definition metadata
dynamically at runtime.

```
from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest
from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import FlowDefinitionReferences

def entry_point_flow() -> Flow:
    return workflow_definition_references_example_flow()

def workflow_definition_references_example_flow() -> Flow:
    """Example showing how to use WorkflowDefinitionReferences to access flow metadata."""

def log_flow_info(flow_identifier: str, _hs_block_instance: HsBlockInstance) -> None:
        _hs_block_instance.log(f'Running flow: {flow_identifier}')

log_block = PythonBlock(
        reference_name='log_flow',
        code=log_flow_info,
        code_input={'flow_identifier': FlowDefinitionReferences.IDENTIFIER},
    )

return Flow(
        title='Workflow Definition References Example',
        description='Demonstrates using WorkflowDefinitionReferences.IDENTIFIER',
        blocks=[log_block],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(identifier='WORKFLOW_DEF_REF_EXAMPLE', input=[]),
        uuid=UUID('d3e4f5a6-b7c8-9d0e-1f2a-3b4c5d6e7f8a'),
        input={},
    )
```

IDENTIFIER _='${workflow.definition.identifier}'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.FlowDefinitionReferences.IDENTIFIER "Link to this definition")

Reference to the flow definition identifier.

This will resolve at runtime to the unique identifier of the flow definition,
which is set in the flow’s Manifest (e.g., ‘MY\_CUSTOM\_FLOW’).

_class_ flows\_sdk.utils.FlowRunReferences [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.FlowRunReferences "Link to this definition")

Bases: `object`

Container for flow run metadata references.

This class provides constant reference strings that will be resolved at runtime
to access metadata about the current flow run instance (as opposed to the
flow definition).

These references can be used in block inputs to access runtime-specific information
such as the unique run UUID or correlation ID for tracking related flow executions.

For flow run related APIs please refer to the [Flow Run API documentation](https://docs.hyperscience.ai/#flows-runs).

```
import sys
from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import FlowRunReferences

def entry_point_flow() -> Flow:
    return workflow_run_references_example_flow()

def workflow_run_references_example_flow() -> Flow:
    """Example showing how to use WorkflowRunReferences to track flow execution."""

def log_execution(
        run_uuid: str,
        correlation_id: str,
        _hs_block_instance: HsBlockInstance,
    ) -> None:
        _hs_block_instance.log(f'Run UUID: {run_uuid}')
        _hs_block_instance.log(f'Correlation ID: {correlation_id}')

log_block = PythonBlock(
        reference_name='log_execution',
        code=log_execution,
        code_input={
            'run_uuid': FlowRunReferences.UUID,
            'correlation_id': FlowRunReferences.CORRELATION_ID,
        },
    )

return Flow(
        title='Workflow Run References Example',
        description='Demonstrates using WorkflowRunReferences for tracking execution',
        blocks=[log_block],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(identifier='WORKFLOW_RUN_REF_EXAMPLE', input=[]),
        uuid=UUID('e4f5a6b7-c8d9-0e1f-2a3b-4c5d6e7f8a9b'),
        input={},
    )

if __name__ == '__main__':
    export_filename = None
    if len(sys.argv) > 1:
        export_filename = sys.argv[1]

export_flow(flow=entry_point_flow(), filename=export_filename)
```

UUID _='${workflow.run.uuid}'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.FlowRunReferences.UUID "Link to this definition")

Reference to the unique identifier of the current flow run.

This will resolve at runtime to a UUID that uniquely identifies this specific
execution instance of the flow (e.g., ‘550e8400-e29b-41d4-a716-446655440000’).

CORRELATION\_ID _='${workflow.run.correlation\_id}'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.utils.FlowRunReferences.CORRELATION_ID "Link to this definition")

Reference to the correlation ID of the current flow run.

The correlation ID is a string that gets passed from parent to child flow runs
and can be used to identify flow runs that belong to the same execution graph.
This is particularly useful for tracking related flows in a distributed system
or when flows trigger subflows.

If the flow run is associated with a Submission, the correlation ID will match
the Submission’s correlation\_id. See the Submission API documentation for more details:
[Submission API documentation](https://docs.hyperscience.ai/#submissions).

## types.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.types "Link to this heading")

This module contains types that can be useful for writing flows but are not part of the flow’s
topology.

Available in v35 and later.

_class_ flows\_sdk.types.StoreBlobRequest [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobRequest "Link to this definition")

Bases: `object`

Encapsulates the request parameters for storing a blob.

name _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobRequest.name "Link to this definition")

Name that will be associated with the blob.

content _:`bytes`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobRequest.content "Link to this definition")

Blob content in bytes.

_class_ flows\_sdk.types.StoreBlobResponse [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobResponse "Link to this definition")

Bases: `object`

The result of storing a blob. Contains data that can be used to retrieve the blob.

uuid _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobResponse.uuid "Link to this definition")

Unique identifier of the created blob.

name _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobResponse.name "Link to this definition")

Name associated with the blob.

_class_ flows\_sdk.types.Blob [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Blob "Link to this definition")

Bases: `object`

The result of fetching a blob from the object store.

content _:`bytes`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Blob.content "Link to this definition")

The content of the blob in bytes.

_class_ flows\_sdk.types.HsTask [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask "Link to this definition")

Bases: `object`

A task in the flow engine runtime - contains task and flow metadata.

Available in v35 and later.

task\_id _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask.task_id "Link to this definition")

Unique identifier of the task withing the system

flow\_run\_id _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask.flow_run_id "Link to this definition")

Unique identifier of the flow run withing the system

correlation\_id _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask.correlation_id "Link to this definition")

A string that gets passed from parent to child flow runs and can be used to
identify flow runs that belong to the same execution graph

task\_name _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask.task_name "Link to this definition")

The name of the block that runs the task, e.g. PYTHON\_CODE

reference\_name _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsTask.reference_name "Link to this definition")

The unique identifier of the task in the flow definition

_class_ flows\_sdk.types.Measurement [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement "Link to this definition")

Bases: `object`

The datum or numerical value associated with a metric, described over the respective dimensions.
AKA Fact in the OLAP world.

Measurements do not have uniqueness properties. Two measurements with the exact same values will
not be considered duplicate, but rather different measurements taken at the same time,
for the same metric, with the same value and dimensions.

Available in V41 and later.

name _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement.name "Link to this definition")

Name is the machine identifier for a metric. Names cannot be blank and, alongside namespace,
represent the identifier of a metric.

namespace _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement.namespace "Link to this definition")

Namespace represents the context under which the metric is used. This property avoids
conflicts between metrics with the same name, used in different contexts. Similar to the name
field, a namespace cannot be blank.

time _:`datetime`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement.time "Link to this definition")

The ‘time’ field refers to the “occurrence time” where the measurement was taken. In
event-driven architecture (EDA) this is usually referred to as “event time”, and just like in
EDA, this has no relation to the real time, i.e., the time at which the event/measurement is
processed by the analytics system. This datetime field must be timezone-aware.

dimensions _:`dict`\[`str`,`str`\]_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement.dimensions "Link to this definition")

Dimensions provide the “who, what, where, when, why, and how” context surrounding a
measurement. They contain the attributes which characterize the measurement, which, at the
analytical level, can be used to filter and/or group measurements.

value _:`float`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement.value "Link to this definition")

The datum associated with a metric, described over the respective dimensions. AKA Fact in
the OLAP world.

_class_ flows\_sdk.types.HsBlockInstance [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance "Link to this definition")

Bases: `ABC`

Provides an interface for the base block APIs in code blocks.

Available in v35 and later.

_class_ LogLevel( _\*values_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel "Link to this definition")

Bases: `Enum`

DEBUG _=1_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel.DEBUG "Link to this definition")INFO _=2_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel.INFO "Link to this definition")WARN _=3_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel.WARN "Link to this definition")ERROR _=4_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel.ERROR "Link to this definition")_abstractmethod_ store\_blob( _blob_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.store_blob "Link to this definition")

Stores a blob in the object store. For more information see store\_blobs.

Parameters:

**blob** ([`StoreBlobRequest`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobRequest "flows_sdk.types.StoreBlobRequest")) – The binary object to store.

Returns:

A response object that contains the blob’s unique identifier.

Return type:

[StoreBlobResponse](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobResponse "flows_sdk.types.StoreBlobResponse")

_abstractmethod_ store\_blobs( _blobs_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.store_blobs "Link to this definition")

Stores a list of blobs in the object store. The blobs will be associated with the
flow run where this code gets executed. When this flow run is deleted the blobs will be
deleted as well.

Parameters:

**blobs** (`Iterable`\[ [`StoreBlobRequest`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobRequest "flows_sdk.types.StoreBlobRequest")\]) – List of blobs to store.

Returns:

A list of response objects that contain the blobs’ unique identifiers.

Return type:

List\[ [StoreBlobResponse](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.StoreBlobResponse "flows_sdk.types.StoreBlobResponse")\]

_abstractmethod_ fetch\_blob( _blob\_reference_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.fetch_blob "Link to this definition")

Retrieves the contents for the provided blob reference string.

Parameters:

**blob\_reference** (`str`) – The unique identifier of the blob to be fetched.

Returns:

The blob’s content. If no blob with the provided reference exists, an exception
will be raised.

Return type:

[Blob](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Blob "flows_sdk.types.Blob")

_abstractmethod_ log( _msg_, _\*args_, _level=LogLevel.INFO_, _\*\*kwargs_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.log "Link to this definition")

Logs a message at the given log level both to the standard out/err stream and to the
flow execution context, making it available in the Logs tab of the View Flow Execution UI.
Prepends meta information that makes the logs easier to trace (e.g. by correlation id)

This signature of the function is only available when using a PYTHON\_3\_13 block or later
(i.e. python version 3.13+), introduced in version 42.2.

When using an older block (e.g. PYTHON\_3\_12), logging arguments are not supported,
and using brackets {} when logging will result in an error.

Parameters:

- **msg** (`str`) – The message to log.

- **level** ([`LogLevel`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.LogLevel "flows_sdk.types.HsBlockInstance.LogLevel")) – (Optional) The log level to use. Defaults to INFO.

- **args** (`Any`) – (Optional) wildcard arguments to format the log with.
Uses {} style formatting. **(PYTHON\_3\_13+)**

- **kwargs** (`Any`) – (Optional) wildcard keyword arguments to format the log with.
Uses {} style formatting. **(PYTHON\_3\_13+)**

Return type:

`None`

hs\_request( _method_, _app\_endpoint_, _\*\*kwargs_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.hs_request "Link to this definition")

**Available in v40.2 and later.**

Makes an authenticated HTTP request to the Hyperscience platform.
This method will automatically take care of prepending the base url and will include
the appropriate authentication headers with the request.

Mandatory parameters:

Parameters:

- **method** (`str`) – HTTP method to use, e.g. ‘GET’, ‘POST’, etc.

- **app\_endpoint** (`str`) – relative url, beginning with a forward slash, to an HTTP endpoint
served by the Hyperscience platform. For reference on the available endpoints, check out
[https://docs.hyperscience.ai/](https://docs.hyperscience.ai/).
Absolute URLs that include a hostname are allowed only if the flow has received them
as part of the response of a previous API request that the flow has made.
Do not hardcode the hostname of the Hyperscience instance in your flow definitions as
they will not work when you run the flows on another instance,
or if the hostname changes.

Optional Parameters:

Parameters:

**kwargs** (`Dict`\[`Any`, `Any`\]) – Additional keyword arguments to pass to the request call - the requests
library is used under the hood. E.g. POST requests probably need a body passed via
the data or json argument.
See [https://requests.readthedocs.io/en/latest/api/#requests.request](https://requests.readthedocs.io/en/latest/api/#requests.request).

Return type:

`Any`

Returns:

A requests.Response object

hs\_get( _app\_endpoint_, _\*\*kwargs_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.hs_get "Link to this definition")

**Available in v40.2 and later.**

Makes an authenticated GET request to the Hyperscience platform.
See docs for HsBlockInstance.hs\_request above.

Return type:

`Any`

hs\_post( _app\_endpoint_, _\*\*kwargs_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.hs_post "Link to this definition")

**Available in v40.2 and later.**

Makes an authenticated POST request to the Hyperscience platform.
See docs for HsBlockInstance.hs\_request above.

Return type:

`Any`

_abstractmethod_ publish\_measurements( _measurements_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.HsBlockInstance.publish_measurements "Link to this definition")

**Available in v41 and later.**

Publishes measurements to the Hyperscience analytics system.

Mandatory parameters:

Parameters:

**measurements** (`list`\[ [`Measurement`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.Measurement "flows_sdk.types.Measurement")\]) – The provided measurements are persisted in the system, and are
eventually available in Hyperscience reports.

Return type:

`None`

_class_ flows\_sdk.types.CompatibilitySpec [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec "Link to this definition")

Bases: `object`

**Available in v38 and later.**

Provides a definition of the compatibility spec used in blocks. The spec allows to change
the identifier of a block or a flow to the identifier of a block that corresponds to the spec.

filter\_roles _:`Optional`\[`List`\[`str`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec.filter_roles "Link to this definition")

Takes a list of strings that represent flow or block roles and allows the UI to filter by them

filter\_schema _:`Optional`\[`Dict`\[`str`,`Any`\]\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec.filter_schema "Link to this definition")

Represents a JSON schema, that allows for more intricate filtering.

label _:`Optional`\[`str`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec.label "Link to this definition")

Represents the UI label in FlowStudio

_static_ compatibility\_spec\_dict\_factory( _data_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.types.CompatibilitySpec.compatibility_spec_dict_factory "Link to this definition")Return type:

`Dict`\[`str`, `Any`\]

## mocks.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.mocks "Link to this heading")

_class_ flows\_sdk.mocks.MockHsTask( _task\_id='dummy\_task\_id'_, _flow\_run\_id='dummy\_flow\_run\_id'_, _correlation\_id='dummy\_correlation\_id'_, _task\_name='dummy\_task\_name'_, _reference\_name='dummy\_reference\_name'_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.MockHsTask "Link to this definition")

Bases: `object`

Used as a stand in HsTask for the purpose of unit-testing CCBs.
Implements its instance variables via custom value or dummy defaults.

_class_ flows\_sdk.mocks.BaseMockHsBlockInstance [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance "Link to this definition")

Bases: `ABC`

Base class for all MockHsBlockInstance classes.

It’s derivatives are used as a stand in HsBlockInstance for the purpose of unit-testing CCBs.
Fully implements its interface in a semi-functional way.

Logs, blobs, and measurements are stored in memory
(therefore can’t be meaningfully used in a multiprocess test).

All of its methods are mocked via \_MockFunction, therefore conforming them to
the Mock testing interface, e.g. mock\_instance.hs\_request.assert\_called\_once().

_class_ LogLevel( _\*values_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance.LogLevel "Link to this definition")

Bases: `Enum`

DEBUG _=1_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance.LogLevel.DEBUG "Link to this definition")INFO _=2_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance.LogLevel.INFO "Link to this definition")WARN _=3_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance.LogLevel.WARN "Link to this definition")ERROR _=4_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance.LogLevel.ERROR "Link to this definition")_class_ flows\_sdk.mocks.MockHsBlockInstanceV1 [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.MockHsBlockInstanceV1 "Link to this definition")

Bases: [`BaseMockHsBlockInstance`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.BaseMockHsBlockInstance "flows_sdk.mocks.BaseMockHsBlockInstance")

First iteration of the MockHsBlockInstance class, this imitates the older interface of
HsBlockInstance prior to the introduction of logging arguments (pre-PYTHON\_3\_13).

Use this class if you are running code blocks with python version <= 3.12.

_class_ flows\_sdk.mocks.MockHsBlockInstanceV2 [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.MockHsBlockInstanceV2 "Link to this definition")

Second iteration of the MockHsBlockInstance class, this imitates the interface of
HsBlockInstance after the introduction of logging arguments (PYTHON\_3\_13 and above).

Use this class if you are running code blocks with python version >= 3.13.

flows\_sdk.mocks.MockHsBlockInstance [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.MockHsBlockInstance "Link to this definition")

alias of [`MockHsBlockInstanceV2`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.mocks.MockHsBlockInstanceV2 "flows_sdk.mocks.MockHsBlockInstanceV2")

Mock objects for unit testing Custom Code Blocks. These classes simulate the Hyperscience
runtime environment, allowing you to test your CCBs locally without needing a running
Hyperscience instance.

For comprehensive testing strategies and examples, see [Unit Testing Custom Code Blocks](https://flows-sdk.hyperscience.ai/pages/testing.html#unit-testing-custom-code-blocks).

## error\_handling.py [](https://flows-sdk.hyperscience.ai/pages/source-docs.html\#module-flows_sdk.error_handling "Link to this heading")

This module contains classes related to flow error handling. Currently, the error handling
framework includes:

- **Automatic retries of block errors.** Available in R36+. This feature allows configuring an
error retry policy on the flow or block level. If a block has an error retry policy defined on
itself or on its parent flow and its execution results in an error a new block with the same
input parameters will get scheduled by the flow engine after a time interval, calculated based
on the effective error retry policy of the failed block.

- **On error flows.** Available in R38+. This feature allows users to configure a flow to run in
case a flow run fails. Each failed flow run will trigger its own on error flow which will receive
the UUID for the failed flow run as input under the input key “failed\_run\_uuid”. The on error
flow is invoked only on terminal failure, errors that will be retried by the block error retry
policy will not trigger it. Currently, can be configured only in a flow level error handling
section.

_class_ flows\_sdk.error\_handling.RetryMethod( _\*values_) [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod "Link to this definition")

Bases: `str`, `Enum`

Retry methods for handling block errors in flows. Each method has its own formula for
calculating the time between a failure and the next retry attempt. Check each enum member for
the specific formula.
When calculating your error retry policy values keep in mind that what you get as a result
using the formulas below is the **minimum** time between retry attempts, blocks may take
additional time to actually run the work.

Available in v36 and later.

FIXED _='FIXED'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod.FIXED "Link to this definition")

Each retry runs after retry\_interval\_seconds seconds.

LINEAR\_BACKOFF _='LINEAR\_BACKOFF'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod.LINEAR_BACKOFF "Link to this definition")

Next retry is scheduled after retry\_interval\_seconds \* attempt\_number seconds.

EXPONENTIAL\_BACKOFF _='EXPONENTIAL\_BACKOFF'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod.EXPONENTIAL_BACKOFF "Link to this definition")

Next retry is scheduled after retry\_interval\_seconds \* 2 ^ (attempt\_number - 1) seconds. This
means that the first attempt runs after retry\_interval\_seconds seconds, the second one after
2 \* retry\_interval\_seconds, then 4 \* retry\_interval\_seconds, etc.

NO\_RETRY _='NO\_RETRY'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod.NO_RETRY "Link to this definition")

Not retried.

_class_ flows\_sdk.error\_handling.ErrorRetryPolicy [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorRetryPolicy "Link to this definition")

Bases: `object`

Base class for all error retry policies.

method _: [`RetryMethod`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod "flows_sdk.error_handling.RetryMethod")_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorRetryPolicy.method "Link to this definition")

The algorithm that should be applied when calculating the delay between each retry attempt.

_class_ flows\_sdk.error\_handling.SimpleErrorRetryPolicy [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.SimpleErrorRetryPolicy "Link to this definition")

Bases: [`ErrorRetryPolicy`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorRetryPolicy "flows_sdk.error_handling.ErrorRetryPolicy")

Base class for simple error retry policies that are defined by their method, retry count
and a retry interval in seconds.

Available in v36 and later.

retry\_count _:`int`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.SimpleErrorRetryPolicy.retry_count "Link to this definition")

How many time a failed block should be retried. This means that if retry\_count=3 the block can
run up to 4 times within a single flow run - 1 original run + 3 retries.

retry\_interval\_seconds _:`float`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.SimpleErrorRetryPolicy.retry_interval_seconds "Link to this definition")

The base interval to use when calculating the delay between each retry attempt. Will result in
different delays depending on the selected retry method.

method _: [`RetryMethod`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod "flows_sdk.error_handling.RetryMethod")_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.SimpleErrorRetryPolicy.method "Link to this definition")

The algorithm that should be applied when calculating the delay between each retry attempt.

_class_ flows\_sdk.error\_handling.NoRetryErrorRetryPolicy [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.NoRetryErrorRetryPolicy "Link to this definition")

No retry, fail on the first unhandled error during execution. This policy can be used to
prevent retries for specific blocks in flows that define an error retry policy.

Available in v36 and later.

method _: [`RetryMethod`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.RetryMethod "flows_sdk.error_handling.RetryMethod")_ _='NO\_RETRY'_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.NoRetryErrorRetryPolicy.method "Link to this definition")

The algorithm that should be applied when calculating the delay between each retry attempt.

_class_ flows\_sdk.error\_handling.OnErrorFlow [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.OnErrorFlow "Link to this definition")

Bases: `object`

Specifies the details of the flow that should run in case a flow run fails. Note that the
OnErrorFlow would be called only in case of terminal failure, failures that will be retried
will not trigger it.

Check out the example for more details -
[Submission-aware OnError flow](https://flows-sdk.hyperscience.ai/pages/examples.html#submission-aware-on-error-flow)

**Available since v38.**

identifier _:`str`_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.OnErrorFlow.identifier "Link to this definition")

The identifier of the flow that should be called on error.

_class_ flows\_sdk.error\_handling.ErrorHandling [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorHandling "Link to this definition")

Bases: `object`

Container for error handling related concepts for blocks and flows.

Available in v36 and later.

block\_error\_retry\_policy _:`Union`\[ [`SimpleErrorRetryPolicy`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.SimpleErrorRetryPolicy "flows_sdk.error_handling.SimpleErrorRetryPolicy"), [`NoRetryErrorRetryPolicy`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.NoRetryErrorRetryPolicy "flows_sdk.error_handling.NoRetryErrorRetryPolicy"),`None`\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorHandling.block_error_retry_policy "Link to this definition")

This error retry policy will be applied to all blocks in the flow. Individual blocks can
override this using the same class.

Example

```
from uuid import UUID

from flows_sdk.blocks import CodeBlock
from flows_sdk.error_handling import (
    ErrorHandling,
    NoRetryErrorRetryPolicy,
    RetryMethod,
    SimpleErrorRetryPolicy,
)
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import workflow_input

BLOCK_RETRIES_FLOW_IDENTIFIER = 'BLOCK_RETRIES_SHOWCASE'

BLOCK_RETRIES_FLOW_UUID = UUID('59ebf286-686c-4df0-9b5f-f949607780ad')

class FlowInputs:
    N_SECONDS = 'n_seconds'

def entry_point_flow() -> Flow:
    return block_error_retries_showcase_flow()

def block_error_retries_showcase_flow() -> Flow:
    def _create_start_date() -> str:
        from datetime import datetime

return datetime.now().isoformat()

create_start_date = CodeBlock(
        reference_name='create_start_date',
        code=_create_start_date,
        code_input={},
    ).with_error_handling(
        ErrorHandling(
            # overrides the flow level retry policy for this block with NO_RETRY
            block_error_retry_policy=NoRetryErrorRetryPolicy()
        )
    )

def _fail_until_n_seconds_later(start_date: str, n_seconds: int) -> None:
        from datetime import datetime, timedelta

if datetime.now() < datetime.fromisoformat(start_date) + timedelta(seconds=n_seconds):
            raise Exception('Too early to complete!')

fail_until_n_seconds_later = CodeBlock(
        reference_name='fail_until_n_seconds_later',
        code=_fail_until_n_seconds_later,
        code_input={
            'start_date': create_start_date.output(),
            'n_seconds': workflow_input(FlowInputs.N_SECONDS),
        },
    )

return Flow(
        title='Block retries sample flow',
        description='Sample flow with a retry policy',
        blocks=[create_start_date, fail_until_n_seconds_later],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(
            identifier=BLOCK_RETRIES_FLOW_IDENTIFIER,
            input=[\
                Parameter(\
                    name=FlowInputs.N_SECONDS,\
                    title='Fail for this may seconds',\
                    type='integer',\
                    optional=False,\
                )\
            ],
        ),
        uuid=BLOCK_RETRIES_FLOW_UUID,
        input={FlowInputs.N_SECONDS: 30},
        error_handling=ErrorHandling(
            # retry up to 20 times with 5 seconds between attempts
            # will be applied to all blocks in the flow
            block_error_retry_policy=SimpleErrorRetryPolicy(
                method=RetryMethod.FIXED, retry_count=20, retry_interval_seconds=5
            )
        ),
    )

if __name__ == '__main__':
    export_flow(flow=entry_point_flow())
```

on\_error\_flow _:`Optional`\[ [`OnErrorFlow`](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.OnErrorFlow "flows_sdk.error_handling.OnErrorFlow")\]_ _=None_ [](https://flows-sdk.hyperscience.ai/pages/source-docs.html#flows_sdk.error_handling.ErrorHandling.on_error_flow "Link to this definition")

Specification for a flow that should run on error. Can be used only for error handling
attached on the flow level.

Available since v38.