Merge pull request #2171 from TEAMSchools/2112-dagsterdeanslistresour…

…ce-refactor-response-for-pagination 2112 dagsterdeanslistresource refactor response for pagination
TEAMSchools · Nov 12, 2024 · 04b5dec · 04b5dec
2 parents 422048b + 6773826
commit 04b5dec
Show file tree

Hide file tree

Showing 22 changed files with 279 additions and 179 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,3 +1,5 @@
+dagster
+
 # Created by https://www.toptal.com/developers/gitignore/api/linux,macos,windows,python,dbt
 # Edit at https://www.toptal.com/developers/gitignore?templates=linux,macos,windows,python,dbt
 

diff --git a/src/teamster/code_locations/kippcamden/deanslist/assets.py b/src/teamster/code_locations/kippcamden/deanslist/assets.py
@@ -1,5 +1,4 @@
 import pathlib
-import re
 
 from dagster import (
     MonthlyPartitionsDefinition,
@@ -9,10 +8,14 @@
 )
 
 from teamster.code_locations.kippcamden import CODE_LOCATION, LOCAL_TIMEZONE
-from teamster.code_locations.kippcamden.deanslist.schema import ASSET_SCHEMA
+from teamster.code_locations.kippcamden.deanslist.schema import (
+    ASSET_SCHEMA,
+    BEHAVIOR_SCHEMA,
+)
 from teamster.core.utils.classes import FiscalYearPartitionsDefinition
 from teamster.libraries.deanslist.assets import (
     build_deanslist_multi_partition_asset,
+    build_deanslist_paginated_multi_partition_asset,
     build_deanslist_static_partition_asset,
 )
 
