344 use dagster embedded elt to sync illuminate data

.devcontainer/scripts/postCreate.sh

@@ -55,6 +55,10 @@ op inject -f --in-file=.devcontainer/tpl/dbt_cloud.yml.tpl \
   --out-file=env/dbt_cloud.yml &&
   sudo mv -f env/dbt_cloud.yml /home/vscode/.dbt/dbt_cloud.yml
 
+op inject -f --in-file=.devcontainer/tpl/gcloud_teamster_dlt_keyfile.json.tpl \
+  --out-file=env/gcloud_teamster_dlt_keyfile.json &&
+  sudo mv -f env/gcloud_teamster_dlt_keyfile.json /etc/secret-volume/gcloud_teamster_dlt_keyfile.json
+
 # install uv
 curl -LsSf https://astral.sh/uv/install.sh | sh || true
 

.devcontainer/tpl/.env.tpl

@@ -39,6 +39,9 @@ DEANSLIST_SFTP_HOST=op://Data Team/DeansList SFTP/host
 DEANSLIST_SFTP_PASSWORD=op://Data Team/DeansList SFTP/password
 DEANSLIST_SFTP_USERNAME=op://Data Team/DeansList SFTP/username
 DEANSLIST_SUBDOMAIN=op://Data Team/DeansList API/subdomain
+DESTINATION__BIGQUERY__CREDENTIALS__CLIENT_EMAIL=op://Data Team/Teamster Service Account - dlt/username
+DESTINATION__BIGQUERY__CREDENTIALS__PRIVATE_KEY=op://Data Team/Teamster Service Account - dlt/password
+DESTINATION__BIGQUERY__CREDENTIALS__PROJECT_ID=op://Data Team/Teamster Service Account - dlt/project id
 EDPLAN_SFTP_HOST=op://Data Team/edplan SFTP - Camden/host
 EDPLAN_SFTP_PASSWORD_KIPPCAMDEN=op://Data Team/edplan SFTP - Camden/password
 EDPLAN_SFTP_PASSWORD_KIPPNEWARK=op://Data Team/edplan SFTP - Newark/password
@@ -126,6 +129,12 @@ SCHOOLMINT_GROW_CLIENT_SECRET=op://Data Team/SchooMint Grow API/client secret
 SCHOOLMINT_GROW_DISTRICT_ID=op://Data Team/SchooMint Grow API/district id
 SLACK_TOKEN=op://Data Team/Slack API - Teamster/credential
 SMARTRECRUITERS_SMARTTOKEN=op://Data Team/SmartRecruiters API/smart token
+SOURCES__SQL_DATABASE__CREDENTIALS__DATABASE=op://Data Team/Illuminate ODBC/database
+SOURCES__SQL_DATABASE__CREDENTIALS__DRIVERNAME=op://Data Team/Illuminate ODBC/driver
+SOURCES__SQL_DATABASE__CREDENTIALS__HOST=op://Data Team/Illuminate ODBC/ip
+SOURCES__SQL_DATABASE__CREDENTIALS__PASSWORD=op://Data Team/Illuminate ODBC/password
+SOURCES__SQL_DATABASE__CREDENTIALS__PORT=op://Data Team/Illuminate ODBC/port
+SOURCES__SQL_DATABASE__CREDENTIALS__USERNAME=op://Data Team/Illuminate ODBC/username
 TABLEAU_PERSONAL_ACCESS_TOKEN=op://Data Team/Tableau Server PAT - Dagster/credential
 TABLEAU_SERVER_ADDRESS=op://Data Team/Tableau Server PAT - Dagster/hostname
 TABLEAU_SITE_ID=op://Data Team/Tableau Server PAT - Dagster/site id

.devcontainer/tpl/gcloud_teamster_dlt_keyfile.json.tpl

@@ -0,0 +1 @@
+op://Data Team/Teamster Service Account - dlt/keyfile.json

.trunk/trunk.yaml

@@ -41,6 +41,12 @@ lint:
       linters:
         - sqlfluff
         - sqlfmt
