|
8 | 8 |
|
9 | 9 | import os |
10 | 10 | from functools import partial |
| 11 | +from pathlib import Path |
11 | 12 | from typing import Any, Literal |
12 | 13 |
|
13 | 14 | from dagster import ( |
|
21 | 22 | job, |
22 | 23 | ) |
23 | 24 | from dagster_aws.s3 import S3Resource |
24 | | -from dagster_dbt import DbtCliResource, dbt_assets |
| 25 | +from dagster_dbt import DbtCliResource, DbtProject, dbt_assets |
25 | 26 | from ol_orchestrate.io_managers.filepath import ( |
26 | 27 | FileObjectIOManager, |
27 | 28 | S3FileObjectIOManager, |
28 | 29 | ) |
29 | 30 | from ol_orchestrate.lib.constants import DAGSTER_ENV, VAULT_ADDRESS |
30 | | - |
31 | | -# Import dbt project |
32 | | -from ol_orchestrate.lib.dbt_project import dbt_project |
33 | 31 | from ol_orchestrate.lib.utils import authenticate_vault |
34 | 32 | from ol_orchestrate.resources.gcp_gcs import GCSConnection |
35 | 33 | from ol_orchestrate.resources.github import GithubApiClientFactory |
|
69 | 67 | upload_files_to_s3, |
70 | 68 | ) |
71 | 69 |
|
| 70 | +# Initialize dbt project - handle both local dev and Docker paths |
| 71 | +if Path("/app/ol_dbt").exists(): |
| 72 | + # In Docker container |
| 73 | + DBT_PROJECT_DIR = Path("/app/ol_dbt") |
| 74 | +else: |
| 75 | + # Local development |
| 76 | + DBT_PROJECT_DIR = Path(__file__).resolve().parents[3] / "src" / "ol_dbt" |
| 77 | + |
| 78 | +dbt_project = DbtProject(project_dir=DBT_PROJECT_DIR) |
| 79 | + |
72 | 80 | # Initialize vault with resilient loading |
73 | 81 | try: |
74 | 82 | vault = authenticate_vault(DAGSTER_ENV, VAULT_ADDRESS) |
@@ -288,7 +296,7 @@ def edxorg_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource): |
288 | 296 | sync_edxorg_program_reports, |
289 | 297 | ], |
290 | 298 | assets=[ |
291 | | - edxorg_dbt_assets, |
| 299 | + edxorg_dbt_assets, # Re-enable dbt assets |
292 | 300 | edxorg_raw_data_archive.to_source_asset(), |
293 | 301 | edxorg_raw_tracking_logs.to_source_asset(), |
294 | 302 | normalize_edxorg_tracking_log, |
|
0 commit comments