@@ -45,11 +48,7 @@
 
 static_partitions_assets = [
     build_deanslist_static_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_STATIC_PARTITIONS_DEF,
         **e,
@@ -61,11 +60,8 @@
 
 monthly_multi_partitions_assets = [
     build_deanslist_multi_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
+        api_version="v1",
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_MONTHLY_MULTI_PARTITIONS_DEF,
         **e,
@@ -77,11 +73,8 @@
 
 fiscal_multi_partitions_assets = [
     build_deanslist_multi_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
+        api_version="v1",
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
         **e,
@@ -91,6 +84,16 @@
     ]
 ]
 
+behavior = build_deanslist_paginated_multi_partition_asset(
+    code_location=CODE_LOCATION,
+    endpoint="behavior",
+    api_version="v1",
+    schema=BEHAVIOR_SCHEMA,
+    partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
+)
+
+fiscal_multi_partitions_assets = [behavior, *fiscal_multi_partitions_assets]
+
 assets = [
     *static_partitions_assets,
     *monthly_multi_partitions_assets,

diff --git a/src/teamster/code_locations/kippcamden/deanslist/config/multi-partition-fiscal-assets.yaml b/src/teamster/code_locations/kippcamden/deanslist/config/multi-partition-fiscal-assets.yaml
@@ -1,5 +1,3 @@
 endpoints:
   - endpoint: comm-log
-    api_version: v1
   - endpoint: followups
-    api_version: v1
diff --git a/src/teamster/code_locations/kippcamden/deanslist/config/multi-partition-monthly-assets.yaml b/src/teamster/code_locations/kippcamden/deanslist/config/multi-partition-monthly-assets.yaml
@@ -1,20 +1,8 @@
 endpoints:
-  - endpoint: behavior
-    api_version: v1
-    params:
-      IncludeDeleted: Y
-    op_tags:
-      dagster-k8s/config:
-        container_config:
-          resources:
-            requests:
-              memory: 3.0Gi
   - endpoint: homework
-    api_version: v1
     params:
       IncludeDeleted: Y
   - endpoint: incidents
-    api_version: v1
     params:
       IncludeDeleted: Y
       cf: Y
diff --git a/src/teamster/code_locations/kippcamden/deanslist/schedules.py b/src/teamster/code_locations/kippcamden/deanslist/schedules.py
@@ -12,31 +12,31 @@
 from teamster.libraries.deanslist.schedules import build_deanslist_job_schedule
 
 deanslist_static_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_static_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist__static_partitions_assets_job",
     selection=static_partitions_assets,
     partitions_def=DEANSLIST_STATIC_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_monthly_multi_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_monthly_multi_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist__monthly_multi_partitions_assets_job",
     selection=monthly_multi_partitions_assets,
     partitions_def=DEANSLIST_MONTHLY_MULTI_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_fiscal_multi_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_fiscal_multi_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist__fiscal_multi_partitions_assets_job",
     selection=fiscal_multi_partitions_assets,
     partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_midday_commlog_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_midday_commlog_job",
+    job_name=f"{CODE_LOCATION}__deanslist__midday_commlog_job",
     selection=[AssetKey([CODE_LOCATION, "deanslist", "comm_log"])],
     partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
     cron_schedule="0 14 * * *",

diff --git a/src/teamster/code_locations/kippcamden/deanslist/schema.py b/src/teamster/code_locations/kippcamden/deanslist/schema.py
@@ -100,6 +100,10 @@ class incidents_record(Incident):
 
 pas_options = py_avro_schema.Option.NO_DOC | py_avro_schema.Option.NO_AUTO_NAMESPACE
 
+BEHAVIOR_SCHEMA = json.loads(
+    py_avro_schema.generate(py_type=behavior_record, options=pas_options)
+)
+
 ASSET_SCHEMA = {
     "comm-log": json.loads(py_avro_schema.generate(py_type=CommLog)),
     "lists": json.loads(py_avro_schema.generate(py_type=ListModel)),
@@ -114,9 +118,6 @@ class incidents_record(Incident):
     "reconcile_suspensions": json.loads(
         py_avro_schema.generate(py_type=ReconcileSuspensions)
     ),
-    "behavior": json.loads(
-        py_avro_schema.generate(py_type=behavior_record, options=pas_options)
-    ),
     "followups": json.loads(
         py_avro_schema.generate(py_type=followups_record, options=pas_options)
     ),

diff --git a/src/teamster/code_locations/kippmiami/deanslist/assets.py b/src/teamster/code_locations/kippmiami/deanslist/assets.py
@@ -1,5 +1,4 @@
 import pathlib
-import re
 
 from dagster import (
     MonthlyPartitionsDefinition,
@@ -9,10 +8,14 @@
 )
 
 from teamster.code_locations.kippmiami import CODE_LOCATION, LOCAL_TIMEZONE
-from teamster.code_locations.kippmiami.deanslist.schema import ASSET_SCHEMA
+from teamster.code_locations.kippmiami.deanslist.schema import (
+    ASSET_SCHEMA,
+    BEHAVIOR_SCHEMA,
+)
 from teamster.core.utils.classes import FiscalYearPartitionsDefinition
 from teamster.libraries.deanslist.assets import (
     build_deanslist_multi_partition_asset,
+    build_deanslist_paginated_multi_partition_asset,
     build_deanslist_static_partition_asset,
 )
 
@@ -43,11 +46,7 @@
 
 static_partitions_assets = [
     build_deanslist_static_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_STATIC_PARTITIONS_DEF,
         **e,
@@ -59,11 +58,8 @@
 
 monthly_multi_partitions_assets = [
     build_deanslist_multi_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
+        api_version="v1",
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_MONTHLY_MULTI_PARTITIONS_DEF,
         **e,
@@ -75,11 +71,8 @@
 
 fiscal_multi_partitions_assets = [
     build_deanslist_multi_partition_asset(
-        asset_key=[
-            CODE_LOCATION,
-            "deanslist",
-            re.sub(pattern=r"\W", repl="_", string=e["endpoint"]),
-        ],
+        code_location=CODE_LOCATION,
+        api_version="v1",
         schema=ASSET_SCHEMA[e["endpoint"]],
         partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
         **e,
@@ -89,6 +82,16 @@
     ]
 ]
 
+behavior = build_deanslist_paginated_multi_partition_asset(
+    code_location=CODE_LOCATION,
+    endpoint="behavior",
+    api_version="v1",
+    schema=BEHAVIOR_SCHEMA,
+    partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
+)
+
+fiscal_multi_partitions_assets = [behavior, *fiscal_multi_partitions_assets]
+
 assets = [
     *static_partitions_assets,
     *monthly_multi_partitions_assets,

diff --git a/src/teamster/code_locations/kippmiami/deanslist/config/multi-partition-fiscal-assets.yaml b/src/teamster/code_locations/kippmiami/deanslist/config/multi-partition-fiscal-assets.yaml
@@ -1,5 +1,3 @@
 endpoints:
   - endpoint: comm-log
-    api_version: v1
   - endpoint: followups
-    api_version: v1
diff --git a/src/teamster/code_locations/kippmiami/deanslist/config/multi-partition-monthly-assets.yaml b/src/teamster/code_locations/kippmiami/deanslist/config/multi-partition-monthly-assets.yaml
@@ -1,14 +1,8 @@
 endpoints:
-  - endpoint: behavior
-    api_version: v1
-    params:
-      IncludeDeleted: Y
   - endpoint: homework
-    api_version: v1
     params:
       IncludeDeleted: Y
   - endpoint: incidents
-    api_version: v1
     params:
       IncludeDeleted: Y
       cf: Y
diff --git a/src/teamster/code_locations/kippmiami/deanslist/schedules.py b/src/teamster/code_locations/kippmiami/deanslist/schedules.py
@@ -12,31 +12,31 @@
 from teamster.libraries.deanslist.schedules import build_deanslist_job_schedule
 
 deanslist_static_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_static_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist____static_partitions_assets_job",
     selection=static_partitions_assets,
     partitions_def=DEANSLIST_STATIC_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_monthly_multi_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_monthly_multi_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist__monthly_multi_partitions_assets_job",
     selection=monthly_multi_partitions_assets,
     partitions_def=DEANSLIST_MONTHLY_MULTI_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_fiscal_multi_partitions_assets_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_fiscal_multi_partitions_assets_job",
+    job_name=f"{CODE_LOCATION}__deanslist__fiscal_multi_partitions_assets_job",
     selection=fiscal_multi_partitions_assets,
     partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
     cron_schedule="0 0 * * *",
     execution_timezone=LOCAL_TIMEZONE.name,
 )
 
 deanslist_midday_commlog_job_schedule = build_deanslist_job_schedule(
-    job_name=f"{CODE_LOCATION}_deanslist_midday_commlog_job",
+    job_name=f"{CODE_LOCATION}__deanslist__midday_commlog_job",
     selection=[AssetKey([CODE_LOCATION, "deanslist", "comm_log"])],
     partitions_def=DEANSLIST_FISCAL_MULTI_PARTITIONS_DEF,
     cron_schedule="0 14 * * *",

diff --git a/src/teamster/code_locations/kippmiami/deanslist/schema.py b/src/teamster/code_locations/kippmiami/deanslist/schema.py
@@ -100,6 +100,10 @@ class incidents_record(Incident):
 
 pas_options = py_avro_schema.Option.NO_DOC | py_avro_schema.Option.NO_AUTO_NAMESPACE
 
+BEHAVIOR_SCHEMA = json.loads(
+    py_avro_schema.generate(py_type=behavior_record, options=pas_options)
+)
+
 ASSET_SCHEMA = {
     "comm-log": json.loads(py_avro_schema.generate(py_type=CommLog)),
     "lists": json.loads(py_avro_schema.generate(py_type=ListModel)),
@@ -114,9 +118,6 @@ class incidents_record(Incident):
     "reconcile_suspensions": json.loads(
         py_avro_schema.generate(py_type=ReconcileSuspensions)
     ),
-    "behavior": json.loads(
-        py_avro_schema.generate(py_type=behavior_record, options=pas_options)
-    ),
     "followups": json.loads(
         py_avro_schema.generate(py_type=followups_record, options=pas_options)
     ),