dbt select job blueprints example

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/README.md

@@ -0,0 +1,39 @@
+## Example: writing a custom blueprint type for defining jobs from dbt selections
+
+Imagine that analytics engineers in your organization want to be able to easily create executable dbt selections.
+
+Ideally, all they need to write to set up one of these is:
+
+```yaml
+- type: dbt_select_job
+  name: marketing_models
+  select: "tag:marketing"
+```
+
+Custom blueprint types can help with this. Using them involves two kinds of files:
+
+- The YAML files themselves, which contain blobs that look like the above. E.g. [dbt_job_blueprints/dbt_jobs.yaml](dbt_job_blueprints/dbt_jobs.yaml).
+- Python code that defines our custom blueprint type and uses it to load these YAML files into Dagster definitions. This is located in the [dbt_job_blueprints/definitions.py](dbt_job_blueprints/definitions.py) file.
+
+### Try it out
+
+Make sure the blueprints library is installed, using the instructions [here](../../README.md#install).
+
+Install the example:
+
+```python
+pip install -e .
+```
+
+Launch Dagster to see the definitions loaded from the blueprints:
+
+```bash
+
+dagster dev
+```
+
+Print out the JSON schema for the blueprints:
+
+```bash
+dagster-blueprints print-schema
+```

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/dbt_job_blueprints/__init__.py

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

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/dbt_job_blueprints/definitions.py

@@ -0,0 +1,44 @@
+import os
+from pathlib import Path
+from typing import Literal, Optional
+
+from dagster import AssetExecutionContext, Definitions, define_asset_job
+from dagster_blueprints import YamlBlueprintsLoader
+from dagster_blueprints.blueprint import Blueprint, BlueprintDefinitions
+from dagster_dbt import DbtCliResource, build_dbt_asset_selection, dbt_assets
+
+dbt_project_dir = Path(__file__).parent.parent / "jaffle_shop"
+dbt_manifest_path = dbt_project_dir / "target" / "manifest.json"
+
+
+@dbt_assets(manifest=dbt_manifest_path)
+def jaffle_shop_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
+    yield from dbt.cli(["build"], context=context).stream()
+
+
+class DbtSelectJobBlueprint(Blueprint):
+    type: Literal["dbt_select_job"]
+    name: str
+    select: str
+    exclude: Optional[str]
+
+    def build_defs(self) -> BlueprintDefinitions:
+        job_def = define_asset_job(
+            name=self.name,
+            selection=build_dbt_asset_selection(
+                [jaffle_shop_dbt_assets], dbt_select=self.select, dbt_exclude=self.exclude
+            ),
+        )
+        return BlueprintDefinitions(jobs=[job_def])
+
+
+loader = YamlBlueprintsLoader(
+    per_file_blueprint_type=DbtSelectJobBlueprint,
+    path=Path(__file__).parent / "dbt_jobs.yaml",
+)
+resources = {"dbt": DbtCliResource(project_dir=os.fspath(dbt_project_dir))}
+
+defs = Definitions.merge(
+    Definitions(assets=[jaffle_shop_dbt_assets], resources=resources),
+    loader.load_defs(resources=resources),
+)

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/pyproject.toml

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

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/setup.cfg

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

examples/experimental/dagster-blueprints/examples/dbt-job-blueprints/setup.py

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