Merge pull request #78 from zizzic/ci-cd-workflow

[feat]: add pythonpath env for plugin check

.github/workflows/sync_dag_code.yml

@@ -36,7 +36,7 @@ jobs:
 
     build:
         needs: path-filter
-        if: needs.path-filter.outputs.dags-changed == 'true' && github.event_name == 'pull_request'
+        if: (needs.path-filter.outputs.dags-changed == 'true'|| needs.path-filter.outputs.plugin-created == 'true') && github.event_name == 'pull_request'
         runs-on: ubuntu-latest
         steps:
             - uses: actions/checkout@v4
@@ -61,8 +61,9 @@ jobs:
             - name: Install Pylint
               run: pip install pylint
             - name: Run Pylint
-              run: find dags/ tests/dags/ -name "*.py" | xargs pylint --output-format=colorized --disable=missing-docstring,invalid-name,W0104,C0301
-
+              env:
+                  PYTHONPATH: plugins
+              run: pylint --output-format=colorized --disable=missing-docstring,invalid-name,W0104,C0301,C0411 $(find dags/ tests/dags/ plugins/ -name "*.py")
             - name: Test DAG integrity
               run: pytest tests/
 

dags/game/game_comment_to_s3.py

@@ -5,7 +5,7 @@
 from airflow.models import Variable
 
 from datetime import datetime, timedelta
-import slack
+from plugins import slack
 
 import requests
 import json

dags/game/game_price_to_s3.py

@@ -59,7 +59,7 @@ def get_price():
 
                 if temp and temp[appid]["success"]:
                     break
-                print(f"The response of {appid, name} failed! {i+1} times")
+                print(f"The response of {appid, name} failed! {i + 1} times")
             else:
                 print(f"Failed to get price of {appid, name}!")
 

dags/streaming/dag_raw_data.py

@@ -21,11 +21,15 @@ def get_s_list(**kwargs):
     # RDS 연결 설정
     try:
         mysql_hook = MySqlHook(mysql_conn_id="aws_rds_conn_id")
-        result = mysql_hook.get_records(
-            "SELECT STREAMER_ID, CHZ_ID, AFC_ID FROM STREAMER_INFO;"
+        result = (
+            mysql_hook.get_records(
+                "SELECT STREAMER_ID, CHZ_ID, AFC_ID FROM STREAMER_INFO;"
+            )
+            or []
         )
+        # None 인경우 []로 처리
         chzzk, afc = [], []
-        if result:  # result가 None이 아닌 경우에만 처리
+        if result:
             for row in result:
                 if row[1]:
                     chzzk.append((row[0], row[1]))