Differential D8684 Diff 40807 examples/docs_snippets/docs_snippets/concepts/solids_pipelines/dynamic_pipeline/dynamic_pipeline.py

Changeset View

Standalone View

	# pylint: disable=unused-argument, no-value-for-parameter			# pylint: disable=unused-argument, no-value-for-parameter

	# start_marker			# start_marker
	import os			import os
	from typing import List			from typing import List

	from dagster import Field, pipeline, solid			from dagster import Field, pipeline, solid
	from dagster.experimental import DynamicOutput, DynamicOutputDefinition			from dagster.experimental import DynamicOutput, DynamicOutputDefinition
	from dagster.utils import file_relative_path			from dagster.utils import file_relative_path


	@solid(			@solid(
	config_schema={"path": Field(str, default_value=file_relative_path(__file__, "sample"))},			config_schema={
				"path": Field(
				str, default_value=file_relative_path(__file__, "sample")
				)
				},
	output_defs=[DynamicOutputDefinition(str)],			output_defs=[DynamicOutputDefinition(str)],
	)			)
	def files_in_directory(context):			def files_in_directory(context):
	path = context.solid_config["path"]			path = context.solid_config["path"]
	dirname, _, filenames = next(os.walk(path))			dirname, _, filenames = next(os.walk(path))
	for file in filenames:			for file in filenames:
	yield DynamicOutput(			yield DynamicOutput(
	value=os.path.join(dirname, file),			value=os.path.join(dirname, file),
	Show All 24 Lines