feat(dags): rewrite dags with outlet in an asset manner

AIP-75

Author: Lee-W

dags/app/discord.py

@@ -14,7 +14,8 @@
         AssetWatcher(
             name="finance_report_watcher",
             trigger=FinanceReportTrigger(
-                poke_interval=86400,  # 60*60*24
+                # poke_interval=86400,  # 60*60*24
+                poke_interval=5,  # 60*60*24
             ),
         )
     ],
@@ -24,8 +25,8 @@
 @dag(
     schedule=(
         finance_report_asset
-        | Asset.ref(name="CFP_summary")
-        | Asset.ref(name="kktix_order_report")
+        | Asset(name="proposal_count")
+        | Asset(name="registration_statistics")
     ),
     start_date=datetime(2025, 6, 28),
     catchup=False,
@@ -45,20 +46,22 @@ def discord_message_notification():
     def send_discord_message(**context: Context) -> None:
         triggering_asset_events = context["triggering_asset_events"]
         session = requests.session()
-        for asset_uri, asset_event in triggering_asset_events.items():
+        logger.info(f"Receive asset events {triggering_asset_events}")
+        for asset_uri, asset_events in triggering_asset_events.items():
             logger.info(f"Receive asset event from Asset uri={asset_uri}")
-            if asset_event.extra.get("from_trigger", False):  # type: ignore[attr-defined]
-                details = asset_event.extra["payload"]  # type: ignore[attr-defined]
-            else:
-                details = asset_event.extra  # type: ignore[attr-defined]
+            for asset_event in asset_events:  # type: ignore[attr-defined]
+                if asset_event.extra.get("from_trigger", False):
+                    details = asset_event.extra["payload"]
+                else:
+                    details = asset_event.extra
 
-            session.post(
-                details.get("webhook_url"),
-                json={
-                    "username": details.get("username"),
-                    "content": details.get("content"),
-                },
-            )
+                session.post(
+                    details.get("webhook_url"),
+                    json={
+                        "username": details.get("username"),
+                        "content": details.get("content"),
+                    },
+                )
 
     send_discord_message()
 

dags/app/proposal_reminder/dag.py

@@ -2,50 +2,34 @@
 Send Proposal Summary to Discord
 """
 
-from datetime import datetime, timedelta
-
-from airflow.sdk import Asset, Metadata, Variable, dag, task
+from airflow.sdk import Metadata, Variable, asset
 from app.proposal_reminder.udf import get_proposal_summary
 
-DEFAULT_ARGS = {
-    "owner": "Henry Lee",
-    "depends_on_past": False,
-    "start_date": datetime(2025, 2, 25),
-    "end_date": datetime(2025, 4, 9),
-    "retries": 2,
-    "retry_delay": timedelta(minutes=5),
-}
-
+# DEFAULT_ARGS = {
+#     "owner": "Henry Lee",
+#     "depends_on_past": False,
+#     "start_date": datetime(2025, 2, 25),
+#     "end_date": datetime(2025, 4, 9),
+#     "retries": 2,
+#     "retry_delay": timedelta(minutes=5),
+# }
 
-cfp_summary_asset = Asset(name="CFP_summary")
 
-
-@dag(
-    default_args=DEFAULT_ARGS,
+@asset(
+    name="proposal_count",
+    dag_id="proposal_count",
     schedule="0 16 * * *",  # At 16:00 (00:00 +8)
-    max_active_runs=1,
-    catchup=False,
 )
-def DISCORD_PROPOSAL_REMINDER_v3():
-    @task(outlets=[cfp_summary_asset])
-    def fetch_proposal_summary():
-        summary = get_proposal_summary()
-        n_talk = summary["num_proposed_talk"]
-        n_tutorial = summary["num_proposed_tutorial"]
-
-        yield Metadata(
-            cfp_summary_asset,
-            extra={
-                "webhook_url": Variable.get("DISCORD_PROGRAM_REMINDER_WEBHOOK"),
-                "username": "Program talk reminder",
-                "content": f"目前投稿議程數: {n_talk}; 課程數: {n_tutorial}",
-            },
-        )
-
-    fetch_proposal_summary()
-
-
-dag_obj = DISCORD_PROPOSAL_REMINDER_v3()
-
-if __name__ == "__main__":
-    dag_obj.test()
+def DISCORD_PROPOSAL_REMINDER_v3(self):
+    summary = get_proposal_summary()
+    n_talk = summary["num_proposed_talk"]
+    n_tutorial = summary["num_proposed_tutorial"]
+
+    yield Metadata(
+        self,
+        extra={
+            "webhook_url": Variable.get("DISCORD_PROGRAM_REMINDER_WEBHOOK"),
+            "username": "Program talk reminder",
+            "content": f"目前投稿議程數: {n_talk}; 課程數: {n_tutorial}",
+        },
+    )

dags/app/team_registration_bot/dag.py

@@ -2,51 +2,34 @@
 Send daily ordering metrics to discord channel
 """
 
-from datetime import datetime, timedelta
-
-from airflow.sdk import Asset, Metadata, Variable, dag, task
+from airflow.sdk import Metadata, Variable, asset
 from app.team_registration_bot.udf import (
     _compose_discord_msg,
     _get_statistics_from_bigquery,
 )
 
-DEFAULT_ARGS = {
-    "owner": "David Jr.",
-    "depends_on_past": False,
-    "start_date": datetime(2022, 7, 4),
-    "retries": 2,
-    "retry_delay": timedelta(minutes=5),
-    "on_failure_callback": lambda x: "Need to send notification to Discord!",
-}
-
+# DEFAULT_ARGS = {
+#     "owner": "David Jr.",
+#     "depends_on_past": False,
+#     "start_date": datetime(2022, 7, 4),
+#     "retries": 2,
+#     "retry_delay": timedelta(minutes=5),
+#     "on_failure_callback": lambda x: "Need to send notification to Discord!",
+# }
 
-kktix_order_report_asset = Asset(name="kktix_order_report")
 
-
-@dag(
-    default_args=DEFAULT_ARGS,
+@asset(
+    name="registration_statistics",
+    dag_id="registration_statistics",
     schedule="@daily",
-    max_active_runs=1,
-    catchup=False,
 )
-def KKTIX_DISCORD_BOT_FOR_TEAM_REGISTRATION():
-    @task(outlets=[kktix_order_report_asset])
-    def LOAD_TO_DISCORD():
-        statistics = _get_statistics_from_bigquery()
-        yield Metadata(
-            kktix_order_report_asset,
-            extra={
-                "webhook_url": Variable.get("discord_webhook_registration_endpoint"),
-                "username": "KKTIX order report",
-                "content": _compose_discord_msg(statistics),
-            },
-        )
-
-    LOAD_TO_DISCORD()
-
-
-dag_obj = KKTIX_DISCORD_BOT_FOR_TEAM_REGISTRATION()
-
-
-if __name__ == "__main__":
-    dag_obj.test()
+def KKTIX_DISCORD_BOT_FOR_TEAM_REGISTRATION(self):
+    statistics = _get_statistics_from_bigquery()
+    yield Metadata(
+        self,
+        extra={
+            "webhook_url": Variable.get("discord_webhook_registration_endpoint"),
+            "username": "KKTIX order report",
+            "content": _compose_discord_msg(statistics),
+        },
+    )