Demonstrates some limitations of passing secrets to blocks
import sys
from typing import Any
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.utils import str_to_deterministic_uuid_4, workflow_input
SECRETS_IN_CCB_FLOW_IDENTIFIER = 'SECRET_IN_CCB_FLOW_VARIANTS'
SECRET_IN_CCB_FLOW_UUID = str_to_deterministic_uuid_4(SECRETS_IN_CCB_FLOW_IDENTIFIER)
def entry_point_flow() -> Flow:
return secrets_in_ccb_flow()
class FlowInputs:
SECRET_FLOW_INPUT_OBJECT = 'secret_flow_input_object'
SECRET_FLOW_INPUT_STRING = 'secret_flow_input_string'
def secrets_in_ccb_flow() -> Flow:
def use_secret(secret: Any) -> dict:
# CAUTION: This is just an example to help experimenting with secrets by using
# View Flow Run UI.
# DO NOT return secrets from blocks in real code!!!
return {'received_secret_obj': secret}
# This is allowed and works, the secret can be a full-blown json object.
use_secret_first_level = CodeBlock(
reference_name='use_secret_first_level',
code=use_secret,
code_input={'secret': workflow_input(FlowInputs.SECRET_FLOW_INPUT_OBJECT)},
)
# Compiles and runs, but will not use the correct secret value.
# Check the block output in View Flow Run.
use_secret_deep_nesting = CodeBlock(
reference_name='use_secret_deep_nesting',
code=use_secret,
code_input={
'secret': {'deeply_nested': workflow_input(FlowInputs.SECRET_FLOW_INPUT_OBJECT)}
},
)
# Compiles and runs, but will not use the correct secret value.
# Check the block output in View Flow Run.
use_secret_substring = CodeBlock(
reference_name='use_secret_substring',
code=use_secret,
code_input={'secret': f'concatenate {workflow_input(FlowInputs.SECRET_FLOW_INPUT_STRING)}'},
)
return Flow(
title='Flow that demonstrates limitations of passing down of secrets in various ways',
description='Flow that demonstrates limitations of passing down of secrets in various ways',
blocks=[use_secret_first_level, use_secret_deep_nesting, use_secret_substring],
owner_email='flows.sdk@hyperscience.com',
manifest=Manifest(
identifier=SECRETS_IN_CCB_FLOW_IDENTIFIER,
input=[
Parameter(
name=FlowInputs.SECRET_FLOW_INPUT_OBJECT,
title='Secret object',
type='json',
secret=True,
optional=True,
),
Parameter(
name=FlowInputs.SECRET_FLOW_INPUT_STRING,
title='Secret string',
type='string',
secret=True,
optional=True,
),
],
),
uuid=SECRET_IN_CCB_FLOW_UUID,
)
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)