Merge pull request #1833 from TEAMSchools/600-slack-alerts-for-integr…

…ation-job-errors

600 slack alerts for integration job errors

.k8s/1password/items.yaml

@@ -350,3 +350,11 @@ metadata:
   namespace: dagster-cloud
 spec:
   itemPath: vaults/Data Team/items/Overgrad API - Camden
+---
+apiVersion: onepassword.com/v1
+kind: OnePasswordItem
+metadata:
+  name: op-slack-api
+  namespace: dagster-cloud
+spec:
+  itemPath: vaults/Data Team/items/Slack API - Teamster

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
   "scikit-learn>=1.4.0",
   "tableauserverclient>=0.25",
   "tenacity>=8.2.3",
+  "dagster-slack>=0.24.8",
 ]
 requires-python = ">=3.12,<3.13"
 license = { text = "GPL-3.0-or-later" }

requirements.txt

@@ -2,7 +2,7 @@
 # Please do not edit it manually.
 
 agate==1.9.1
-alembic==1.13.2
+alembic==1.13.3
 annotated-types==0.7.0
 attrs==24.2.0
 avro==1.12.0
@@ -31,6 +31,7 @@ dagster-gcp==0.24.8
 dagster-k8s==0.24.8
 dagster-pandas==0.24.8
 dagster-pipes==1.8.8
+dagster-slack==0.24.8
 dagster-ssh==0.24.8
 db-dtypes==1.3.0
 dbt-adapters==1.7.0
@@ -137,6 +138,7 @@ scipy==1.14.1
 setuptools==75.1.0
 shellingham==1.5.4
 six==1.16.0
+slack-sdk==3.33.1
 soupsieve==2.6
 sqlalchemy==2.0.35
 sqlglot[rs]==25.22.0

src/teamster/code_locations/kipptaf/_google/directory/assets.py

@@ -144,10 +144,6 @@ def members(context: AssetExecutionContext, google_directory: GoogleDirectoryRes
 ]
 
 assets = [
-    groups,
-    members,
-    orgunits,
-    role_assignments,
-    roles,
-    users,
+    *google_directory_nonpartitioned_assets,
+    *google_directory_partitioned_assets,
 ]

src/teamster/code_locations/kipptaf/_google/directory/jobs.py

