update flow

.github/workflows/daily.yml

@@ -43,4 +43,3 @@ jobs:
         DELTA: ${{ github.event.inputs.delta}}
       run: |
         python daily.py
-        python glue.py

daily.py

@@ -13,12 +13,12 @@ from urllib.parse import urlparse
 from prefect import flow, task
 
 from lxml import etree
-
+from glue import glue_job_run
 from utils import (crawl, datemodifier, encode, encode_content,
                    extract_from_pdf, extract_reference, fetch_url,
                    sentiment_computation, translate, update_content)
 
-
+@task(name = "crawl_eastmoney")
 def crawl_eastmoney(url, article):
     """
     Crawls the given URL and extracts information from the webpage.
@@ -66,8 +66,8 @@ def crawl_eastmoney(url, article):
     extract_reference(article)
     update_content(article)
 
-@flow(name = "Data Collection China - Daily", log_prints = True)
-def main():
+@task(name = "data collection")
+def daily():
     with open('xpath.json', 'r', encoding='UTF-8') as f:
         xpath_dict = json.load(f)
 
@@ -525,5 +525,10 @@ def main():
                         except Exception as error:
                             print(error)
 
+@flow(name = "Data Collection China - Daily", log_prints=True)
+def main():
+    daily()
+    glue_job_run()
+
 if __name__ == '__main__':
     main()

glue.py

@@ -1,6 +1,7 @@
 """Trigger Parquet Snapshot Glue job"""
 import os
 import boto3
+from prefect import task
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -23,12 +24,22 @@ def get_client_connection():
         aws_secret_access_key=AWS_SECRET_ACCESS_KEY
     )
 
-glue = get_client_connection()
-response = glue.start_job_run(
-    JobName='Article Snapshot China'
-)
-print(response)
-response = glue.start_job_run(
-    JobName='Reference China'
-)
-print(response)
+@task(name = "glue_job_run")
+def glue_job_run():
+    """
+    Triggers the Glue job run for the Parquet Snapshot.
+
+    :return: None
+    """
+    glue = get_client_connection()
+    response = glue.start_job_run(
+        JobName='Article Snapshot China'
+    )
+    print(response)
+    response = glue.start_job_run(
+        JobName='Reference China'
+    )
+    print(response)
+
+if __name__ == "__main__":
+    glue_job_run()