0
# Component System
1
2
Integration with Dagster's component system for declarative dbt project configuration. The component system provides a way to define dbt projects as reusable components within Dagster code locations.
3
4
## Capabilities
5
6
### dbt Project Component
7
8
#### DbtProjectComponent
9
10
Dagster component for declarative dbt project configuration and management.
11
12
```python { .api }
13
class DbtProjectComponent(Component):
14
"""
15
Dagster component for dbt project integration.
16
17
Provides declarative configuration for dbt projects within Dagster
18
code locations, enabling automatic asset discovery and resource
19
management.
20
21
Attributes:
22
- dbt_project_path: Path to the dbt project directory
23
- dbt_profiles_path: Path to the dbt profiles directory (optional)
24
- dbt_target: dbt target name (optional)
25
- dbt_parse_on_load: Whether to parse project on component load
26
"""
27
28
dbt_project_path: str
29
dbt_profiles_path: Optional[str] = None
30
dbt_target: Optional[str] = None
31
dbt_parse_on_load: bool = True
32
33
def build_defs(self) -> Definitions:
34
"""
35
Build Dagster definitions from dbt project.
36
37
Returns:
38
Definitions object containing dbt assets and resources
39
"""
40
41
def build_assets(self) -> Sequence[AssetsDefinition]:
42
"""
43
Build asset definitions from dbt project.
44
45
Returns:
46
Sequence of AssetsDefinition objects for dbt models
47
"""
48
49
def build_resources(self) -> Mapping[str, ResourceDefinition]:
50
"""
51
Build resource definitions for dbt integration.
52
53
Returns:
54
Mapping of resource names to ResourceDefinition objects
55
"""
56
```
57
58
## Usage Examples
59
60
### Basic Component Configuration
61
62
```python
63
from dagster import Definitions
64
from dagster_dbt.components import DbtProjectComponent
65
66
# Define dbt project component
67
dbt_component = DbtProjectComponent(
68
dbt_project_path="./my_dbt_project",
69
dbt_target="dev"
70
)
71
72
# Build definitions from component
73
defs = dbt_component.build_defs()
74
```
75
76
### Multi-Environment Component
77
78
```python
79
import os
80
from dagster_dbt.components import DbtProjectComponent
81
82
class EnvironmentDbtComponent(DbtProjectComponent):
83
def __init__(self, project_path: str):
84
environment = os.getenv("DAGSTER_ENV", "dev")
85
86
super().__init__(
87
dbt_project_path=project_path,
88
dbt_target=environment,
89
dbt_profiles_path=f"./profiles/{environment}"
90
)
91
92
def build_defs(self) -> Definitions:
93
"""Build environment-specific definitions."""
94
base_defs = super().build_defs()
95
96
# Add environment-specific configuration
97
return Definitions(
98
assets=base_defs.assets,
99
resources=base_defs.resources,
100
jobs=base_defs.jobs,
101
schedules=base_defs.schedules,
102
sensors=base_defs.sensors
103
)
104
105
# Use environment component
106
component = EnvironmentDbtComponent("./my_dbt_project")
107
defs = component.build_defs()
108
```
109
110
### Component with Custom Assets
111
112
```python
113
from dagster import asset, AssetExecutionContext
114
from dagster_dbt.components import DbtProjectComponent
115
from dagster_dbt import DbtCliResource
116
117
class CustomDbtComponent(DbtProjectComponent):
118
def build_assets(self) -> Sequence[AssetsDefinition]:
119
"""Build assets with custom pre/post processing."""
120
# Get base dbt assets
121
dbt_assets = super().build_assets()
122
123
# Add custom preprocessing asset
124
@asset
125
def preprocess_data(context: AssetExecutionContext):
126
"""Preprocess data before dbt run."""
127
context.log.info("Preprocessing data for dbt")
128
# Custom preprocessing logic
129
return {"status": "preprocessed"}
130
131
# Add custom postprocessing asset
132
@asset(deps=dbt_assets)
133
def postprocess_results(context: AssetExecutionContext):
134
"""Postprocess dbt results."""
135
context.log.info("Postprocessing dbt results")
136
# Custom postprocessing logic
137
return {"status": "postprocessed"}
138
139
return [preprocess_data, *dbt_assets, postprocess_results]
140
141
component = CustomDbtComponent(
142
dbt_project_path="./my_dbt_project",
143
dbt_target="prod"
144
)
145
defs = component.build_defs()
146
```
147
148
### Component Factory Pattern
149
150
```python
151
from dagster_dbt.components import DbtProjectComponent
152
from typing import Dict, Any
153
154
def create_dbt_component(config: Dict[str, Any]) -> DbtProjectComponent:
155
"""Factory function for creating dbt components from configuration."""
156
return DbtProjectComponent(
157
dbt_project_path=config["project_path"],
158
dbt_profiles_path=config.get("profiles_path"),
159
dbt_target=config.get("target", "dev"),
160
dbt_parse_on_load=config.get("parse_on_load", True)
161
)
162
163
# Configuration-driven component creation
164
project_configs = [
165
{
166
"project_path": "./analytics",
167
"target": "prod",
168
"profiles_path": "./profiles/prod"
169
},
170
{
171
"project_path": "./ml_models",
172
"target": "dev",
173
"parse_on_load": False
174
}
175
]
176
177
components = [create_dbt_component(config) for config in project_configs]
178
all_defs = [component.build_defs() for component in components]
179
180
# Combine all definitions
181
from dagster import Definitions
182
183
combined_assets = []
184
combined_resources = {}
185
186
for defs in all_defs:
187
combined_assets.extend(defs.assets or [])
188
combined_resources.update(defs.resources or {})
189
190
final_defs = Definitions(
191
assets=combined_assets,
192
resources=combined_resources
193
)
194
```
195
196
## Type Definitions
197
198
```python { .api }
199
from dagster import Component, Definitions, AssetsDefinition, ResourceDefinition
200
from typing import Optional, Sequence, Mapping
201
```