Differential D8684 Diff 40955 examples/docs_snippets/docs_snippets_tests/intro_tutorial_tests/advanced/test_warehouse.py

Changeset View

Standalone View

View Options

examples/docs_snippets/docs_snippets_tests/intro_tutorial_tests/advanced/test_warehouse.py

	import csv			import csv
	import os			import os

	from dagster import execute_pipeline			from dagster import execute_pipeline
	from dagster.utils import pushd, script_relative_path			from dagster.utils import pushd, script_relative_path
	from docs_snippets.intro_tutorial.advanced.pipelines.modes import (			from docs_snippets.intro_tutorial.advanced.pipelines.modes import (
	SqlAlchemyPostgresWarehouse as sapw1,			SqlAlchemyPostgresWarehouse as sapw1,
	)			)
	from docs_snippets.intro_tutorial.advanced.pipelines.modes import modes_pipeline			from docs_snippets.intro_tutorial.advanced.pipelines.modes import (
				modes_pipeline,
				)
	from docs_snippets.intro_tutorial.advanced.pipelines.presets import (			from docs_snippets.intro_tutorial.advanced.pipelines.presets import (
	SqlAlchemyPostgresWarehouse as sapw2,			SqlAlchemyPostgresWarehouse as sapw2,
	)			)
	from docs_snippets.intro_tutorial.advanced.pipelines.presets import presets_pipeline			from docs_snippets.intro_tutorial.advanced.pipelines.presets import (
				presets_pipeline,
				)
	from docs_snippets.intro_tutorial.test_util import patch_cereal_requests			from docs_snippets.intro_tutorial.test_util import patch_cereal_requests

	BUILDKITE = bool(os.getenv("BUILDKITE"))			BUILDKITE = bool(os.getenv("BUILDKITE"))


	def test_warehouse(postgres):			def test_warehouse(postgres):
	with open(script_relative_path("../../../docs_snippets/intro_tutorial/cereal.csv"), "r") as fd:			with open(
				script_relative_path(
				"../../../docs_snippets/intro_tutorial/cereal.csv"
				),
				"r",
				) as fd:
	cereals = [row for row in csv.DictReader(fd)]			cereals = [row for row in csv.DictReader(fd)]

	for SqlAlchemyPostgresWarehouse in [sapw1, sapw2]:			for SqlAlchemyPostgresWarehouse in [sapw1, sapw2]:
	warehouse = SqlAlchemyPostgresWarehouse(postgres)			warehouse = SqlAlchemyPostgresWarehouse(postgres)
	warehouse.update_normalized_cereals(cereals)			warehouse.update_normalized_cereals(cereals)


	def test_warehouse_resource(postgres):			def test_warehouse_resource(postgres):
	run_config = {			run_config = {
	"solids": {"download_csv": {"inputs": {"url": {"value": "something"}}}},			"solids": {
				"download_csv": {"inputs": {"url": {"value": "something"}}}
				},
	"resources": {"warehouse": {"config": {"conn_str": postgres}}},			"resources": {"warehouse": {"config": {"conn_str": postgres}}},
	}			}

	@patch_cereal_requests			@patch_cereal_requests
	def execute_with_mode():			def execute_with_mode():
	result = execute_pipeline(			result = execute_pipeline(
	pipeline=modes_pipeline,			pipeline=modes_pipeline,
	mode="dev",			mode="dev",
	Show All 14 Lines