Spaces:
Sleeping
Sleeping
File size: 1,925 Bytes
25a9f11 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | import os
from supabase import create_client, Client
SUPABASE_URL = os.environ.get("SUPABASE_URL")
SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
if not SUPABASE_URL or not SUPABASE_KEY:
raise RuntimeError("Set SUPABASE_URL and SUPABASE_KEY in environment (HF secrets).")
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
# helper functions (examples)
def get_pending_tasks(role: str):
# tasks table columns: id, type, agent_role, assigned_to, bed_id, patient_id, status, options
resp = supabase.table("tasks").select("*").eq("status", "pending").or_(
f"agent_role.eq.{role},target_role.eq.HUMAN"
).execute()
return resp.data
def lock_task(task_id: int):
return supabase.table("tasks").update({"status": "locked"}).eq("id", task_id).execute()
def unlock_task(task_id: int):
return supabase.table("tasks").update({"status": "pending"}).eq("id", task_id).execute()
def submit_task_decision(task_id: int, decision: dict):
# store decision in a decisions table OR update task result
supabase.table("task_decisions").insert([{
"task_id": task_id,
"decision": decision
}]).execute()
# mark task as completed/assigned as appropriate (backend should validate)
return supabase.table("tasks").update({"status": "completed"}).eq("id", task_id).execute()
def fetch_context_for_task(task):
# example: get beds, staff, patient info needed for decision
data = {}
# get available beds
beds = supabase.table("beds").select("*").in_("status", ["available"]).execute().data
data["available_beds"] = beds or []
# staff list
staff = supabase.table("staff").select("*").execute().data
data["staff"] = staff or []
# patient if any
if task.get("patient_id"):
patient = supabase.table("patients").select("*").eq("id", task["patient_id"]).single().execute().data
data["patient"] = patient
return data
|