add example with custom blueprints

examples/experimental/dagster-blueprints/examples/custom-blueprints/custom_blueprints/__init__.py

@@ -0,0 +1,4 @@
+from .definitions import (
+    defs as defs,
+    loader as loader,
+)

examples/experimental/dagster-blueprints/examples/custom-blueprints/custom_blueprints/curl_assets/cereals.yaml

@@ -0,0 +1,3 @@
+type: curl_asset
+table_name: cereals
+csv_url: https://raw.githubusercontent.com/dagster-io/dagster/master/docs/next/public/assets/cereal.csv

examples/experimental/dagster-blueprints/examples/custom-blueprints/custom_blueprints/curl_assets/customers.yaml

@@ -0,0 +1,3 @@
+type: curl_asset
+table_name: customers
+csv_url: https://raw.githubusercontent.com/dagster-io/dagster/master/docs/next/public/assets/customers.csv

examples/experimental/dagster-blueprints/examples/custom-blueprints/custom_blueprints/definitions.py

@@ -0,0 +1,47 @@
+import tempfile
+from pathlib import Path
+from subprocess import Popen
+from typing import Literal
+
+from dagster import asset
+from dagster_blueprints import YamlBlueprintsLoader
+from dagster_blueprints.blueprint import Blueprint, BlueprintDefinitions
+
+
+def local_csv_to_snowflake(table_name: str, csv_path: Path) -> None:
+    """Left as an exercise to the reader."""
+
+
+def curl_csv_to_snowflake(table_name: str, csv_url: str) -> None:
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        path = Path(tmp_dir) / "file.csv"
+        command = ["curl", "-o", str(path), csv_url]
+        process = Popen(command)
+        process.wait()
+
+        if process.returncode != 0:
+            raise ValueError(f"External execution process failed with code {process.returncode}")
+
+        local_csv_to_snowflake(table_name, path)
+
+
+class CurlCsvSnowflakeAssetBlueprint(Blueprint):
+    """Blueprint for an asset in Snowflake, populated from a CSV file on the internet."""
+
+    type: Literal["curl_asset"]
+    csv_url: str
+    table_name: str
+
+    def build_defs(self) -> BlueprintDefinitions:
+        @asset(key=self.table_name)
+        def _asset():
+            curl_csv_to_snowflake(table_name=self.table_name, csv_url=self.csv_url)
+
+        return BlueprintDefinitions(assets=[_asset])
+
+
+loader = YamlBlueprintsLoader(
+    per_file_blueprint_type=CurlCsvSnowflakeAssetBlueprint,
+    path=Path(__file__).parent / "curl_assets",
+)
+defs = loader.load_defs()

examples/experimental/dagster-blueprints/examples/custom-blueprints/pyproject.toml

@@ -0,0 +1,6 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[tool.dagster]
+module_name = "custom_blueprints"

examples/experimental/dagster-blueprints/examples/custom-blueprints/setup.cfg

@@ -0,0 +1,2 @@
+[metadata]
+name = custom_blueprints

examples/experimental/dagster-blueprints/examples/custom-blueprints/setup.py

@@ -0,0 +1,12 @@
+from setuptools import find_packages, setup
+
+setup(
+    name="custom-blueprints",
+    packages=find_packages(exclude=["custom-blueprints"]),
+    install_requires=[
+        "dagster",
+        "dagster-blueprints",
+        "dagster-webserver",
+    ],
+    extras_require={"dev": ["pytest"]},
+)