Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@lambda_solid(output_def=OutputDefinition(typing.List[int]))
def emit_list():
return [1]
def create_dummy_output_info(expect_def):
return dagster.OutputExpectationInfo(
solid=dagster.SolidDefinition(
name='dummy',
inputs=[],
transform_fn=lambda _context, _args: None,
output=dagster.OutputDefinition(expectations=[expect_def])
),
expectation_def=expect_def
)
@lambda_solid(output_def=OutputDefinition(String))
def return_str():
return 'foo'
output=OutputDefinition(Int),
)
def multer(num1, num2):
return num1 * num2
OutputDefinition(Nothing, 'complete'),
],
)
def yield_values(_context):
yield Output(1, 'num_1')
yield Output(2, 'num_2')
yield Output(None, 'complete')
@solid(outputs=[OutputDefinition(types.Int, 'num_one'), OutputDefinition(types.Int, 'num_two')])
def return_one_two(_info):
yield Result(1, 'num_one')
yield Result(2, 'num_two')
@lambda_solid(input_defs=[InputDefinition('num', Int)], output_def=OutputDefinition(Int))
def add_two(num):
return num + 2
def create_root_success_solid(name):
def root_fn(_context, _args):
passed_rows = []
passed_rows.append({name: 'compute_called'})
return passed_rows
return single_output_solid(
name=name, input_defs=[], compute_fn=root_fn, output_def=OutputDefinition()
)
@solid(output_defs=[OutputDefinition(int, 'three'), OutputDefinition(int, 'four')])
def return_three_and_four(_):
yield Output(3, 'three')
yield Output(4, 'four')
def define_hello_world_explicit_yield():
return dagstermill.define_dagstermill_solid(
'hello_world_explicit_yield_pipeline',
nb_test_path('hello_world_explicit_yield'),
[],
[OutputDefinition()],
)