[component tree] Example of x-component access using dbt (dagster-io#31146)

## Summary

Showcases the utility of x-component accesses using dbt - adds a new method to the `DbtProjectComponent` which can be invoked from other components to get an asset key from a dbt model name:

```yaml
type: MyComponent

attributes:
  assets:
    - key: my_asset_name
      deps:
        - "{{ load_component_at_path('jaffle_shop_dbt').get_asset_key_for_model('customers') }}"
```

~~Extremely verbose at the moment~~, but otherwise feels pretty good to me.

## Test Plan

New unit tests of using this utility from YAML and from Python. New docs.

Author: benpankow

docs/docs/guides/build/components/integrations/dbt-component-tutorial.md

@@ -75,3 +75,22 @@ You can customize the properties of the assets emitted by each dbt model using t
 <WideContent maxSize={1100}>
 <CliInvocationExample path="docs_snippets/docs_snippets/guides/components/integrations/dbt-component/13-list-defs.txt" />
 </WideContent>
+
+## 7. Depending on dbt assets in other components
+
+If you want to refer to assets built by the dbt component elsewhere in your Dagster project, you can use the `asset_key_for_model` method on the dbt component.
+This lets you refer to an asset by the model name without having to know how that model is translated to an asset key.
+
+Imagine a `PythonScriptComponent` that exports the `customers` model to a CSV file:
+
+<CliInvocationExample path="docs_snippets/docs_snippets/guides/components/integrations/dbt-component/14-scaffold-python-script-component.txt" />
+
+<CliInvocationExample path="docs_snippets/docs_snippets/guides/components/integrations/dbt-component/15-touch-export-customers.txt" />
+
+You can refer to the `customers` asset in this component by using the `asset_key_for_model` method on the dbt component:
+
+<CodeExample path="docs_snippets/docs_snippets/guides/components/integrations/dbt-component/16-component.yaml" title="my_project/defs/my_python_script/defs.yaml" language="yaml" />
+
+<WideContent maxSize={1100}>
+<CliInvocationExample path="docs_snippets/docs_snippets/guides/components/integrations/dbt-component/17-list-defs.txt" />
+</WideContent>

examples/docs_snippets/docs_snippets/guides/components/integrations/dbt-component/14-scaffold-python-script-component.txt

@@ -0,0 +1,3 @@
+dg scaffold defs dagster.PythonScriptComponent my_python_script
+
+Creating defs at /.../my-project/src/my_project/defs/my_python_script.

examples/docs_snippets/docs_snippets/guides/components/integrations/dbt-component/15-touch-export-customers.txt

@@ -0,0 +1 @@
+touch src/my_project/defs/my_python_script/export_customers.py

examples/docs_snippets/docs_snippets/guides/components/integrations/dbt-component/16-component.yaml

@@ -0,0 +1,9 @@
+type: dagster.PythonScriptComponent
+
+attributes:
+  execution:
+    path: my_script.py
+  assets:
+    - key: customers_export
+      deps:
+        - "{{ load_component_at_path('dbt_ingest').asset_key_for_model('customers') }}"

examples/docs_snippets/docs_snippets/guides/components/integrations/dbt-component/17-list-defs.txt

@@ -0,0 +1,28 @@
+dg list defs
+
+┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
+┃ Section      ┃ Definitions                                                                                           ┃
+┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
+│ Assets       │ ┏━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ │
+│              │ ┃ Key              ┃ Group      ┃ Deps          ┃ Kinds  ┃ Description                              ┃ │
+│              │ ┡━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩ │
+│              │ │ customers        │ dbt_models │ stg_customers │ dbt    │ Transforms data using dbt model          │ │
+│              │ │                  │            │ stg_orders    │ duckdb │ customers                                │ │
+│              │ │                  │            │ stg_payments  │        │                                          │ │
+│              │ ├──────────────────┼────────────┼───────────────┼────────┼──────────────────────────────────────────┤ │
+│              │ │ customers_export │ default    │ customers     │        │                                          │ │
+│              │ ├──────────────────┼────────────┼───────────────┼────────┼──────────────────────────────────────────┤ │
+│              │ │ stg_customers    │ default    │               │        │                                          │ │
+│              │ ├──────────────────┼────────────┼───────────────┼────────┼──────────────────────────────────────────┤ │
+│              │ │ stg_orders       │ default    │               │        │                                          │ │
+│              │ ├──────────────────┼────────────┼───────────────┼────────┼──────────────────────────────────────────┤ │
+│              │ │ stg_payments     │ default    │               │        │                                          │ │
+│              │ └──────────────────┴────────────┴───────────────┴────────┴──────────────────────────────────────────┘ │
+│ Asset Checks │ ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━┓                                │
+│              │ ┃ Key                                      ┃ Deps      ┃ Description ┃                                │
+│              │ ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━┩                                │
+│              │ │ customers:not_null_customers_customer_id │ customers │             │                                │
+│              │ ├──────────────────────────────────────────┼───────────┼─────────────┤                                │
+│              │ │ customers:unique_customers_customer_id   │ customers │             │                                │
+│              │ └──────────────────────────────────────────┴───────────┴─────────────┘                                │
+└──────────────┴───────────────────────────────────────────────────────────────────────────────────────────────────────┘

examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/integrations/test_dbt_component.py

@@ -175,3 +175,36 @@ def test_components_docs_dbt_project(
         _run_command(
             cmd="dg launch --assets '*'",
         )
+
+        # scaffold a PythonScriptComponent
+        context.run_command_and_snippet_output(
+            cmd="dg scaffold defs dagster.PythonScriptComponent my_python_script",
+            snippet_path=f"{context.get_next_snip_number()}-scaffold-python-script-component.txt",
+        )
+        context.run_command_and_snippet_output(
+            cmd="touch src/my_project/defs/my_python_script/export_customers.py",
+            snippet_path=f"{context.get_next_snip_number()}-touch-export-customers.txt",
+        )
+
+        context.create_file(
+            Path("src") / "my_project" / "defs" / "my_python_script" / "defs.yaml",
+            contents=textwrap.dedent(
+                """\
+                type: dagster.PythonScriptComponent
+
+                attributes:
+                  execution:
+                    path: my_script.py
+                  assets:
+                    - key: customers_export
+                      deps:
+                        - "{{ load_component_at_path('dbt_ingest').asset_key_for_model('customers') }}"
+                """
+            ),
+            snippet_path=f"{context.get_next_snip_number()}-component.yaml",
+        )
+
+        context.run_command_and_snippet_output(
+            cmd="dg list defs",
+            snippet_path=f"{context.get_next_snip_number()}-list-defs.txt",
+        )

python_modules/libraries/dagster-dbt/dagster_dbt/components/dbt_project/component.py

@@ -5,9 +5,11 @@
 from typing import Annotated, Any, Callable, Optional, Union
 
 from dagster import Resolvable
+from dagster._core.definitions.asset_key import AssetKey
 from dagster._core.definitions.assets.definition.asset_spec import AssetSpec
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
+from dagster._utils.cached_method import cached_method
 from dagster.components.component.component import Component
 from dagster.components.core.context import ComponentLoadContext
 from dagster.components.core.tree import ComponentTree
@@ -21,9 +23,15 @@
 from dagster_dbt.asset_utils import DBT_DEFAULT_EXCLUDE, DBT_DEFAULT_SELECT, get_node
 from dagster_dbt.components.dbt_project.scaffolder import DbtProjectComponentScaffolder
 from dagster_dbt.core.resource import DbtCliResource
-from dagster_dbt.dagster_dbt_translator import DagsterDbtTranslator, DagsterDbtTranslatorSettings
+from dagster_dbt.dagster_dbt_translator import (
+    DagsterDbtTranslator,
+    DagsterDbtTranslatorSettings,
+    validate_translator,
+)
+from dagster_dbt.dbt_manifest import validate_manifest
 from dagster_dbt.dbt_manifest_asset_selection import DbtManifestAssetSelection
 from dagster_dbt.dbt_project import DbtProject
+from dagster_dbt.utils import ASSET_RESOURCE_TYPES
 
 TranslationFn: TypeAlias = Callable[[AssetSpec, Mapping[str, Any]], AssetSpec]
 
@@ -236,6 +244,34 @@ def _fn(context: AssetExecutionContext):
     def execute(self, context: AssetExecutionContext, dbt: DbtCliResource) -> Iterator:
         yield from dbt.cli(["build"], context=context).stream()
 
+    @cached_property
+    def _validated_manifest(self):
+        return validate_manifest(self.project.manifest_path)
+
+    @cached_property
+    def _validated_translator(self):
+        return validate_translator(self.translator)
+
+    @cached_method
+    def asset_key_for_model(self, model_name: str) -> AssetKey:
+        dagster_dbt_translator = self._validated_translator
+        manifest = self._validated_manifest
+
+        matching_model_ids = [
+            unique_id
+            for unique_id, value in manifest["nodes"].items()
+            if value["name"] == model_name and value["resource_type"] in ASSET_RESOURCE_TYPES
+        ]
+
+        if len(matching_model_ids) == 0:
+            raise KeyError(f"Could not find a dbt model, seed, or snapshot with name: {model_name}")
+
+        return dagster_dbt_translator.get_asset_spec(
+            manifest,
+            next(iter(matching_model_ids)),
+            self.project,
+        ).key
+
 
 class ProxyDagsterDbtTranslator(DagsterDbtTranslator):
     # get_description conflicts on Component, so cant make it directly a translator

python_modules/libraries/dagster-dbt/dagster_dbt_tests/components/code_locations/dependency_on_dbt_project_location/defs/depends_on_dbt/definitions.py

@@ -1,17 +1,18 @@
+from pathlib import Path
+
 import dagster as dg
-from dagster.components.core.context import ComponentLoadContext
-from dagster_dbt.asset_utils import get_asset_key_for_model
+from dagster.components.core.defs_module import ComponentPath
+from dagster_dbt import DbtProjectComponent
 
 
 @dg.definitions
-def defs(context: ComponentLoadContext) -> dg.Definitions:
-    dbt_assets = (
-        context.component_tree.build_defs_at_path("jaffle_shop_dbt")
-        .resolve_asset_graph()
-        .assets_defs
-    )
+def defs(context: dg.ComponentLoadContext) -> dg.Definitions:
+    customers_asset_key = context.component_tree.load_component_at_path(
+        ComponentPath(file_path=Path("jaffle_shop_dbt"), instance_key=0),
+        expected_type=DbtProjectComponent,
+    ).asset_key_for_model("customers")
 
-    @dg.asset(deps={get_asset_key_for_model(dbt_assets, "customers")})
+    @dg.asset(deps={customers_asset_key})
     def downstream_of_customers():
         pass
 

python_modules/libraries/dagster-dbt/dagster_dbt_tests/components/code_locations/dependency_on_dbt_project_location/defs/depends_on_dbt_yaml/component.py

@@ -0,0 +1,19 @@
+from collections.abc import Iterable
+from typing import Union
+
+import dagster as dg
+from dagster.components.lib.executable_component.component import ExecutableComponent
+from dagster.components.resolved.core_models import OpSpec
+
+
+class MyExecutableComponent(ExecutableComponent):
+    def invoke_execute_fn(
+        self,
+        context: Union[dg.AssetExecutionContext, dg.AssetCheckExecutionContext],
+        component_load_context: dg.ComponentLoadContext,
+    ) -> Iterable:
+        return []
+
+    @property
+    def op_spec(self) -> OpSpec:
+        return OpSpec(name="my_executable_component")

python_modules/libraries/dagster-dbt/dagster_dbt_tests/components/code_locations/dependency_on_dbt_project_location/defs/depends_on_dbt_yaml/defs.yaml

@@ -0,0 +1,7 @@
+type: dependency_on_dbt_project_location.defs.depends_on_dbt_yaml.component.MyExecutableComponent
+
+attributes:
+  assets:
+    - key: downstream_of_customers_two
+      deps:
+        - "{{ load_component_at_path('jaffle_shop_dbt').asset_key_for_model('customers') }}"