35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
"""
|
|
订单数据访问层
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, Optional, cast
|
|
|
|
from app.core.supabase import get_supabase
|
|
|
|
|
|
def create_order(user_id: str, out_trade_no: str, amount: float) -> Dict[str, Any]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("orders").insert({
|
|
"user_id": user_id,
|
|
"out_trade_no": out_trade_no,
|
|
"amount": amount,
|
|
"status": "pending",
|
|
}).execute()
|
|
return cast(Dict[str, Any], (result.data or [{}])[0])
|
|
|
|
|
|
def get_order_by_trade_no(out_trade_no: str) -> Optional[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("orders").select("*").eq("out_trade_no", out_trade_no).single().execute()
|
|
return cast(Optional[Dict[str, Any]], result.data or None)
|
|
|
|
|
|
def update_order_status(out_trade_no: str, status: str, trade_no: str | None = None) -> None:
|
|
supabase = get_supabase()
|
|
payload: Dict[str, Any] = {"status": status}
|
|
if trade_no:
|
|
payload["trade_no"] = trade_no
|
|
if status == "paid":
|
|
payload["paid_at"] = datetime.now(timezone.utc).isoformat()
|
|
supabase.table("orders").update(payload).eq("out_trade_no", out_trade_no).execute()
|