Spaces:
Configuration error
Configuration error
File size: 3,950 Bytes
6733714 | 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 | from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional, Dict
from datetime import datetime
import uuid
from app.database import get_session_collection
from app.security import get_current_user
router = APIRouter()
class SessionCreate(BaseModel):
title: Optional[str] = "New Chat"
class Session(BaseModel):
session_id: str
title: str
created_at: datetime
updated_at: datetime
messages: List[Dict] = []
@router.post("/new", response_model=Session)
def create_session(data: SessionCreate, current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
session_id = str(uuid.uuid4())
user_id = current_user["username"]
new_session = {
"session_id": session_id,
"user_id": user_id,
"title": data.title,
"created_at": datetime.now(),
"updated_at": datetime.now(),
"messages": []
}
if collection is not None:
collection.insert_one(new_session)
new_session.pop("_id")
return new_session
else:
return new_session
@router.get("/list", response_model=List[Session])
def list_sessions(current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
if collection is None:
return []
user_id = current_user["username"]
sessions_cursor = collection.find(
{"user_id": user_id},
{"_id": 0, "messages": 0}
).sort("updated_at", -1)
return list(sessions_cursor)
@router.get("/{session_id}", response_model=Session)
def get_session(session_id: str, current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
if collection is None:
return {"session_id": session_id, "title": "Offline Chat", "created_at": datetime.now(), "updated_at": datetime.now(), "messages": []}
user_id = current_user["username"]
session = collection.find_one({"session_id": session_id, "user_id": user_id}, {"_id": 0})
if not session:
raise HTTPException(status_code=404, detail="Session not found")
return session
class SessionRename(BaseModel):
title: str
@router.patch("/{session_id}/rename")
def rename_session(session_id: str, data: SessionRename, current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
if collection is None:
return {"session_id": session_id, "title": data.title}
user_id = current_user["username"]
result = collection.update_one(
{"session_id": session_id, "user_id": user_id},
{"$set": {"title": data.title, "updated_at": datetime.now()}}
)
if result.matched_count == 0:
raise HTTPException(status_code=404, detail="Session not found")
return {"session_id": session_id, "title": data.title}
@router.get("/search", response_model=List[Session])
def search_sessions(q: str, current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
if collection is None:
return []
user_id = current_user["username"]
regex = {"$regex": q, "$options": "i"}
sessions_cursor = collection.find(
{"user_id": user_id, "$or": [
{"title": regex},
{"messages.content": regex},
]},
{"_id": 0, "messages": 0}
).sort("updated_at", -1).limit(20)
return list(sessions_cursor)
@router.delete("/{session_id}")
def delete_session(session_id: str, current_user: dict = Depends(get_current_user)):
collection = get_session_collection()
if collection is None:
return {"status": "deleted (offline)", "session_id": session_id}
user_id = current_user["username"]
result = collection.delete_one({"session_id": session_id, "user_id": user_id})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="Session not found")
return {"status": "deleted", "session_id": session_id}
|