+    - paths:
+        - src/teamster/libraries/dlt_sources/**
+      linters:
+        - pyright
+        - sqlfluff
+        - sqlfmt
     - paths:
         - src/dbt/**
       linters:
@@ -52,6 +58,7 @@ lint:
         - markdownlint
         - osv-scanner
         - oxipng
+        - pyright
         - ruff
         - shellcheck
         - shfmt

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
   "dagster-k8s",
   "dagster-slack",
   "dagster-ssh",
+  "dagster-embedded-elt",
   "dbt-bigquery>=1.8,<1.9",
   "dbt-core>=1.8,<1.9",
   "beautifulsoup4>=4.12.2",
@@ -26,6 +27,7 @@ dependencies = [
   "gspread>=5.12.0",
   "ldap3>=2.9.1",
   "oracledb>=1.4.2",
+  "psycopg[binary,pool]>=3.2.3",
   "py-avro-schema>=3.4.1",
   "pycryptodome>=3.19.0",
   "pypdf>=5.0.0",

src/teamster/code_locations/kipptaf/dagster-cloud.yaml

@@ -41,6 +41,11 @@ locations:
                     items:
                       - key: service_account_gserviceaccount.json
                         path: gcloud_service_account_json
+                - secret:
+                    name: op-gcp-service-account-dlt
+                    items:
+                      - key: keyfile.json
+                        path: gcloud_teamster_dlt_keyfile.json
         server_k8s_config:
           container_config:
             resources:
@@ -379,6 +384,31 @@ locations:
                   secretKeyRef:
                     name: op-slack-api
                     key: credential
+              - name: ILLUMINATE_DB_DRIVERNAME
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: driver
+              - name: ILLUMINATE_DB_HOST
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: ip
+              - name: ILLUMINATE_DB_DATABASE
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: database
+              - name: ILLUMINATE_DB_USERNAME
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: username
+              - name: ILLUMINATE_DB_PASSWORD
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: password
         run_k8s_config:
           container_config:
             resources:
@@ -717,3 +747,28 @@ locations:
                   secretKeyRef:
                     name: op-slack-api
                     key: credential
+              - name: ILLUMINATE_DB_DRIVERNAME
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: driver
+              - name: ILLUMINATE_DB_HOST
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: ip
+              - name: ILLUMINATE_DB_DATABASE
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: database
+              - name: ILLUMINATE_DB_USERNAME
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: username
+              - name: ILLUMINATE_DB_PASSWORD
+                valueFrom:
+                  secretKeyRef:
+                    name: op-illuminate-db
+                    key: password

src/teamster/code_locations/kipptaf/definitions.py

@@ -5,6 +5,7 @@
     build_sensor_for_freshness_checks,
     load_assets_from_modules,
 )
+from dagster_embedded_elt.dlt import DagsterDltResource
 from dagster_k8s import k8s_job_executor
 
 from teamster.code_locations.kipptaf import (
@@ -19,6 +20,7 @@
     deanslist,
     extracts,
     fivetran,
+    illuminate,
     ldap,
     overgrad,
     performance_management,
@@ -54,6 +56,7 @@
                 amplify,
                 extracts,
                 deanslist,
+                illuminate,
                 ldap,
                 overgrad,
                 performance_management,
@@ -97,6 +100,7 @@
         "db_bigquery": BIGQUERY_RESOURCE,
         "dbt_cli": get_dbt_cli_resource(DBT_PROJECT),
         "dds": resources.DIBELS_DATA_SYSTEM_RESOURCE,
+        "dlt": DagsterDltResource(),
         "fivetran": resources.FIVETRAN_RESOURCE,
         "gcs": GCS_RESOURCE,
         "google_directory": resources.GOOGLE_DIRECTORY_RESOURCE,

src/teamster/code_locations/kipptaf/illuminate/__init__.py

@@ -0,0 +1,5 @@
+from teamster.code_locations.kipptaf.illuminate.assets import assets
+
+__all__ = [
+    "assets",
+]

src/teamster/code_locations/kipptaf/illuminate/assets.py

@@ -0,0 +1,45 @@
+import json
+
+from dagster import AssetExecutionContext, EnvVar, _check
+from dagster_embedded_elt.dlt import DagsterDltResource, dlt_assets
+from dlt import pipeline
+from dlt.sources.sql_database import sql_database
+from sqlalchemy import URL, create_engine
+
+
+@dlt_assets(
+    dlt_source=sql_database(
+        credentials=create_engine(
+            url=URL.create(
+                drivername=_check.not_none(
+                    value=EnvVar("ILLUMINATE_DB_DRIVERNAME").get_value()
+                ),
+                host=EnvVar("ILLUMINATE_DB_HOST").get_value(),
+                database=EnvVar("ILLUMINATE_DB_DATABASE").get_value(),
+                username=EnvVar("ILLUMINATE_DB_USERNAME").get_value(),
+                password=EnvVar("ILLUMINATE_DB_PASSWORD").get_value(),
+            )
+        ),
+        schema="dna_assessments",
+        table_names=["assessments", "agg_student_responses_standard"],
+        defer_table_reflect=True,
+    ),
+    dlt_pipeline=pipeline(
+        pipeline_name="illuminate",
+        destination="bigquery",
+        dataset_name="dlt_illuminate_dna_assessments",
+        progress="log",
+    ),
+)
+def illuminate_dna_assessments(context: AssetExecutionContext, dlt: DagsterDltResource):
+    yield from dlt.run(
+        context=context,
+        credentials=json.load(
+            fp=open(file="/etc/secret-volume/gcloud_teamster_dlt_keyfile.json")
+        ),
+    )
+
+
+assets = [
+    illuminate_dna_assessments,
+]

src/teamster/code_locations/kipptaf/resources.py

@@ -1,5 +1,6 @@
 from dagster import EnvVar
 from dagster_airbyte import AirbyteCloudResource
+from dagster_embedded_elt.dlt import DagsterDltResource
 from dagster_fivetran import FivetranResource
 
 from teamster.libraries.adp.workforce_manager.resources import (
@@ -50,6 +51,8 @@
     username=EnvVar("AMPLIFY_DDS_USERNAME"), password=EnvVar("AMPLIFY_DDS_PASSWORD")
 )
 
+DLT_RESOURCE = DagsterDltResource()
+
 FIVETRAN_RESOURCE = FivetranResource(
     api_key=EnvVar("FIVETRAN_API_KEY"),
     api_secret=EnvVar("FIVETRAN_API_SECRET"),