Spaces:
Runtime error
Runtime error
File size: 4,374 Bytes
b245237 299424a b245237 11213cb b245237 299424a b245237 11213cb b245237 11213cb b245237 11213cb b245237 299424a b245237 11213cb b245237 11213cb 299424a b245237 299424a b245237 11213cb 299424a b245237 11213cb b245237 11213cb b245237 11213cb b245237 299424a b245237 11213cb b245237 11213cb 0890b20 b245237 11213cb b245237 11213cb b245237 11213cb b245237 11213cb b245237 11213cb b245237 11213cb |
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
import datetime
import io
import qrcode
from fastapi.responses import StreamingResponse
from fastapi import APIRouter, Depends, HTTPException, Response
from app.dependencies import get_current_user
from app import db, logger
from enum import Enum
from app.graphdb.main import insert2PersonAndSetFriend
router = APIRouter(prefix="/friend_request", tags=["friend_request"])
class RequestStatus(Enum):
WAITING_INVITEE = "waiting_invitee"
WAITING_INVITER = "waiting_inviter"
COMPLETE = "complete"
COLLECTION_NAME = "friend_request"
EXPIRE_MINUTES = 15
@router.get("")
def getFriendRequest(current_user=Depends(get_current_user)):
if current_user["sub"] is None:
return HTTPException(status_code=400, detail="User not found")
friend_requests = (
db.collection(COLLECTION_NAME)
.where("inviter", "==", current_user["sub"])
.stream()
)
return {
"friend_requests": [
{**Request.to_dict(), "id": Request.id} for Request in friend_requests
]
}
@router.post("")
def createRequest(user=Depends(get_current_user)):
if user["sub"] is None:
raise HTTPException(status_code=400, detail="User not found")
_, fr_ref = db.collection(COLLECTION_NAME).add(
{
"inviter": user["sub"],
"status": RequestStatus.WAITING_INVITEE.value,
"expire_at": datetime.datetime.now(tz=datetime.timezone.utc)
+ datetime.timedelta(minutes=EXPIRE_MINUTES),
}
)
try:
qr = qrcode.make(fr_ref.id)
buf = io.BytesIO()
qr.save(buf)
buf.seek(0)
return StreamingResponse(buf, media_type="image/png")
except Exception as e:
logger.error(e)
fr_ref.delete()
return Response(content="Failed to encode image", status_code=500)
@router.patch("/{RequestId}")
async def acceptRequest(RequestId: str, user=Depends(get_current_user)):
if user["sub"] is None:
raise HTTPException(status_code=400, detail="User not found")
fr_ref = db.collection(COLLECTION_NAME).document(RequestId)
fr = fr_ref.get()
if not fr.exists:
raise HTTPException(status_code=404, detail="Friend request not found")
fr = fr.to_dict()
if isRequestExpired(fr):
raise HTTPException(status_code=400, detail="Friend request expired")
if isRequestDone(fr):
raise HTTPException(status_code=400, detail="Friend request already done")
if isInviter(user, fr):
if isInviteeEmpty(fr):
raise HTTPException(status_code=400, detail="Invitee is empty")
fr_ref.update({"status": RequestStatus.COMPLETE.value})
await makeFriend(fr["invitee"], fr["inviter"])
return {"status": "OK"}
if isInviteeEmpty(fr) and not isInviter(user, fr):
fr_ref.update(
{"invitee": user["sub"], "status": RequestStatus.WAITING_INVITER.value}
)
sendNotificationToInviter(fr["inviter"], user)
return {"status": "OK"}
def sendNotificationToInviter(inviterId: str, invitee):
return HTTPException(status_code=501, detail="Not implemented yet")
async def makeFriend(inviteeId: str, inviterId: str):
await insert2PersonAndSetFriend(inviteeId, inviterId)
@router.delete("/{RequestId}")
def deleteRequest(RequestId: str, user=Depends(get_current_user)):
if user.sub is None:
raise HTTPException(status_code=400, detail="User not found")
Request_ref = db.collection(COLLECTION_NAME).document(RequestId)
Request = Request_ref.get().to_dict()
if not Request.exists:
raise HTTPException(status_code=404, detail="Friend request not found")
if isInviter(user, Request):
Request_ref.delete()
return {"status": "OK"}
else:
raise HTTPException(status_code=400, detail="You are not inviter")
def isRequestExpired(request):
return request["expire_at"] < datetime.datetime.now(tz=datetime.timezone.utc)
def isInviter(user, Request):
return Request["inviter"] == user["sub"]
def isRequestDone(Request):
return Request["status"] == RequestStatus.COMPLETE.value
def isInviteeEmpty(Request):
return True if Request.get("invitee", None) is None else False
|