Untested Ormar and FastAPI update

This commit is contained in:
Mawoka
2025-05-20 12:12:06 +02:00
parent 2bfd73412d
commit 03e85f8586
31 changed files with 719 additions and 694 deletions
@@ -35,7 +35,7 @@ async def join_game(data: JoinGameInput) -> JoinGameResponse:
if game_pin is None:
raise HTTPException(status_code=404, detail="Game not found")
game = await redis.get(f"game:{game_pin}")
game = PlayGame.parse_raw(game)
game = PlayGame.model_validate_json(game)
# Check if game is already running
if game.started:
raise HTTPException(status_code=400, detail="Game started already")
@@ -25,7 +25,7 @@ async def submit_answer_fn(data_answer: int, game_pin: str, player_id: str, now:
username = await redis.get(f"game:cqc:player:{player_id}")
if redis_res_game is None or username is None:
raise HTTPException(status_code=404, detail="id not existent")
game = PlayGame.parse_raw(redis_res_game)
game = PlayGame.model_validate_json(redis_res_game)
if not game.question_show:
return
@@ -96,7 +96,7 @@ async def websocket_endpoint(ws: WebSocket, game_id: str):
player_id, game_pin = game_id.split(":")
if player_id is None or game_pin is None:
await ws.send_text(WebSocketRequest(type=WebSocketTypes.Error, data="BadId").json())
await ws.send_text(WebSocketRequest(type=WebSocketTypes.Error, data="BadId").model_dump_json())
await ws.close(code=status.WS_1003_UNSUPPORTED_DATA)
username = await redis.get(f"game:cqc:player:{player_id}")
await sio.emit(
@@ -104,17 +104,19 @@ async def websocket_endpoint(ws: WebSocket, game_id: str):
{"username": username, "sid": None},
room=f"admin:{game_pin}",
)
await redis.sadd(f"game_session:{game_pin}:players", GamePlayer(username=username, sid=None).json())
await redis.sadd(f"game_session:{game_pin}:players", GamePlayer(username=username, sid=None).model_dump_json())
while True:
raw_data = await ws.receive_text()
try:
data = WebSocketRequest.parse_raw(raw_data)
data = WebSocketRequest.model_validate_json(raw_data)
except ValidationError as e:
print("ValError")
print(e)
print(raw_data)
await ws.send_text(WebSocketRequest(type=WebSocketTypes.Error, data="ValidationError").json())
await ws.send_text(
WebSocketRequest(type=WebSocketTypes.Error, data="ValidationError").model_dump_json()
)
continue
if data.type == WebSocketTypes.ButtonPress:
@@ -122,7 +124,9 @@ async def websocket_endpoint(ws: WebSocket, game_id: str):
try:
answer_index = button_to_index_map[data.data.lower()]
except (KeyError, AttributeError):
await ws.send_text(WebSocketRequest(type=WebSocketTypes.Error, data="InvalidButton").json())
await ws.send_text(
WebSocketRequest(type=WebSocketTypes.Error, data="InvalidButton").model_dump_json()
)
continue
await submit_answer_fn(answer_index, game_pin, player_id, now)
+4 -4
View File
@@ -18,7 +18,7 @@ router = APIRouter()
class SetControllerUpInput(BaseModel):
player_name: str | None
player_name: str | None = None
name: str
@@ -57,7 +57,7 @@ async def get_controller(id: uuid.UUID, user: User = Depends(get_current_user))
controller = await Controller.objects.get_or_none(id=id, user=user.id)
if controller is None:
raise HTTPException(status_code=404, detail="Controller not found")
return GetControllerResponse(**controller.dict())
return GetControllerResponse(**controller.model_dump())
class ModifyControllerInput(BaseModel):
@@ -76,7 +76,7 @@ async def modify_controller(
controller.player_name = data.player_name
controller.name = data.name
await controller.update()
return GetControllerResponse(**controller.dict())
return GetControllerResponse(**controller.model_dump())
@router.get("/list")
@@ -86,7 +86,7 @@ async def get_all_controllers(user: User = Depends(get_current_user)) -> list[Ge
return []
return_list = []
for controller in controllers:
return_list.append(GetControllerResponse(**controller.dict()))
return_list.append(GetControllerResponse(**controller.model_dump()))
return return_list
+6 -4
View File
@@ -60,7 +60,9 @@ async def init_editor(edit: bool, quiz_id: Optional[UUID] = None, user: User = D
edit_id = os.urandom(4).hex()
await redis.sadd("edit_sessions", edit_id)
await redis.set(
f"edit_session:{edit_id}", EditSessionData(quiz_id=quiz_id, edit=edit, user_id=user.id).json(), ex=3600
f"edit_session:{edit_id}",
EditSessionData(quiz_id=quiz_id, edit=edit, user_id=user.id).model_dump_json(),
ex=3600,
)
return InitEditorResponse(token=edit_id)
@@ -70,7 +72,7 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
session_data = await redis.get(f"edit_session:{edit_id}")
if session_data is None:
raise HTTPException(status_code=401, detail="Edit ID not found!")
session_data = EditSessionData.parse_raw(session_data)
session_data = EditSessionData.model_validate_json(session_data)
quiz_input.title = bleach.clean(quiz_input.title, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True)
quiz_input.description = bleach.clean(quiz_input.description, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True)
if quiz_input.background_color is not None:
@@ -122,7 +124,7 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
quiz.public = quiz_input.public
quiz.description = quiz_input.description
quiz.updated_at = datetime.now()
quiz.questions = quiz_input.dict()["questions"]
quiz.questions = quiz_input.model_dump()["questions"]
quiz.cover_image = quiz_input.cover_image
quiz.background_color = quiz_input.background_color
quiz.background_image = quiz_input.background_image
@@ -140,7 +142,7 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
return quiz
else:
quiz = Quiz(
**quiz_input.dict(),
**quiz_input.model_dump(),
user_id=session_data.user_id,
id=session_data.quiz_id,
created_at=datetime.now(),
+3 -3
View File
@@ -52,7 +52,7 @@ async def export_quiz(quiz_id: uuid.UUID, user: User = Depends(get_current_user)
if quiz.cover_image is not None:
image_urls[-1] = quiz.cover_image
quiz.cover_image = -1
quiz_dict = quiz.dict()
quiz_dict = quiz.model_dump()
del quiz_dict["user_id"], quiz_dict["id"]
quiz_dict["created_at"] = quiz_dict["created_at"].isoformat()
quiz_dict["updated_at"] = quiz_dict["updated_at"].isoformat()
@@ -129,7 +129,7 @@ async def import_quiz(file: UploadFile = File(), user: User = Depends(get_curren
question["image"] = image_urls[question["image"]]
if quiz_dict["cover_image"] is not None:
quiz_dict["cover_image"] = image_urls[-1]
quiz = Quiz.parse_obj(quiz_dict)
quiz = Quiz.model_validate(quiz_dict)
quiz.user_id = user.id
quiz.imported_from_kahoot = None
quiz.mod_rating = None
@@ -165,7 +165,7 @@ async def export_quiz_as_excel(quiz_id: uuid.UUID, user: User = Depends(get_curr
],
)
for i, question in enumerate(quiz.questions):
question = QuizQuestion.parse_obj(question)
question = QuizQuestion.model_validate(question)
data: list[Any] = [None] * 9
data[0] = i + 1
data[1] = question.question
+19 -17
View File
@@ -7,7 +7,7 @@ import json
import uuid
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, validator
from pydantic import BaseModel, field_validator
from classquiz.config import settings, redis
from classquiz.db.models import (
PlayGame,
@@ -41,7 +41,7 @@ class _ABCDQuizAnswer(ABCDQuizAnswer):
class _QuizQuestion(QuizQuestion):
answers: list[ABCDQuizAnswer] | RangeQuizAnswer | list[VotingQuizAnswer]
@validator("answers")
@field_validator("answers")
def answers_not_none_if_abcd_type(cls, v, values):
# if values["type"] == QuizQuestionType.ABCD and type(v[0]) != _ABCDQuizAnswer:
# print(type(v[0]), values)
@@ -83,7 +83,7 @@ async def get_live_game_data(
redis_res = await redis.get(f"game:{game_pin}")
if redis_res is None or user_id is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
game = _PlayGame.parse_raw(redis_res)
game = _PlayGame.model_validate_json(redis_res)
if game.user_id != user_id:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
for i, question in enumerate(game.questions):
@@ -105,18 +105,18 @@ async def get_live_game_data(
data_redis_res = await redis.get(f"game_session:{game_pin}")
if data_redis_res is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
data = GameSession.parse_raw(data_redis_res)
data = GameSession.model_validate_json(data_redis_res)
for i in range(0, len(game.questions)):
res = await redis.get(f"game_session:{game_pin}:{i}")
if res is None:
break
else:
res = json.loads(res)
ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.model_validate(i) for i in res])
data.answers.append(ga_1)
player_count = await redis.scard(f"game_session:{game_pin}:players")
total_questions = len(game.questions)
game = _GetLivePlayGame(**{**game.dict(), "total_questions": total_questions})
game = _GetLivePlayGame(**{**game.model_dump(), "total_questions": total_questions})
if in_human_count:
game.current_question += 1
@@ -160,10 +160,10 @@ async def get_game_session(game_pin: str, api_key: str | None = None, game_id: u
redis_res = await redis.get(f"game_session:{game_pin}")
if redis_res is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
data = GameSession.parse_raw(redis_res)
data = GameSession.model_validate_json(redis_res)
if user_id is None and data.game_id != str(game_id):
raise HTTPException(status_code=401, detail="Game not found or API key not found")
game = PlayGame.parse_raw(await redis.get(f"game:{game_pin}"))
game = PlayGame.model_validate_json(await redis.get(f"game:{game_pin}"))
if game.user_id != user_id and data.game_id != str(game_id):
raise HTTPException(status_code=404, detail="Game not found or API key not found")
for i in range(0, len(game.questions)):
@@ -172,12 +172,12 @@ async def get_game_session(game_pin: str, api_key: str | None = None, game_id: u
break
else:
res = json.loads(res)
ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.model_validate(i) for i in res])
data.answers.append(ga_1)
players = await redis.smembers(f"game_session:{game_pin}:players")
player_list = []
for p in players:
player_list.append(GamePlayer.parse_raw(p))
player_list.append(GamePlayer.model_validate_json(p))
return player_list
@@ -190,16 +190,18 @@ async def set_next_question(game_pin: str, question_number: int, api_key: str):
redis_res = await redis.get(f"game:{game_pin}")
if redis_res is None or user_id is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
game_data = PlayGame.parse_raw(redis_res)
game_data = PlayGame.model_validate_json(redis_res)
if game_data.user_id != user_id:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
game_data.current_question = question_number
await redis.set(f"game:{game_pin}", game_data.json(), ex=18000)
await redis.set(f"game:{game_pin}", game_data.model_dump_json(), ex=18000)
await sio.emit(
"set_question_number",
{
"question_index": question_number,
"question": ReturnQuestion(**game_data.dict(include={"questions"})["questions"][question_number]).dict(),
"question": ReturnQuestion(
**game_data.model_dump(include={"questions"})["questions"][question_number]
).model_dump(),
},
room=game_pin,
)
@@ -231,7 +233,7 @@ async def too_stupid_to_come_up_with_a_name(game_pin: str, api_key: str, in_huma
redis_res = await redis.get(f"game:{game_pin}")
if redis_res is None or user_id is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
game = PlayGame.parse_raw(redis_res)
game = PlayGame.model_validate_json(redis_res)
for i, question in enumerate(game.questions):
if question.type == QuizQuestionType.ABCD:
for o, answer in enumerate(question.answers):
@@ -244,7 +246,7 @@ async def too_stupid_to_come_up_with_a_name(game_pin: str, api_key: str, in_huma
if game.current_question >= 0:
return [
{
**game.questions[game.current_question].dict(),
**game.questions[game.current_question].model_dump(),
"current_question": game.current_question + 1 if in_human_count else game.current_question,
"total_questions": len(game.questions),
}
@@ -263,13 +265,13 @@ async def voting_results(game_pin: str, api_key: str, as_array: bool = False):
redis_res = await redis.get(f"game:{game_pin}")
if redis_res is None or user_id is None:
raise HTTPException(status_code=404, detail="Game not found or API key not found")
game = PlayGame.parse_raw(redis_res)
game = PlayGame.model_validate_json(redis_res)
if game.questions[game.current_question].type != QuizQuestionType.VOTING:
return
answer_data = await redis.get(f"game_session:{game_pin}:{game.current_question}")
if answer_data is None:
return
answer_list = AnswerDataList.parse_raw(answer_data)
answer_list = AnswerDataList.model_validate_json(answer_data)
answer_dict = {}
for answer in game.questions[game.current_question].answers:
answer_dict[answer.answer] = 0
+7 -7
View File
@@ -50,7 +50,7 @@ class LoginSession(BaseModel):
user_id: str
step_1: set[StartLoginResponseTypes]
step_2: set[StartLoginResponseTypes]
webauthn_challenge: str | None
webauthn_challenge: str | None = None
step1_success: bool = False
@@ -58,12 +58,12 @@ class StartLoginResponse(BaseModel):
step_1: set[StartLoginResponseTypes]
step_2: set[StartLoginResponseTypes]
session_id: str
webauthn_data: None | str
webauthn_data: None | str = None
def verify_webauthn(data, fidocredentialss: list[FidoCredentials], login_session: LoginSession):
try:
credential = AuthenticationCredential.parse_obj(data)
credential = AuthenticationCredential.model_validate(data)
except ValidationError:
print("ValidationError")
raise HTTPException(401)
@@ -141,7 +141,7 @@ async def start_login(data: StartLoginInput):
webauthn_challenge=webauthn_challenge,
)
session_id = os.urandom(16).hex()
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
await redis.set(f"login_session:{session_id}", login_session.model_dump_json(), ex=600)
return StartLoginResponse(step_1=step_1, step_2=step_2, session_id=session_id, webauthn_data=webauthn_data)
@@ -157,7 +157,7 @@ async def step_1_endpoint(session_id: str, data: StepInput, request: Request, re
redis_res = await redis.get(f"login_session:{session_id}")
if redis_res is None:
raise HTTPException(401, detail="wrong credentials")
login_session = LoginSession.parse_raw(redis_res)
login_session = LoginSession.model_validate_json(redis_res)
if step_id == 1:
if data.auth_type not in {*login_session.step_1, StartLoginResponseTypes.BACKUP}:
@@ -177,7 +177,7 @@ async def step_1_endpoint(session_id: str, data: StepInput, request: Request, re
return await log_user_in(user, request, response)
else:
login_session.step1_success = True
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
await redis.set(f"login_session:{session_id}", login_session.model_dump_json(), ex=600)
return Response(status_code=202)
else:
print("Wrong Password")
@@ -189,7 +189,7 @@ async def step_1_endpoint(session_id: str, data: StepInput, request: Request, re
return await log_user_in(user, request, response)
else:
login_session.step1_success = True
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
await redis.set(f"login_session:{session_id}", login_session.model_dump_json(), ex=600)
return Response(status_code=202)
else:
raise HTTPException(401, detail="webauthn failed")
+1 -1
View File
@@ -47,7 +47,7 @@ async def get_newest_quizzes(
class SetModRatingForQuizInput(BaseModel):
rating: int | None
rating: int | None = None
@router.post("/rating/set/{quiz_id}")
+9 -9
View File
@@ -71,7 +71,7 @@ async def get_public_quiz(quiz_id: uuid.UUID):
else:
quiz.views += 1
await quiz.update()
return PublicQuizResponse(**quiz.dict())
return PublicQuizResponse(**quiz.model_dump())
@router.post("/start/{quiz_id}")
@@ -130,21 +130,21 @@ async def start_quiz(
if cqcs_enabled:
code = generate_code(6)
await redis.set(f"game:cqc:code:{code}", game_pin, ex=3600)
await redis.set(f"game:{str(game.game_pin)}", game.json(), ex=18000)
await redis.set(f"game:{str(game.game_pin)}", game.model_dump_json(), ex=18000)
await redis.set(f"game_pin:{user.id}:{quiz_id}", game_pin, ex=18000)
await redis.set(
f"game_in_lobby:{user.id.hex}",
GameInLobby(game_id=game.game_id, game_pin=str(game_pin), quiz_title=quiz.title).json(),
GameInLobby(game_id=game.game_id, game_pin=str(game_pin), quiz_title=quiz.title).model_dump_json(),
ex=900,
)
return {**quiz.dict(exclude={"id"}), **game.dict(exclude={"questions"}), "cqc_code": code}
return {**quiz.model_dump(exclude={"id"}), **game.model_dump(exclude={"questions"}), "cqc_code": code}
class CheckIfCaptchaEnabledResponse(BaseModel):
enabled: bool
game_mode: str | None
custom_field: str | None
game_mode: str | None = None
custom_field: str | None = None
@router.get("/play/check_captcha/{game_pin}", response_model=CheckIfCaptchaEnabledResponse)
@@ -152,7 +152,7 @@ async def check_if_captcha_enabled(game_pin: str):
game = await redis.get(f"game:{game_pin}")
if game is None:
return JSONResponse(status_code=404, content={"detail": "game not found"})
game = PlayGame.parse_raw(game)
game = PlayGame.model_validate_json(game)
if game.captcha_enabled:
return CheckIfCaptchaEnabledResponse(enabled=True, game_mode=game.game_mode, custom_field=game.custom_field)
else:
@@ -228,7 +228,7 @@ async def export_quiz_answers(export_token: str, game_pin: str):
raise HTTPException(status_code=404, detail="export token not found")
data = json.loads(data)
data2 = await redis.get(f"game:{game_pin}")
game_data = PlayGame.parse_raw(data2)
game_data = PlayGame.model_validate_json(data2)
quiz = await Quiz.objects.get_or_none(id=game_data.quiz_id)
if quiz is None:
raise HTTPException(status_code=404, detail="quiz not found")
@@ -255,4 +255,4 @@ async def export_quiz_answers(export_token: str, game_pin: str):
@router.post("/excel-import")
async def import_from_excel(file: UploadFile = File(), user: User = Depends(get_current_user)) -> Quiz:
quiz = await handle_import_from_excel(file.file, user)
return Quiz.parse_obj(quiz.dict(exclude={"user_id": ...}))
return Quiz.model_validate(quiz.model_dump(exclude={"user_id": ...}))
+4 -2
View File
@@ -19,7 +19,9 @@ router.include_router(shares_router, prefix="/shares")
@router.post("/create", response_model_exclude={"user": ...})
async def create_quiztivity(data: QuizTivityInput, user: User = Depends(get_current_user)) -> QuizTivity:
quiztivity = QuizTivity.parse_obj({**data.dict(), "user": user, "id": uuid4(), "created_at": datetime.now()})
quiztivity = QuizTivity.model_validate(
{**data.model_dump(), "user": user, "id": uuid4(), "created_at": datetime.now()}
)
return await quiztivity.save()
@@ -36,7 +38,7 @@ async def put_quiztivity(data: QuizTivityInput, uuid: UUID, user: User = Depends
quiztivity = await QuizTivity.objects.get_or_none(id=uuid, user=user)
if quiztivity is None:
raise HTTPException(status_code=404, detail="QuizTivity not found")
quiztivity.pages = data.dict()["pages"]
quiztivity.pages = data.model_dump()["pages"]
quiztivity.title = data.title
return await quiztivity.update()
+4 -4
View File
@@ -25,9 +25,9 @@ async def get_shares(user: User = Depends(get_current_user)) -> list[PublicQuizT
class CreateShareInput(BaseModel):
name: str | None
name: str | None = None
quiztivity: UUID
expire_in: int | None
expire_in: int | None = None
@router.post("/")
@@ -55,8 +55,8 @@ async def delete_share(uuid: UUID, user: User = Depends(get_current_user)):
class UpdateShareInput(BaseModel):
name: str | None
expire_in: int | None
name: str | None = None
expire_in: int | None = None
@router.put("/{uuid}")
+1 -1
View File
@@ -17,5 +17,5 @@ async def get_game_in_lobby(user: User = Depends(get_current_user)):
game_in_lobby_raw = await redis.get(f"game_in_lobby:{user.id.hex}")
if game_in_lobby_raw is None:
raise HTTPException(status_code=404, detail="No game waiting")
game_in_lobby = GameInLobby.parse_raw(game_in_lobby_raw)
game_in_lobby = GameInLobby.model_validate_json(game_in_lobby_raw)
return game_in_lobby
+1 -1
View File
@@ -45,7 +45,7 @@ class SitemapQuiz(BaseModel):
updated_at: datetime.datetime
class Config:
orm_mode = True
from_attributes = True
@router.get("/get")
+4 -4
View File
@@ -66,7 +66,7 @@ router.include_router(oauth.router, tags=["users", "oauth"], prefix="/oauth")
async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User | JSONResponse:
if settings.registration_disabled:
raise HTTPException(status_code=423)
user = User(**user.dict(), id=uuid.uuid4(), avatar=gzipped_user_avatar(), created_at=datetime.now())
user = User(**user.model_dump(), id=uuid.uuid4(), avatar=gzipped_user_avatar(), created_at=datetime.now())
try:
validate_email(user.email)
except EmailNotValidError as e:
@@ -232,7 +232,7 @@ async def reset_password_with_token(reset_password: ResetPassword, response: Res
@router.get("/sessions/list", response_model=list[UserSession], response_model_exclude={"user", "session_key", "quizs"})
async def list_sessions(user: User = Depends(get_current_user)):
sessions = await UserSession.objects.filter(user=user).all()
return [session.dict() for session in sessions]
return [session.model_dump() for session in sessions]
@router.delete("/sessions/{session_id}")
@@ -297,7 +297,7 @@ async def get_other_avatar(respo: Response, user_id: uuid.UUID):
class InternalAuthData(BaseModel):
rememberme: str
jwt: str | None
jwt: str | None = None
@router.post("/auth/internal")
@@ -341,7 +341,7 @@ async def get_email_from_jwt(data: GetEmailFromJWT):
async def generate_api_key(user: User = Depends(get_current_user)):
key = ApiKey(key=os.urandom(24).hex(), user=user)
await key.save()
return key.dict(include={"key"})
return key.model_dump(include={"key"})
@router.get("/api_keys", response_model=list[ApiKey], response_model_include={"key"})
+1 -1
View File
@@ -59,7 +59,7 @@ class IpResponse(BaseModel):
@router.get("/ip-lookup/{ip}", response_model=IpResponse)
async def get_ip_data(ip: str, _: User = Depends(get_current_user)):
async with ClientSession() as session, session.get(f"http://ip-api.com/json/{ip}") as response:
data = await response.json()
data = await response.model_dump_json()
try:
return IpResponse(**data)
except ValidationError: