Update batch_test.py

batch_test.py

@@ -4,46 +4,13 @@ import pendulum
 from airflow.operators.bash import BashOperator
 from airflow import DAG
 from data.gasan_data import gasanData   # import cj-gasan data
-from dateutil.relativedelta import relativedelta
 from airflow.models import Variable
+from fail_alert import fail_alert   # 실패 시 Teams 알림 발송
 
 # set timezone
 local_tz = pendulum.timezone("Asia/Seoul")
 
-def fail_alert(context):
-    context_data = """
-    {
-        "@context": "https://schema.org/extensions",
-        "@type":"MessageCard",
-        "themeColor":"0072C6",
-        "title":"Batch Job Error",
-        "summary":"test",
-        "sections": [
-            {
-                "facts":[
-                    {"name":"■ Exec Time", "value": "%s"},
-                    {"name":"■ Task", "value": "%s"},
-                    {"name":"■ DAG", "value": "%s"},
-                    {"name":"■ Reason", "value": "%s"},
-                    {"name":"■ Log URL", "value": "%s"},
-                ]
-            }
-        ]
-    }
-    """ % (
-        context.get('execution_date')+relativedelta(hours=9), 
-        context.get('task_instance').task_id, 
-        context.get('task_instance').dag_id,
-        context.get('exception'),
-        context.get('task_instance').log_url.replace("localhost:8080", Variable.get("AIRFLOW_WEB_URL"))
-    )
-
-    alert = BashOperator(
-        task_id='fail_alert',
-        bash_command="curl -d \'{data}\' -H \"Content-Type: Application/JSON\" -X POST {teams_url}".format(data=context_data, teams_url=Variable.get("TEAMS_WEBHOOKS_URL"))
-    )
-    return alert.execute(context=context)
-
+# set dag
 with DAG(
     'batch_test',
     default_args={
@@ -61,7 +28,9 @@ with DAG(
     # (Task1) 헬스체크
     health_check = BashOperator(
         task_id='health_check',
-        bash_command="curl -X GET -v {api_url}/api/v1/core/health \'-H accept: application/json\'".format(api_url=Variable.get("INF_API_URL")),
+        bash_command="curl -X GET -v {api_url}/api/v1/core/health \'-H accept: application/json\'".format(
+            api_url=Variable.get("INF_API_URL")
+        ),
     )
 
     # (Task2) gasan 작업 병렬처리
@@ -70,7 +39,14 @@ with DAG(
     for i, data in enumerate(gasanData):
         post_gasan_task = BashOperator(
             task_id='post_gasan'+str(i+1),
-            bash_command="curl -X \'POST\' \'%s/api/v1/camera/writeimage\' -H \'Content-Type: application/json\' -d \'{\"id\": \"%s\", \"%s\": \"test\", \"ip\": \"%s\", \"serialNum\": \"%s\", \"camName\": \"%s\"}\'" %(Variable.get("INF_API_URL"), Variable.get("INF_API_ID"), Variable.get("INF_API_PW"), data["ip"], data["serialNum"], data["camName"]),
+            bash_command="curl -X \'POST\' \'%s/api/v1/camera/writeimage\' -H \'Content-Type: application/json\' -d \'{\"id\": \"%s\", \"%s\": \"test\", \"ip\": \"%s\", \"serialNum\": \"%s\", \"camName\": \"%s\"}\'" %(
+                Variable.get("INF_API_URL"), 
+                Variable.get("INF_API_ID"), 
+                Variable.get("INF_API_PW"), 
+                data["ip"], 
+                data["serialNum"], 
+                data["camName"]
+            ),
         )
 
         post_gasan_tasks.append(post_gasan_task)