@@ -14,7 +14,7 @@
 
 
 @job(
-    name=f"{CODE_LOCATION}_google_directory_user_sync_job",
+    name=f"{CODE_LOCATION}__google__directory__user_sync_job",
     config=RunConfig(
         ops={
             "bigquery_get_table_op": BigQueryGetTableOpConfig(
@@ -35,7 +35,7 @@ def google_directory_user_sync_job():
 
 
 @job(
-    name=f"{CODE_LOCATION}_google_directory_role_assignments_sync_job",
+    name=f"{CODE_LOCATION}__google__directory__role_assignments_sync_job",
     config=RunConfig(
         ops={
             "bigquery_get_table_op": BigQueryGetTableOpConfig(

src/teamster/code_locations/kipptaf/_google/directory/schedules.py

@@ -11,7 +11,7 @@
 
 google_directory_nonpartitioned_asset_schedule = ScheduleDefinition(
     job=define_asset_job(
-        name=f"{CODE_LOCATION}_google_directory_nonpartitioned_asset_job",
+        name=f"{CODE_LOCATION}__google__directory__nonpartitioned_asset_job",
         selection=[a.key for a in google_directory_nonpartitioned_assets],
     ),
     cron_schedule="0 1 * * *",

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

@@ -709,3 +709,8 @@ locations:
                   secretKeyRef:
                     name: op-overgrad-api-kippnewark
                     key: credential
+              - name: SLACK_TOKEN
+                valueFrom:
+                  secretKeyRef:
+                    name: op-slack-api
+                    key: credential

src/teamster/code_locations/kipptaf/definitions.py

@@ -32,6 +32,7 @@
     BIGQUERY_RESOURCE,
     GCS_RESOURCE,
     OVERGRAD_RESOURCE,
+    SLACK_RESOURCE,
     SSH_COUCHDROP,
     get_dbt_cli_resource,
     get_io_manager_gcs_avro,
@@ -110,6 +111,7 @@
         "overgrad": OVERGRAD_RESOURCE,
         "ps_enrollment": resources.POWERSCHOOL_ENROLLMENT_RESOURCE,
         "schoolmint_grow": resources.SCHOOLMINT_GROW_RESOURCE,
+        "slack": SLACK_RESOURCE,
         "smartrecruiters": resources.SMARTRECRUITERS_RESOURCE,
         "ssh_adp_workforce_now": resources.SSH_RESOURCE_ADP_WORKFORCE_NOW,
         "ssh_clever": resources.SSH_RESOURCE_CLEVER,

src/teamster/core/resources.py

@@ -1,6 +1,7 @@
 from dagster import EnvVar
 from dagster_dbt import DbtCliResource
 from dagster_gcp import BigQueryResource, GCSResource
+from dagster_slack import SlackResource
 
 from teamster import GCS_PROJECT_NAME
 from teamster.core.io_managers.gcs import GCSIOManager
@@ -28,6 +29,8 @@
 
 OVERGRAD_RESOURCE = OvergradResource(api_key=EnvVar("OVERGRAD_API_KEY"), page_limit=100)
 
+SLACK_RESOURCE = SlackResource(token=EnvVar("SLACK_TOKEN"))
+
 SSH_COUCHDROP = SSHResource(
     remote_host=EnvVar("COUCHDROP_SFTP_HOST"),
     username=EnvVar("COUCHDROP_SFTP_USERNAME"),

src/teamster/libraries/adp/workforce_now/api/resources.py

@@ -29,7 +29,6 @@ def setup_for_execution(self, context: InitResourceContext) -> None:
         self._session.cert = (self.cert_filepath, self.key_filepath)
 
         # authorize client
-        # trunk-ignore(bandit/B106)
         token_dict = self._session.fetch_token(
             token_url="https://accounts.adp.com/auth/oauth/v2/token",
             auth=HTTPBasicAuth(username=self.client_id, password=self.client_secret),
@@ -45,12 +44,10 @@ def setup_for_execution(self, context: InitResourceContext) -> None:
     @retry(
         stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)
     )
-    def _request(self, method, url, **kwargs):
-        response = Response()
+    def _request(self, method, url, **kwargs) -> Response:
+        response = self._session.request(method=method, url=url, **kwargs)
 
         try:
-            response = self._session.request(method=method, url=url, **kwargs)
-
             response.raise_for_status()
             return response
         except HTTPError as e:

src/teamster/libraries/google/directory/resources.py

@@ -159,13 +159,7 @@ def batch_insert_users(self, users):
         def callback(request_id, response, exception):
             if exception is not None:
                 self._log.error(exception)
-                if exception.status_code == 403:
-                    raise exception
-                elif exception.status_code == 409 and exception.reason not in [
-                    "Entity already exists.",
-                    "Invalid Given/Family Name: GivenName",
-                ]:
-                    raise exception
+                raise exception
             else:
                 self._log.info(
                     msg=(
@@ -200,8 +194,7 @@ def batch_update_users(self, users):
         def callback(request_id, response, exception):
             if exception is not None:
                 self._log.error(exception)
-                if exception.status_code in [403, 409]:
-                    raise exception
+                raise exception
             else:
                 self._log.info(
                     msg=(
@@ -238,13 +231,7 @@ def batch_insert_members(self, members):
         def callback(request_id, response, exception):
             if exception is not None:
                 self._log.error(exception)
-                if exception.status_code == 403:
-                    raise exception
-                elif (
-                    exception.status_code == 409
-                    and exception.reason != "Member already exists."
-                ):
-                    raise exception
+                raise exception
 
         # Queries per minute per user == 2400 (40/sec)
         batches = self._batch_list(list=members, size=40)
@@ -272,14 +259,7 @@ def batch_insert_role_assignments(self, role_assignments, customer=None):
         def callback(request_id, response, exception):
             if exception is not None:
                 self._log.error(exception)
-                if exception.status_code == 403:
-                    raise exception
-                elif (
-                    exception.status_code == 409
-                    and exception.reason
-                    != "Role assignment already exists for the role"
-                ):
-                    raise exception
+                raise exception
 
         batches = self._batch_list(list=role_assignments, size=10)
 

src/teamster/libraries/schoolmint/grow/ops.py

@@ -1,32 +1,47 @@
 from dagster import OpExecutionContext, op
+from dagster_slack import SlackResource
 
 from teamster.libraries.schoolmint.grow.resources import SchoolMintGrowResource
 
 
 @op
 def schoolmint_grow_user_update_op(
-    context: OpExecutionContext, schoolmint_grow: SchoolMintGrowResource, users
+    context: OpExecutionContext,
+    schoolmint_grow: SchoolMintGrowResource,
+    slack: SlackResource,
+    users,
 ):
+    exceptions = []
+    slack_client = slack.get_client()
+
     for u in users:
         if u["surrogate_key_source"] == u["surrogate_key_destination"]:
             continue
 
+        request_args = ["users"]
+
         user_id = u["user_id"]
         inactive = u["inactive"]
         user_email = u["user_email"]
 
+        exception_str = [user_email]
+
         # restore
         if inactive == 0 and u["archived_at"] is not None:
             try:
                 context.log.info(f"RESTORING\t{user_email}")
+                request_args.extend([user_id, "restore"])
+                exception_str.extend([*request_args, "PUT"])
+
                 schoolmint_grow.put(
-                    "users",
-                    user_id,
-                    "restore",
-                    params={"district": schoolmint_grow.district_id},
+                    *request_args, params={"district": schoolmint_grow.district_id}
                 )
             except Exception as e:
                 context.log.exception(e)
+                exception_str.append(e)
+
+                exceptions.append("\t".join(exception_str))
+
                 continue
 
         # build user payload
@@ -49,21 +64,38 @@ def schoolmint_grow_user_update_op(
             # create
             if inactive == 0 and user_id is None:
                 context.log.info(f"CREATING\t{user_email}")
-                create_response = schoolmint_grow.post("users", json=payload)
+                exception_str.extend([*request_args, "POST"])
+
+                create_response = schoolmint_grow.post(*request_args, json=payload)
 
                 u["user_id"] = create_response["_id"]
             # update
             elif inactive == 0 and user_id is not None:
                 context.log.info(f"UPDATING\t{user_email}")
-                schoolmint_grow.put("users", user_id, json=payload)
+                request_args.append(user_id)
+                exception_str.extend([*request_args, "PUT"])
+
+                schoolmint_grow.put(*request_args, json=payload)
             # archive
             elif inactive == 1 and user_id is not None and u["archived_at"] is None:
                 context.log.info(f"ARCHIVING\t{user_email}")
-                schoolmint_grow.delete("users", user_id)
+                request_args.append(user_id)
+                exception_str.extend([*request_args, "DELETE"])
+
+                schoolmint_grow.delete(*request_args, "users", user_id)
         except Exception as e:
             context.log.exception(e)
+            exception_str.append(e)
+
+            exceptions.append("\t".join(exception_str))
+
             continue
 
+    if exceptions:
+        slack_client.chat_postMessage(
+            channel="#dagster-alerts", text="\n".join(exceptions)
+        )
+
     return users