feat: Convert PDP ingestion notebooks into DAB

notebooks/nsc_sftp_automated_data_ingestion/01_sftp_receive_scan.ipynb

@@ -112,14 +112,14 @@
     "from pyspark.sql import functions as F\n",
     "\n",
     "from edvise.utils.sftp import connect_sftp, list_receive_files\n",
-    "from edvise.ingestion.constants import (\n",
+    "from edvise.ingestion.nsc_sftp.constants import (\n",
     "    MANIFEST_TABLE_PATH,\n",
     "    QUEUE_TABLE_PATH,\n",
     "    SFTP_REMOTE_FOLDER,\n",
     "    SFTP_SOURCE_SYSTEM,\n",
     "    SFTP_TMP_DIR,\n",
     ")\n",
-    "from edvise.ingestion.nsc_sftp_helpers import (\n",
+    "from edvise.ingestion.nsc_sftp.helpers import (\n",
     "    build_listing_df,\n",
     "    download_new_files_and_queue,\n",
     "    ensure_manifest_and_queue_tables,\n",

notebooks/nsc_sftp_automated_data_ingestion/02_file_institution_expand.ipynb

@@ -80,8 +80,8 @@
     "from pyspark.sql import types as T\n",
     "from databricks.connect import DatabricksSession\n",
     "\n",
-    "from edvise.ingestion.nsc_sftp_helpers import ensure_plan_table, extract_institution_ids\n",
-    "from edvise.ingestion.constants import (\n",
+    "from edvise.ingestion.nsc_sftp.helpers import ensure_plan_table, extract_institution_ids\n",
+    "from edvise.ingestion.nsc_sftp.constants import (\n",
     "    QUEUE_TABLE_PATH,\n",
     "    PLAN_TABLE_PATH,\n",
     "    COLUMN_RENAMES,\n",

notebooks/nsc_sftp_automated_data_ingestion/03_per_institution_bronze_ingest.ipynb

