File size: 1,626 Bytes
21999ba |
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 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import os
import datetime as DT
from typing import TypedDict, List
from supabase import create_client, Client
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
load_dotenv()
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_KEY")
supabase: Client = create_client(url, key)
Activity = TypedDict("Activity", {
"id": str,
"email": str,
"chat_history": list,
"messages": list,
"buttons": list,
"created_at": str,
"updated_at": str,
})
def getUserActivities(emailId: str, id=None) -> List[Activity]:
filterQuery = supabase.table("user_activities") \
.select("*") \
.eq("email", emailId)
if id:
filterQuery = filterQuery.eq("id", id)
response = filterQuery \
.order("updated_at", desc=True) \
.execute()
activities = response.data
return activities or []
def createUserActivity(emailId: str, chatHistory: list, messages: list, buttons: list) -> str:
data, count = supabase.table('user_activities').insert({
"email": emailId,
"chat_history": chatHistory,
"messages": messages,
"buttons": buttons,
}).execute()
if data:
print(f"{data=}")
activityId = data[1][0]["id"]
return activityId
def updateUserActivityById(id: str, chatHistory: list, messages: list, buttons: list):
data, count = supabase.table('user_activities').update({
"chat_history": chatHistory,
"messages": messages,
"buttons": buttons,
"updated_at": DT.datetime.now(ZoneInfo("UTC")).isoformat()
}).eq("id", id).execute()
|