add prefect API and flow for add task

DB_utls.py

@@ -16,7 +16,7 @@ connection_string = blinded_connection_string.replace("<db_password>", MONGODB_P
 def generate_empty_well():
     dbclient = MongoClient(connection_string)
     db = dbclient["LCM-OT-2-SLD"]  
-    collection = db["wells"] 
+    collection = db["wells-test"] 
     rows = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H']
     columns = [str(i) for i in range(1, 13)]
     for row in rows:
@@ -24,7 +24,7 @@ def generate_empty_well():
             well = f"{row}{col}"
             metadata = {
                 "well": well,
-                "status": "empty",
+                "status": "used",
                 "project": "OT2"
             }
 
@@ -40,7 +40,7 @@ def generate_empty_well():
 def update_used_wells(used_wells):
     dbclient = MongoClient(connection_string)
     db = dbclient["LCM-OT-2-SLD"]  
-    collection = db["wells"] 
+    collection = db["wells-test"] 
 
     for well in used_wells:
         metadata = {
@@ -60,7 +60,7 @@ def update_used_wells(used_wells):
 def find_unused_wells():
     dbclient = MongoClient(connection_string)
     db = dbclient["LCM-OT-2-SLD"]  
-    collection = db["wells"] 
+    collection = db["wells-test"] 
     query = {"status": "empty"}
     response = list(collection.find(query))  
     df = pd.DataFrame(response)

app.py

@@ -8,6 +8,8 @@ import secrets
 from DB_utls import find_unused_wells, update_used_wells, save_result, get_student_quota, decrement_student_quota
 import os
 from yt_utils import get_latest_video_id
+from prefect import flow
+from prefect_utils import get_prefect_client
 
 # NOTE: New global dict to store tasks keyed by (student_id, experiment_id)
 tasks_dict = {}
@@ -39,6 +41,9 @@ playlist_id = "PL8uZlc2CEpenUCAo9PeiOtbU04lFv48yD"
 
 video_id = get_latest_video_id(channel_id, playlist_id = "PL8uZlc2CEpenUCAo9PeiOtbU04lFv48yD")
 
+Prefect_client = get_prefect_client()
+print("[TEST]Prefect client initialized")
+
 
 def check_student_quota(student_id):
     """Check student's remaining experiment quota"""
@@ -268,7 +273,7 @@ def update_queue_display():
     except Exception as e:
         return f"Error getting queue status: {str(e)}"
 
-
+@flow
 def add_to_queue(student_id, R, Y, B):
     global queue_counter
     

prefect_utils.py

@@ -0,0 +1,9 @@
+from prefect import get_client
+from prefect.client.orchestration import PrefectClient
+import os
+
+async def get_prefect_client() -> PrefectClient:
+    api_key = os.getenv("PREFECT_API_KEY")
+    if not api_key:
+        raise RuntimeError("PREFECT_API_KEY is missing")
+    return await get_client()