Tweaks after testing

Author: shaidar

dg_projects/edxorg/edxorg/assets/instructor_onboarding.py

@@ -8,7 +8,14 @@
 from datetime import UTC, datetime
 
 import polars as pl
-from dagster import AssetExecutionContext, AssetIn, AssetKey, Output, asset
+from dagster import (
+    AssetExecutionContext,
+    AssetIn,
+    AssetKey,
+    MaterializeResult,
+    Output,
+    asset,
+)
 from github.GithubException import UnknownObjectException
 from ol_orchestrate.resources.github import GithubApiClientFactory
 
@@ -81,29 +88,31 @@ def generate_instructor_onboarding_user_list(
 
 @asset(
     name="update_access_forge_repo",
-    group_name="instructor_onboarding",
-    ins={"instructor_onboarding_user_list": AssetIn()},
-    required_resource_keys={"github_api"},
+    group_name="edxorg_instructor_onboarding",
+    ins={
+        "instructor_onboarding_user_list": AssetIn(
+            key=AssetKey(["instructor_onboarding_user_list"])
+        )
+    },
     description="Updates the access-forge repository with the generated user list",
 )
-def update_access_forge_repository(
+def update_access_forge_repo(
     context: AssetExecutionContext,
-    instructor_onboarding_user_list: str,
     github_api: GithubApiClientFactory,
-) -> Output[dict]:
-    """Push the generated CSV content to the access-forge GitHub repository.
+    instructor_onboarding_user_list: str,
+) -> MaterializeResult:
+    """Update access-forge repository with instructor user list.
 
     This asset updates or creates a CSV file in the private mitodl/access-forge
     repository with the user list generated from the dbt model.
 
     Args:
         context: Dagster execution context
-        instructor_onboarding_user_list: CSV string content to upload with columns:
-            email, role, sent_invite
         github_api: GitHub API client factory resource for authentication
+        instructor_onboarding_user_list: CSV string content from upstream asset
 
     Returns:
-        Output containing metadata about the commit (repo, file path, action, SHA)
+        MaterializeResult containing metadata about the commit
 
     Raises:
         Exception: If GitHub API call fails or authentication issues occur
@@ -119,8 +128,8 @@ def update_access_forge_repository(
     commit_message = "dagster-pipeline - update user list from ol-data-platform"
 
     try:
-        gh_client = github_api.get_client()
-        repo = gh_client.get_repo(repo_name)
+        github_client = github_api.get_client()
+        repo = github_client.get_repo(repo_name)
 
         # Get the base branch reference
         base_ref = repo.get_git_ref(f"heads/{base_branch}")

dg_projects/edxorg/edxorg/definitions.py

@@ -11,6 +11,7 @@
 from typing import Any, Literal
 
 from dagster import (
+    AssetExecutionContext,
     AssetSelection,
     DefaultSensorStatus,
     Definitions,
@@ -20,11 +21,15 @@
     job,
 )
 from dagster_aws.s3 import S3Resource
+from dagster_dbt import DbtCliResource, dbt_assets
 from ol_orchestrate.io_managers.filepath import (
     FileObjectIOManager,
     S3FileObjectIOManager,
 )
 from ol_orchestrate.lib.constants import DAGSTER_ENV, VAULT_ADDRESS
+
+# Import dbt project
+from ol_orchestrate.lib.dbt_project import dbt_project
 from ol_orchestrate.lib.utils import authenticate_vault
 from ol_orchestrate.resources.gcp_gcs import GCSConnection
 from ol_orchestrate.resources.github import GithubApiClientFactory
@@ -49,7 +54,7 @@
 )
 from edxorg.assets.instructor_onboarding import (
     generate_instructor_onboarding_user_list,
-    update_access_forge_repository,
+    update_access_forge_repo,
 )
 from edxorg.assets.openedx_course_archives import (
     dummy_edxorg_course_xml,
@@ -175,6 +180,15 @@ def sync_edxorg_program_reports():
     upload_files_to_s3(download_files_from_s3())
 
 
+@dbt_assets(
+    manifest=dbt_project.manifest_path,
+    project=dbt_project,
+)
+def edxorg_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
+    """Dbt models for edxorg data transformations."""
+    yield from dbt.cli(["build"], context=context).stream()
+
+
 # Sensors
 try:
     from edxorg.sensors.object_storage import (
@@ -228,7 +242,7 @@ def sync_edxorg_program_reports():
             edxorg_program_metadata,
             edxorg_mitx_course_metadata,
             generate_instructor_onboarding_user_list,
-            update_access_forge_repository,
+            update_access_forge_repo,
         ),
     ),
     cron_schedule="0 5 * * *",
@@ -265,6 +279,7 @@ def sync_edxorg_program_reports():
         "s3_download": S3Resource(profile_name="edxorg"),
         "s3_upload": S3Resource(),
         "results_dir": SimpleResultsDir.configure_at_launch(),
+        "dbt": DbtCliResource(project_dir=dbt_project),
     },
     sensors=sensor_list,
     jobs=[
@@ -273,6 +288,7 @@ def sync_edxorg_program_reports():
         sync_edxorg_program_reports,
     ],
     assets=[
+        edxorg_dbt_assets,
         edxorg_raw_data_archive.to_source_asset(),
         edxorg_raw_tracking_logs.to_source_asset(),
         normalize_edxorg_tracking_log,
@@ -283,7 +299,7 @@ def sync_edxorg_program_reports():
         edxorg_program_metadata,
         edxorg_mitx_course_metadata,
         generate_instructor_onboarding_user_list,
-        update_access_forge_repository,
+        update_access_forge_repo,
     ],
     schedules=[edxorg_api_daily_schedule],
 )

dg_projects/edxorg/pyproject.toml

@@ -5,13 +5,15 @@ version = "0.1.0"
 dependencies = [
     "dagster>=1.11.13",
     "dagster-aws ~=0.27.0",
+    "dagster-dbt>=0.27.0",
     "dagster-duckdb ~=0.27.0",
     "dagster-postgres>=0.27.13",
     "flatten-dict ~= 0.4.2",
     "jsonlines ~= 4.0.0",
     "ol-orchestrate-lib",
     "polars ~= 1.19",
     "universal-pathlib ~= 0.3.1",
+    "pygithub>=2.8.1",
 ]
 
 [project.entry-points]