""" 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)