0
# Artifacts
1

2
dbt generates structured artifacts containing metadata about projects, execution results, and data catalogs. These artifacts enable state comparison, documentation generation, and integration with external tools.
3

4
## Artifact Schema Modules
5

6
### Base Schemas
7

8
```python { .api }
9
# dbt.artifacts.schemas.base
10
"""Base artifact schema classes providing common functionality."""
11

12
# dbt.artifacts.schemas.results  
13
"""Result artifact schemas for execution outcomes."""
14

15
# dbt.artifacts.schemas.batch_results
16
"""Batch result schemas for grouped execution results."""
17
```
18

19
### Specialized Artifact Types
20

21
```python { .api }
22
# dbt.artifacts.schemas.catalog
23
"""Catalog artifact schemas containing database metadata and column information."""
24

25
# dbt.artifacts.schemas.freshness
26
"""Source freshness schemas for data freshness monitoring results."""
27

28
# dbt.artifacts.schemas.manifest
29
"""Manifest artifact schemas defining project structure and compiled resources."""
30

31
# dbt.artifacts.schemas.run
32
"""Run result schemas containing detailed execution information."""
33
```
34

35
## Core Artifact Types
36

37
### Manifest Artifact
38

39
The manifest is dbt's compiled representation of a project, containing all nodes, relationships, and metadata.
40

41
```python { .api }
42
# Generated by: dbt parse, dbt compile, dbt run, etc.
43
# Location: target/manifest.json
44
# Schema: dbt.artifacts.schemas.manifest
45

46
class Manifest:
47
    """
48
    Complete compiled representation of a dbt project.
49
    
50
    Contains all nodes, sources, metrics, and their relationships
51
    after compilation and dependency resolution.
52
    """
53
    
54
    nodes: Dict[str, Any]          # All project nodes (models, tests, etc.)
55
    sources: Dict[str, Any]        # Source definitions
56
    macros: Dict[str, Any]         # Macro definitions
57
    metrics: Dict[str, Any]        # Metric definitions
58
    exposures: Dict[str, Any]      # Exposure definitions
59
    parent_map: Dict[str, List[str]]  # Node dependencies
60
    child_map: Dict[str, List[str]]   # Node dependents
61
```
62

63
### Run Results Artifact
64

65
Contains detailed information about execution results from dbt commands.
66

67
```python { .api }
68
# Generated by: dbt run, dbt test, dbt build, dbt seed, etc.
69
# Location: target/run_results.json  
70
# Schema: dbt.artifacts.schemas.run
71

72
class RunExecutionResult:
73
    """
74
    Results from executing dbt commands like run, test, build.
75
    
76
    Contains timing information, success/failure status, and
77
    detailed results for each executed node.
78
    """
79
    
80
    results: List[RunResult]       # Individual node results
81
    elapsed_time: float           # Total execution time
82
    args: Dict[str, Any]          # Command arguments used
83
```
84

85
class RunResult:
86
    """Result for a single node execution."""
87
    
88
    unique_id: str                # Node unique identifier
89
    status: str                   # 'success', 'error', 'skipped', etc.
90
    execution_time: float         # Node execution time
91
    adapter_response: Dict        # Database adapter response
92
    message: Optional[str]        # Result message
93
    failures: Optional[int]       # Number of test failures (for tests)
94
```
95
96
### Catalog Artifact
97

98
Contains database metadata and column information discovered during documentation generation.
99

100
```python { .api }
101
# Generated by: dbt docs generate
102
# Location: target/catalog.json
103
# Schema: dbt.artifacts.schemas.catalog
104

105
class CatalogArtifact:
106
    """
107
    Database catalog information including table and column metadata.
108
    
109
    Generated during docs generation by introspecting the database
110
    to gather actual table structures and statistics.
111
    """
112
    
113
    nodes: Dict[str, CatalogTable]    # Table metadata by node
114
    sources: Dict[str, CatalogTable]  # Source table metadata
115
```
116

117
class CatalogTable:
118
    """Metadata for a single table or view."""
119
    
120
    columns: Dict[str, CatalogColumn]  # Column information
121
    stats: Dict[str, CatalogStats]     # Table statistics
122
    metadata: Dict[str, Any]           # Additional metadata
123
```
124
125
### Source Freshness Results
126

127
Results from source freshness checks.
128

129
```python { .api }
130
# Generated by: dbt source freshness
131
# Location: target/sources.json
132
# Schema: dbt.artifacts.schemas.freshness
133

134
class FreshnessExecutionResult:
135
    """
136
    Results from source freshness checks.
137
    
138
    Contains freshness status and timing information
139
    for each checked source.
140
    """
141
    
142
    results: List[FreshnessResult]  # Individual source results
143
    elapsed_time: float            # Total check time
144
```
145

146
## Accessing Artifacts
147

148
### Reading Manifest
149

150
```python
151
import json
152
from pathlib import Path
153

154
def load_manifest():
155
    """Load the compiled manifest."""
156
    manifest_path = Path('target/manifest.json')
157
    if manifest_path.exists():
158
        with open(manifest_path) as f:
159
            return json.load(f)
160
    return None
161

162
manifest = load_manifest()
163
if manifest:
164
    # Access nodes
165
    models = {k: v for k, v in manifest['nodes'].items() 
166
              if v['resource_type'] == 'model'}
167
    print(f"Found {len(models)} models")
168
    
169
    # Access sources
170
    sources = manifest.get('sources', {})
171
    print(f"Found {len(sources)} sources")
172
```
173

174
### Reading Run Results
175