@@ -19,10 +19,10 @@
    "outputs": [],
    "source": [
     "# Databricks notebook source\n",
-    "# Script 4 — 04_per_institution_bronze_ingest\n",
+    "# Script 4 \u2014 04_per_institution_bronze_ingest\n",
     "#\n",
     "# Purpose:\n",
-    "#   Consume institution_ingest_plan (created by Script 3), and for each (file × institution):\n",
+    "#   Consume institution_ingest_plan (created by Script 3), and for each (file \u00d7 institution):\n",
     "#     - get bearer token from SST staging using X-API-KEY (from Databricks secrets)\n",
     "#     - call /api/v1/institutions/pdp-id/{pdp_id} to resolve institution name\n",
     "#     - map name -> schema prefix via databricksify_inst_name()\n",
@@ -104,14 +104,14 @@
     "from edvise.utils.databricks import (\n",
     "    find_bronze_schema,\n",
     "    find_bronze_volume_name,\n",
-    "    databricksify_inst_name,\n",
     ")\n",
+    "from edvise.utils.institution_naming import databricksify_inst_name\n",
     "from edvise.utils.sftp import output_file_name_from_sftp\n",
-    "from edvise.ingestion.nsc_sftp_helpers import (\n",
+    "from edvise.ingestion.nsc_sftp.helpers import (\n",
     "    process_and_save_file,\n",
     "    update_manifest,\n",
     ")\n",
-    "from edvise.ingestion.constants import (\n",
+    "from edvise.ingestion.nsc_sftp.constants import (\n",
     "    CATALOG,\n",
     "    PLAN_TABLE_PATH,\n",
     "    MANIFEST_TABLE_PATH,\n",

pipelines/ingestion/pdp/.gitignore

@@ -0,0 +1 @@
+.databricks

pipelines/ingestion/pdp/databricks.yml

@@ -0,0 +1,59 @@
+# Dedicated bundle folder: pipelines/ingestion/pdp/
+# Automates NSC/PDP SFTP → bronze ingestion (replaces interactive PIPELINE_pdp_to_databricks).
+bundle:
+  name: Edvise NSC SFTP ingestion
+  uuid: b8f4e2c1-9a3d-4f1e-b7c2-1d4e8f9a0b2c
+
+include:
+  - resources/nsc_sftp_ingestion.yml
+
+variables:
+  git_commit:
+    description: "Commit SHA to fetch (dev/CI)"
+    default: ""
+  git_tag:
+    description: "Release tag to fetch for deployment (prod)"
+    default: ""
+  DB_workspace:
+    description: "Unity Catalog name for NSC ingestion (tables and staging volume)"
+  ds_run_as:
+    description: "Service principal / app ID used as run_as for the job"
+  service_account_executer:
+    description: "Human or technical user granted CAN_MANAGE on the job"
+  datakind_group_to_manage_workflow:
+    description: "Workspace group with CAN_MANAGE on the job"
+  ingestion_slack_webhook_id:
+    description: "Slack webhook ID for ingestion pipeline failure notifications"
+
+run_as:
+  service_principal_name: ${var.ds_run_as}
+
+targets:
+  dev:
+    mode: development
+    default: true
+    variables:
+      DB_workspace: "dev_sst_02"
+      ingestion_slack_webhook_id: "1e30ca91-8d95-4324-be05-9d20bffd747b"  # edvise-data-crew on dev_sst_02
+    resources:
+      jobs:
+        nsc_sftp_automated_ingestion:
+          git_source:
+            git_url: https://github.com/datakind/edvise
+            git_provider: gitHub
+            git_commit: ${var.git_commit}
+
+  prod:
+    mode: production
+    variables:
+      DB_workspace: "staging_sst_01"
+      ingestion_slack_webhook_id: "c51b737e-988d-45dd-ad9b-6f9024a6b56b"  # edvise-support on staging_sst_01
+    workspace:
+      root_path: /Workspace/Shared/bundles/${bundle.name}/${bundle.target}
+    resources:
+      jobs:
+        nsc_sftp_automated_ingestion:
+          git_source:
+            git_url: https://github.com/datakind/edvise
+            git_provider: gitHub
+            git_tag: ${var.git_tag}

pipelines/ingestion/pdp/resources/nsc_sftp_ingestion.yml

@@ -0,0 +1,136 @@
+# NSC/PDP SFTP automated ingestion: Git-sourced spark_python_task chain (01→02→03).
+# Prerequisites: Job parameter DB_workspace (Unity Catalog), matching UC tables/volumes,
+# secret scope nsc-sftp-asset (SFTP + SST API key), and cluster egress for SFTP/APIs.
+#
+# File selection:
+#   - Set cohort_file_name + course_file_name for a manual run, OR
+#   - Leave them empty and use file_selection_mode=uningested|latest (default: uningested).
+
+resources:
+  jobs:
+    nsc_sftp_automated_ingestion:
+      name: nsc_sftp_automated_ingestion
+      max_concurrent_runs: 1
+      queue:
+        enabled: true
+      webhook_notifications:
+        on_failure:
+          - id: ${var.ingestion_slack_webhook_id}
+
+      parameters:
+        - name: DB_workspace
+          default: ${var.DB_workspace}
+        - name: file_selection_mode
+          default: uningested
+        - name: cohort_file_name
+          default: ""
+        - name: course_file_name
+          default: ""
+
+      tasks:
+        - task_key: sftp_receive_scan
+          spark_python_task:
+            python_file: src/edvise/ingestion/nsc_sftp/scripts/01_sftp_receive_scan.py
+            source: GIT
+            parameters:
+              - --DB_workspace
+              - "{{job.parameters.DB_workspace}}"
+              - --file_selection_mode
+              - "{{job.parameters.file_selection_mode}}"
+              - --cohort_file_name
+              - "{{job.parameters.cohort_file_name}}"
+              - --course_file_name
+              - "{{job.parameters.course_file_name}}"
+          job_cluster_key: nsc-sftp-ingestion-cluster
+          libraries:
+            - pypi:
+                package: pandas==2.2.3
+            - pypi:
+                package: numpy==1.26.4
+            - pypi:
+                package: pyarrow>=17.0.0
+            - pypi:
+                package: requests==2.32.5
+            - pypi:
+                package: pyyaml~=6.0
+            - pypi:
+                package: pydantic~=2.10
+            - pypi:
+                package: paramiko~=3.5
+
+        - task_key: file_institution_expand
+          depends_on:
+            - task_key: sftp_receive_scan
+          spark_python_task:
+            python_file: src/edvise/ingestion/nsc_sftp/scripts/02_file_institution_expand.py
+            source: GIT
+            parameters:
+              - --DB_workspace
+              - "{{job.parameters.DB_workspace}}"
+          job_cluster_key: nsc-sftp-ingestion-cluster
+          libraries:
+            - pypi:
+                package: pandas==2.2.3
+            - pypi:
+                package: numpy==1.26.4
+            - pypi:
+                package: pyarrow>=17.0.0
+            - pypi:
+                package: requests==2.32.5
+            - pypi:
+                package: pyyaml~=6.0
+            - pypi:
+                package: pydantic~=2.10
+
+        - task_key: per_institution_bronze_ingest
+          depends_on:
+            - task_key: file_institution_expand
+          spark_python_task:
+            python_file: src/edvise/ingestion/nsc_sftp/scripts/03_per_institution_bronze_ingest.py
+            source: GIT
+            parameters:
+              - --DB_workspace
+              - "{{job.parameters.DB_workspace}}"
+          job_cluster_key: nsc-sftp-ingestion-cluster
+          libraries:
+            - pypi:
+                package: pandas==2.2.3
+            - pypi:
+                package: numpy==1.26.4
+            - pypi:
+                package: pyarrow>=17.0.0
+            - pypi:
+                package: requests==2.32.5
+            - pypi:
+                package: pyyaml~=6.0
+            - pypi:
+                package: pydantic~=2.10
+
+      job_clusters:
+        - job_cluster_key: nsc-sftp-ingestion-cluster
+          new_cluster:
+            cluster_name: ""
+            spark_version: 15.4.x-cpu-ml-scala2.12
+            spark_conf:
+              spark.master: local[*, 4]
+              spark.databricks.cluster.profile: singleNode
+            gcp_attributes:
+              use_preemptible_executors: false
+              availability: ON_DEMAND_GCP
+              zone_id: HA
+            node_type_id: n2-standard-16
+            custom_tags:
+              ResourceClass: SingleNode
+              x-databricks-nextgen-cluster: "true"
+            enable_elastic_disk: true
+            data_security_mode: SINGLE_USER
+            runtime_engine: STANDARD
+            num_workers: 0
+
+      permissions:
+        - group_name: ${var.datakind_group_to_manage_workflow}
+          level: CAN_MANAGE
+        - service_principal_name: ${var.ds_run_as}
+          level: CAN_MANAGE
+        - user_name: ${var.service_account_executer}
+          level: CAN_MANAGE

src/edvise/ingestion/nsc_sftp/__init__.py

@@ -0,0 +1,5 @@
+"""NSC SFTP automated ingestion (constants, helpers, job scripts)."""
+
+from edvise.ingestion.nsc_sftp import constants, file_selection, helpers, runtime
+
+__all__ = ["constants", "file_selection", "helpers", "runtime"]