176
```python
177
def load_run_results():
178
    """Load the latest run results."""
179
    results_path = Path('target/run_results.json')
180
    if results_path.exists():
181
        with open(results_path) as f:
182
            return json.load(f)
183
    return None
184

185
run_results = load_run_results()
186
if run_results:
187
    # Analyze results
188
    total_nodes = len(run_results['results'])
189
    successful = sum(1 for r in run_results['results'] 
190
                    if r['status'] == 'success')
191
    
192
    print(f"Executed {total_nodes} nodes, {successful} successful")
193
    print(f"Total time: {run_results['elapsed_time']:.2f}s")
194
```
195

196
### Programmatic Access
197

198
Access artifacts through the dbtRunner:
199

200
```python
201
from dbt.cli.main import dbtRunner
202

203
runner = dbtRunner()
204

205
# Generate manifest
206
parse_result = runner.invoke(['parse'])
207
if parse_result.success:
208
    manifest = parse_result.result
209
    
210
    # Access manifest data
211
    print(f"Nodes: {len(manifest.nodes)}")
212
    print(f"Sources: {len(manifest.sources)}")
213
    print(f"Macros: {len(manifest.macros)}")
214

215
# Generate run results  
216
run_result = runner.invoke(['run'])
217
if run_result.success:
218
    execution_result = run_result.result
219
    
220
    # Access run data
221
    print(f"Execution time: {execution_result.elapsed_time}")
222
    for node_result in execution_result.results:
223
        print(f"{node_result.unique_id}: {node_result.status}")
224
```
225

226
## Artifact File Locations
227

228
### Standard Locations
229

230
```python
231
from dbt.constants import (
232
    MANIFEST_FILE_NAME,           # "manifest.json"
233
    RUN_RESULTS_FILE_NAME,        # "run_results.json"
234
    CATALOG_FILENAME,             # "catalog.json"
235
    SOURCE_RESULT_FILE_NAME,      # "sources.json"
236
)
237

238
# Standard artifact paths (relative to project root)
239
target_dir = Path('target')
240
manifest_path = target_dir / MANIFEST_FILE_NAME
241
run_results_path = target_dir / RUN_RESULTS_FILE_NAME
242
catalog_path = target_dir / CATALOG_FILENAME
243
sources_path = target_dir / SOURCE_RESULT_FILE_NAME
244
```
245

246
### Custom Artifact Paths
247

248
```python
249
# Artifacts can be written to custom locations
250
runner = dbtRunner()
251

252
# Custom target path
253
result = runner.invoke(['run'], target_path='custom_target')
254

255
# Custom output paths for specific commands
256
result = runner.invoke([
257
    'source', 'freshness', 
258
    '--output-path', 'freshness_results.json'
259
])
260
```
261

262
## State Comparison
263

264
Use artifacts for state-based selection and comparison:
265

266
```python
267
# Run only changed models
268
result = runner.invoke([
269
    'run', 
270
    '--select', 'state:modified',
271
    '--state', 'path/to/previous/artifacts'
272
])
273

274
# Test only new resources
275
result = runner.invoke([
276
    'test',
277
    '--select', 'state:new',
278
    '--state', 'path/to/previous/artifacts'
279
])
280
```
281

282
## Integration Examples
283

284
### CI/CD Pipeline
285

286
```python
287
def ci_pipeline():
288
    """Example CI pipeline using artifacts."""
289
    runner = dbtRunner()
290
    
291
    # Parse project and check for issues
292
    parse_result = runner.invoke(['parse'])
293
    if not parse_result.success:
294
        return False
295
    
296
    manifest = parse_result.result
297
    
298
    # Run only changed models in CI
299
    run_result = runner.invoke([
300
        'run', 
301
        '--select', 'state:modified+',
302
        '--state', 'prod_artifacts/'
303
    ])
304
    
305
    if run_result.success:
306
        # Save artifacts for downstream use
307
        execution_result = run_result.result
308
        save_artifacts(execution_result)
309
    
310
    return run_result.success
311
```
312

313
### Custom Analytics
314

315
```python
316
def analyze_project_health():
317
    """Analyze project health from artifacts."""
318
    manifest = load_manifest()
319
    run_results = load_run_results()
320
    
321
    if not (manifest and run_results):
322
        return
323
    
324
    # Model analysis
325
    models = [n for n in manifest['nodes'].values() 
326
              if n['resource_type'] == 'model']
327
    
328
    # Execution analysis
329
    execution_times = {}
330
    for result in run_results['results']:
331
        execution_times[result['unique_id']] = result.get('execution_time', 0)
332
    
333
    # Find slow models
334
    slow_models = sorted(execution_times.items(), 
335
                        key=lambda x: x[1], reverse=True)[:10]
336
    
337
    print("Slowest models:")
338
    for node_id, time in slow_models:
339
        print(f"  {node_id}: {time:.2f}s")
340
```
341

342
### Artifact Validation
343

344
```python
345
def validate_artifacts():
346
    """Validate artifact completeness and consistency."""
347
    manifest = load_manifest()
348
    run_results = load_run_results()
349
    
350
    if not manifest:
351
        print("Warning: No manifest found")
352
        return False
353
    
354
    if not run_results:
355
        print("Warning: No run results found")
356
        return False
357
    
358
    # Check consistency
359
    manifest_nodes = set(manifest['nodes'].keys())
360
    result_nodes = set(r['unique_id'] for r in run_results['results'])
361
    
362
    missing = manifest_nodes - result_nodes
363
    if missing:
364
        print(f"Warning: {len(missing)} nodes missing from results")
365
    
366
    return len(missing) == 0
367
```