From 02c917a28fa54bbd8173694b728124c326f2e116 Mon Sep 17 00:00:00 2001 From: Mawoka Date: Sat, 1 Nov 2025 15:57:10 +0100 Subject: [PATCH] Further cleanup --- classquiz/helpers/__init__.py | 100 +++++++++---------- classquiz/kahoot_importer/README.md | 6 +- classquiz/kahoot_importer/__init__.py | 45 +++++---- classquiz/kahoot_importer/get.py | 6 +- classquiz/kahoot_importer/import_quiz.py | 4 +- classquiz/kahoot_importer/search.py | 6 +- classquiz/oauth/__init__.py | 6 +- classquiz/oauth/authenticate_user.py | 17 ++-- classquiz/oauth/custom.py | 56 ++++------- classquiz/oauth/github.py | 116 ++++++++++------------- classquiz/oauth/google.py | 53 ++++------- classquiz/routers/admin.py | 6 +- classquiz/routers/avatar.py | 13 --- classquiz/routers/community.py | 22 ++++- classquiz/routers/eximport.py | 4 +- classquiz/routers/users/__init__.py | 91 ++++-------------- 16 files changed, 215 insertions(+), 336 deletions(-) diff --git a/classquiz/helpers/__init__.py b/classquiz/helpers/__init__.py index 0f20b85..9c95dbc 100644 --- a/classquiz/helpers/__init__.py +++ b/classquiz/helpers/__init__.py @@ -6,14 +6,13 @@ import asyncio import uuid from datetime import datetime -from typing import Optional, BinaryIO +from typing import BinaryIO, Any from fastapi import HTTPException from openpyxl import load_workbook -import ormar.exceptions -from classquiz.db.models import Quiz, User, InstanceData, QuizQuestion, ABCDQuizAnswer +from classquiz.db.models import Quiz, User, QuizQuestion, ABCDQuizAnswer import xlsxwriter from aiohttp import ClientSession from io import BytesIO @@ -24,7 +23,7 @@ from classquiz.helpers.hashcash import check as hc_check settings = settings() -async def get_meili_data(quiz: Quiz) -> dict: +async def get_meili_data(quiz: Quiz) -> dict[str, Any]: return { "id": str(quiz.id), "title": quiz.title, @@ -35,14 +34,19 @@ async def get_meili_data(quiz: Quiz) -> dict: } -async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: dict, player_scores: dict) -> BytesIO: +async def generate_spreadsheet( + quiz_results: dict[str, Any], + quiz: Quiz, + player_fields: dict[str, Any], + player_scores: dict[str, Any], +) -> BytesIO: storage = BytesIO() workbook = xlsxwriter.Workbook(storage, {"in_memory": True}) player_worksheet = workbook.add_worksheet() player_worksheet.name = "Players" - player_worksheet.write(0, 0, "Username") - player_worksheet.write(0, 1, "Score") - player_worksheet.write(0, 2, "Custom-Field") + _ = player_worksheet.write(0, 0, "Username") + _ = player_worksheet.write(0, 1, "Score") + _ = player_worksheet.write(0, 2, "Custom-Field") for i, player in enumerate(player_scores.keys()): player_worksheet.write(i + 1, 0, player) player_worksheet.write(i + 1, 1, player_scores[player]) @@ -53,33 +57,33 @@ async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: di worksheet = workbook.add_worksheet() worksheet.name = "Questions" - worksheet.write(0, 0, "Question") - worksheet.write(0, 1, "Time") - worksheet.write(0, 2, "Image") - worksheet.write(0, 3, "Correct answers") - worksheet.write(0, 4, "Correct answers") - worksheet.write(0, 5, "Wrong answers") + _ = worksheet.write(0, 0, "Question") + _ = worksheet.write(0, 1, "Time") + _ = worksheet.write(0, 2, "Image") + _ = worksheet.write(0, 3, "Correct answers") + _ = worksheet.write(0, 4, "Correct answers") + _ = worksheet.write(0, 5, "Wrong answers") for i, _ in enumerate(quiz_results): question = quiz.questions[i] - # print(quiz_results) try: answer_data = quiz_results[str(i)] except KeyError: continue - worksheet.write(i + 1, 0, question["question"]) - worksheet.write(i + 1, 1, question["time"]) + _ = worksheet.write(i + 1, 0, question["question"]) + _ = worksheet.write(i + 1, 1, question["time"]) try: async with ( ClientSession() as session, session.get(f"{settings.root_address}/api/v1/storage/download/{question['image']}") as response, ): - if "image" in response.headers.get("Content-Type"): + content_type = response.headers.get("Content-Type") + if content_type is not None and "image" in content_type: img_data = BytesIO(await response.read()) - worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data}) + _ = worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data}) image = Image.open(img_data) - worksheet.set_row(i + 1, image.height) - worksheet.set_column(2, 2, image.width) + _ = worksheet.set_row(i + 1, image.height) + _ = worksheet.set_column(2, 2, image.width) except TypeError: pass answer_amount = len(answer_data) @@ -91,24 +95,24 @@ async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: di correct_answers += 1 else: wrong_answers += 1 - worksheet.write(i + 1, 3, f"{round(correct_answers / answer_amount * 100)}%") - worksheet.write(i + 1, 4, correct_answers) - worksheet.write(i + 1, 5, wrong_answers) + _ = worksheet.write(i + 1, 3, f"{round(correct_answers / answer_amount * 100)}%") + _ = worksheet.write(i + 1, 4, correct_answers) + _ = worksheet.write(i + 1, 5, wrong_answers) ws = workbook.add_worksheet(f"{i + 1}. Question") - ws.write(0, 0, "Answer") - ws.write(0, 1, "Correct") - ws.write(0, 2, "Username") + _ = ws.write(0, 0, "Answer") + _ = ws.write(0, 1, "Correct") + _ = ws.write(0, 2, "Username") for j, _ in enumerate(answer_data): - ws.write(j + 1, 0, answer_data[j]["answer"]) + _ = ws.write(j + 1, 0, answer_data[j]["answer"]) if answer_data[j]["right"]: - ws.write(j + 1, 1, "True") + _ = ws.write(j + 1, 1, "True") else: - ws.write(j + 1, 1, "False") - ws.write(j + 1, 2, answer_data[j]["username"]) + _ = ws.write(j + 1, 1, "False") + _ = ws.write(j + 1, 2, answer_data[j]["username"]) workbook.close() - storage.seek(0) + _ = storage.seek(0) return storage @@ -118,9 +122,11 @@ async def handle_import_from_excel(data: BinaryIO, user: User) -> Quiz: except KeyError: raise HTTPException(status_code=400, detail="File not in excel format") ws = wb.active + if ws is None: + raise Exception("Workbook is None") questions: list[dict] = [] - title = ws["C5"].value - description = ws["C6"].value + title: str | None = ws["C5"].value + description: str | None = ws["C6"].value if title is None: raise HTTPException(status_code=400, detail="Title missing") if description is None: @@ -182,12 +188,12 @@ async def handle_import_from_excel(data: BinaryIO, user: User) -> Quiz: questions=questions, imported_from_kahoot=False, ) - await quiz.save() + _ = await quiz.save() else: existing_quiz.questions = [*existing_quiz.questions, *questions] existing_quiz.updated_at = datetime.now() existing_quiz.mod_rating = None - await existing_quiz.update() + _ = await existing_quiz.update() quiz = existing_quiz return quiz @@ -227,27 +233,7 @@ async def meilisearch_init(): LOGGER.info("Finished MeiliSearch synchronisation") -async def telemetry_ping(): - try: - instance_data = await InstanceData.objects.first() - except ormar.exceptions.NoMatch: - instance_data = InstanceData() - await instance_data.save() - async with ( - ClientSession() as session, - session.post( - f"https://cit.mawoka.eu.org/public/{instance_data.instance_id}", - json={ - "public_quizzes": await Quiz.objects.filter(public=True).count(), - "private_quizzes": await Quiz.objects.filter(public=False).count(), - "users": await User.objects.count(), - }, - ), - ): - return - - -def check_hashcash(data: str, input_data: str, claim_in: Optional[str] = "19") -> bool: +def check_hashcash(data: str, input_data: str, claim_in: str | None = "19") -> bool: """ It checks that the hashcash is valid, and if it is, it returns True diff --git a/classquiz/kahoot_importer/README.md b/classquiz/kahoot_importer/README.md index 917c947..6e26764 100644 --- a/classquiz/kahoot_importer/README.md +++ b/classquiz/kahoot_importer/README.md @@ -15,8 +15,8 @@ This library is also tested. ```python from classquiz.kahoot_importer.get import get, _Response from asyncio import run -# _Response ia a pydantic-object, so you have access to -# .dict() or .json(exclude={"kahoot"}) +# _Response is a pydantic-object, so you have access to +# .dict() or .model_dump_json(exclude={"kahoot"}) async def main(): kahoot_quiz: _Response = await get("GAME_ID") @@ -31,7 +31,7 @@ run(main()) from classquiz.kahoot_importer.search import search, _Response from asyncio import run # _Response ia a pydantic-object, so you have access to -# .dict() or .json(exclude={"kahoot"}) +# .dict() or .model_dump_json(exclude={"kahoot"}) async def main(): kahoot_quizzes: _Response = await search("QUERY") diff --git a/classquiz/kahoot_importer/__init__.py b/classquiz/kahoot_importer/__init__.py index 5285068..b5d4205 100644 --- a/classquiz/kahoot_importer/__init__.py +++ b/classquiz/kahoot_importer/__init__.py @@ -3,7 +3,7 @@ # SPDX-License-Identifier: MPL-2.0 -from typing import List, Any, Optional +from typing import Any from uuid import UUID from pydantic import BaseModel @@ -33,10 +33,10 @@ class _LastEdit(BaseModel): class _ImageMetadata(BaseModel): id: UUID | None = None - content_type: Optional[str] = None - width: Optional[int] = None - height: Optional[int] = None - resources: Optional[str] = None + content_type: str | None = None + width: int | None = None + height: int | None = None + resources: str | None = None class _SampleQuestion(BaseModel): @@ -48,11 +48,11 @@ class _SampleQuestion(BaseModel): class _Access(BaseModel): - groupRead: List[Any | None] - folderGroupIds: List[Any | None] + groupRead: list[Any | None] + folderGroupIds: list[Any | None] -class _Card(BaseModel): +class Card(BaseModel): type: str title: str description: str @@ -60,12 +60,12 @@ class _Card(BaseModel): cover: str | None = None coverMetadata: _CoverMetadata | dict[None, None] | None = None draftExists: bool - inventoryItemIds: List[Any] = None + inventoryItemIds: list[Any] | None = None number_of_questions: int creator: UUID creator_username: str creator_avatar: _CreatorAvatar | dict[None, None] | None = None - badges: List[str] + badges: list[str] visibility: int locked: bool writeProtection: bool @@ -76,11 +76,11 @@ class _Card(BaseModel): draft: bool combined: bool compatibility_level: int - sample_questions: List[_SampleQuestion] + sample_questions: list[_SampleQuestion] number_of_plays: int number_of_players: int total_favourites: int - question_types: List[str] + question_types: list[str] created: int modified: int access: _Access @@ -89,7 +89,7 @@ class _Card(BaseModel): class _Entity(BaseModel): - card: _Card + card: Card class _Origin(BaseModel): @@ -130,8 +130,8 @@ class _Video(BaseModel): startTime: float endTime: float service: str - full_url: Optional[str] = None - id: Optional[str] = None + full_url: str | None = None + id: str | None = None class _Question(BaseModel): @@ -140,17 +140,17 @@ class _Question(BaseModel): time: int points: bool pointsMultiplier: int - choices: List[_Choice] + choices: list[_Choice] image: str | None = None imageMetadata: _ImageMetadata | None = None - resources: Optional[str] = None + resources: str | None = None video: _Video questionFormat: int languageInfo: _LanguageInfo | None = None - media: List[Any] + media: list[Any] -class _Kahoot(BaseModel): +class Kahoot(BaseModel): uuid: UUID language: str creator: UUID @@ -161,20 +161,19 @@ class _Kahoot(BaseModel): visibility: int difficulty: int | None = None audience: str - audience: str title: str description: str quizType: str - tags: str | None | List[str] = None + tags: str | None | list[str] = None cover: str | None = None coverMetadata: _CoverMetadata | dict[None, None] | None = None - questions: List[_Question] + questions: list[_Question] metadata: _Metadata parent: _Parent | None = None resources: str | None = None slug: str languageInfo: _LanguageInfo | None = None - inventoryItemIds: List[Any] + inventoryItemIds: list[Any] type: str created: int modified: int diff --git a/classquiz/kahoot_importer/get.py b/classquiz/kahoot_importer/get.py index 96a1819..d96f62e 100644 --- a/classquiz/kahoot_importer/get.py +++ b/classquiz/kahoot_importer/get.py @@ -6,12 +6,12 @@ from aiohttp import ClientSession from pydantic import BaseModel -from classquiz.kahoot_importer import _Card, _Kahoot +from classquiz.kahoot_importer import Card, Kahoot class _Response(BaseModel): - card: _Card - kahoot: _Kahoot + card: Card + kahoot: Kahoot async def get(game_id: str) -> _Response | int: diff --git a/classquiz/kahoot_importer/import_quiz.py b/classquiz/kahoot_importer/import_quiz.py index f093da4..3b2e4a9 100644 --- a/classquiz/kahoot_importer/import_quiz.py +++ b/classquiz/kahoot_importer/import_quiz.py @@ -58,7 +58,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | int: if type(quiz) is int: return quiz quiz_questions: list[dict] = [] - quiz_id = uuid.uuid4() + new_quiz_id = uuid.uuid4() meilisearch.delete_index(settings.meilisearch_index) meilisearch.create_index(settings.meilisearch_index) uploaded_images: list[StorageItem] = [] @@ -97,7 +97,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | int: if quiz.kahoot.description is None or quiz.kahoot.description == "": quiz.kahoot.description = "Description Missing!" quiz_data = Quiz( - id=quiz_id, + id=new_quiz_id, public=False, title=bleach.clean(quiz.kahoot.title, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True), description=bleach.clean(quiz.kahoot.description, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True), diff --git a/classquiz/kahoot_importer/search.py b/classquiz/kahoot_importer/search.py index 989438b..59c6f53 100644 --- a/classquiz/kahoot_importer/search.py +++ b/classquiz/kahoot_importer/search.py @@ -3,8 +3,6 @@ # SPDX-License-Identifier: MPL-2.0 -from typing import List - from aiohttp import ClientSession from pydantic import BaseModel @@ -13,7 +11,7 @@ from classquiz.kahoot_importer import _Entity # noqa : E501 class _Response(BaseModel): - entities: List[_Entity] + entities: list[_Entity] totalHits: int cursor: int | None = None pageTimestamp: int @@ -28,7 +26,7 @@ async def search( ) -> _Response: """ - :param inventory_item_id: I dkon't know + :param inventory_item_id: I don't know :param search_cluster: Doesn't seeem to matter :param cursor: The position in the result-list (page) :param query: The search query diff --git a/classquiz/oauth/__init__.py b/classquiz/oauth/__init__.py index 4ee583b..b2c9d5e 100644 --- a/classquiz/oauth/__init__.py +++ b/classquiz/oauth/__init__.py @@ -22,7 +22,7 @@ router.include_router(github.router, prefix="/github") router.include_router(custom.router, prefix="/custom") -async def rememberme_middleware(request: Request, call_next): +async def rememberme_middleware(request: Request, call_next) -> Response: rememberme_cookie = request.cookies.get("rememberme_token") bearer_token = request.cookies.get("access_token") conditions_to_handle_met = True @@ -41,13 +41,13 @@ async def rememberme_middleware(request: Request, call_next): # Verifying the bearer try: - jwt.decode( + _ = jwt.decode( param, settings.secret_key, algorithms=["HS256"] ) # checking if the token is valid, throws error if not conditions_to_handle_met = False except JWTError: try: - jws.verify( + _ = jws.verify( param, settings.secret_key, algorithms=["HS256"] ) # Verifying only the signature of the jwt, throws error if signature is invalid except JWSError: diff --git a/classquiz/oauth/authenticate_user.py b/classquiz/oauth/authenticate_user.py index 8c4c45c..bfa8efc 100644 --- a/classquiz/oauth/authenticate_user.py +++ b/classquiz/oauth/authenticate_user.py @@ -14,18 +14,19 @@ from classquiz.auth import create_access_token settings = settings() -async def log_user_in(user: User, request: Request, response: Response): +async def log_user_in(user: User | None, request: Request, response: Response): if user is None: raise HTTPException(status_code=401, detail="User not matched!") remote_ip = None - if request.headers.get("X-Forwarded-For") is None: + forwarded_for_header = request.headers.get("X-Forwarded-For") + if forwarded_for_header is None: remote_ip = request.client.host else: - if "," in request.headers.get("X-Forwarded-For"): - remote_ip = request.headers.get("X-Forwarded-For").split(", ")[0] + if "," in forwarded_for_header: + remote_ip = forwarded_for_header.split(", ")[0] else: - remote_ip = request.headers.get("X-Forwarded-For") + remote_ip = forwarded_for_header session_key = os.urandom(32).hex() user_session = UserSession( user=user, @@ -47,7 +48,11 @@ async def log_user_in(user: User, request: Request, response: Response): max_age=60 * 60 * 24 * 365, ) response.set_cookie( - key="rememberme_token", value=session_key, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 365 + key="rememberme_token", + value=session_key, + httponly=True, + samesite="lax", + max_age=60 * 60 * 24 * 365, ) return {"access_token": access_token, "token_type": "bearer"} diff --git a/classquiz/oauth/custom.py b/classquiz/oauth/custom.py index 95a8ecd..b9c0520 100644 --- a/classquiz/oauth/custom.py +++ b/classquiz/oauth/custom.py @@ -77,43 +77,27 @@ async def auth(request: Request, response: Response): except (TypeError, ValidationError): raise HTTPException(status_code=401, detail="Something went wrong.") - user_in_db = await User.objects.get_or_none(email=user_data.email) - if user_in_db is None: - # REGISTER USER - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=user_data.preferred_username, - verified=user_data.email_verified, - auth_type=UserAuthTypes.CUSTOM, - google_uid=user_data.sub.hex, - avatar=gzipped_user_avatar(), - ) - # skipcq: PYL-W0703 - except Exception as e: - if type(e) is asyncpg.exceptions.UniqueViolationError: - error = True - counter = 1 - while error: - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=f"{user_data.preferred_username}{counter}", - verified=user_data.email_verified, - auth_type=UserAuthTypes.CUSTOM, - google_uid=user_data.sub.hex, - avatar=gzipped_user_avatar(), - ) - error = False - except asyncpg.exceptions.UniqueViolationError: - counter += 1 - error = True - else: - raise HTTPException(status_code=500, detail=str(e)) + try: + await User.objects.create( + id=uuid.uuid4(), + email=user_data.email, + username=user_data.preferred_username, + verified=user_data.email_verified, + auth_type=UserAuthTypes.CUSTOM, + google_uid=user_data.sub.hex, + avatar=gzipped_user_avatar(), + ) + # skipcq: PYL-W0703 + except asyncpg.exceptions.UniqueViolationError: + # Most likely a duplicate email/username, not UUID. + raise HTTPException(status_code=400, detail="User already exists.") + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) user = await User.objects.get_or_none( - email=user_data.email, google_uid=user_data.sub.hex, auth_type=UserAuthTypes.CUSTOM, verified=True + email=user_data.email, + google_uid=user_data.sub.hex, + auth_type=UserAuthTypes.CUSTOM, + verified=True, ) print(user_data) diff --git a/classquiz/oauth/github.py b/classquiz/oauth/github.py index 3d0f53b..29b72e4 100644 --- a/classquiz/oauth/github.py +++ b/classquiz/oauth/github.py @@ -4,7 +4,6 @@ import uuid -from typing import Optional import asyncpg import authlib.integrations.base_client @@ -35,43 +34,43 @@ class Plan(BaseModel): class GitHubOauthResponse(BaseModel): login: str id: int - email: Optional[str] = None - node_id: Optional[str] = None - avatar_url: Optional[str] = None - gravatar_id: Optional[str] = None - url: Optional[str] = None - html_url: Optional[str] = None - followers_url: Optional[str] = None - following_url: Optional[str] = None - gists_url: Optional[str] = None - starred_url: Optional[str] = None - subscriptions_url: Optional[str] = None - organizations_url: Optional[str] = None - repos_url: Optional[str] = None - events_url: Optional[str] = None - received_events_url: Optional[str] = None - type: Optional[str] = None - site_admin: Optional[bool] = None - name: Optional[str] = None - company: Optional[str] = None - blog: Optional[str] = None - location: Optional[str] = None - hireable: Optional[bool] = None - bio: Optional[str] = None - twitter_username: Optional[str] = None - public_repos: Optional[int] = None - public_gists: Optional[int] = None - followers: Optional[int] = None - following: Optional[int] = None + email: str | None = None + node_id: str | None = None + avatar_url: str | None = None + gravatar_id: str | None = None + url: str | None = None + html_url: str | None = None + followers_url: str | None = None + following_url: str | None = None + gists_url: str | None = None + starred_url: str | None = None + subscriptions_url: str | None = None + organizations_url: str | None = None + repos_url: str | None = None + events_url: str | None = None + received_events_url: str | None = None + type: str | None = None + site_admin: bool | None = None + name: str | None = None + company: str | None = None + blog: str | None = None + location: str | None = None + hireable: bool | None = None + bio: str | None = None + twitter_username: str | None = None + public_repos: int | None = None + public_gists: int | None = None + followers: int | None = None + following: int | None = None created_at: datetime updated_at: datetime - private_gists: Optional[int] = None - total_private_repos: Optional[int] = None - owned_private_repos: Optional[int] = None - disk_usage: Optional[int] = None - collaborators: Optional[int] = None - two_factor_authentication: Optional[bool] = None - plan: Optional[Plan] = None + private_gists: int | None = None + total_private_repos: int | None = None + owned_private_repos: int | None = None + disk_usage: int | None = None + collaborators: int | None = None + two_factor_authentication: bool | None = None + plan: Plan | None = None @router.get("/login") @@ -107,39 +106,20 @@ async def auth(request: Request, response: Response): user_data = GitHubOauthResponse(**data) if user_data.email is None: return RedirectResponse("/account/oauth-error?error=email") - user_in_db = await User.objects.get_or_none(email=user_data.email) - if user_in_db is None: # REGISTER USER - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=user_data.login, - verified=True, - auth_type=UserAuthTypes.GITHUB, - avatar=gzipped_user_avatar(), - ) - # skipcq: PYL-W0703 - except Exception as e: - if type(e) is asyncpg.exceptions.UniqueViolationError: - error = True - counter = 1 - while error: - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=user_data.login, - verified=True, - auth_type=UserAuthTypes.GITHUB, - avatar=gzipped_user_avatar(), - ) - error = False - except asyncpg.exceptions.UniqueViolationError: - counter += 1 - error = True - else: - raise HTTPException(status_code=500, detail=str(e)) + try: + await User.objects.create( + id=uuid.uuid4(), + email=user_data.email, + username=user_data.login, + verified=True, + auth_type=UserAuthTypes.GITHUB, + avatar=gzipped_user_avatar(), + ) + except asyncpg.exceptions.UniqueViolationError: + raise HTTPException(status_code=400, detail="User already exists.") + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) user = await User.objects.get_or_none( email=user_data.email, username=user_data.login, diff --git a/classquiz/oauth/google.py b/classquiz/oauth/google.py index 45bb0d4..058ffdc 100644 --- a/classquiz/oauth/google.py +++ b/classquiz/oauth/google.py @@ -79,43 +79,26 @@ async def auth(request: Request, response: Response): user_data = OauthGoogleResponse(**user_data).userinfo except (TypeError, ValidationError): raise HTTPException(status_code=401, detail="Something went wrong.") - user_in_db = await User.objects.get_or_none(email=user_data.email) - if user_in_db is None: # REGISTER USER - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=user_data.name, - verified=user_data.email_verified, - auth_type=UserAuthTypes.GOOGLE, - google_uid=user_data.sub, - avatar=gzipped_user_avatar(), - ) - # skipcq: PYL-W0703 - except Exception as e: - if type(e) is asyncpg.exceptions.UniqueViolationError: - error = True - counter = 1 - while error: - try: - await User.objects.create( - id=uuid.uuid4(), - email=user_data.email, - username=f"{user_data.name}{counter}", - verified=user_data.email_verified, - auth_type=UserAuthTypes.GOOGLE, - google_uid=user_data.sub, - avatar=gzipped_user_avatar(), - ) - error = False - except asyncpg.exceptions.UniqueViolationError: - counter += 1 - error = True - else: - raise HTTPException(status_code=500, detail=str(e)) + try: + await User.objects.create( + id=uuid.uuid4(), + email=user_data.email, + username=user_data.name, + verified=user_data.email_verified, + auth_type=UserAuthTypes.GOOGLE, + google_uid=user_data.sub, + avatar=gzipped_user_avatar(), + ) + except asyncpg.exceptions.UniqueViolationError: + raise HTTPException(status_code=400, detail="User already exists.") + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) user = await User.objects.get_or_none( - email=user_data.email, google_uid=user_data.sub, auth_type=UserAuthTypes.GOOGLE, verified=True + email=user_data.email, + google_uid=user_data.sub, + auth_type=UserAuthTypes.GOOGLE, + verified=True, ) await log_user_in(user=user, request=request, response=response) diff --git a/classquiz/routers/admin.py b/classquiz/routers/admin.py index 3a0a3a5..3c0cdae 100644 --- a/classquiz/routers/admin.py +++ b/classquiz/routers/admin.py @@ -14,15 +14,15 @@ router = APIRouter() @router.delete("/user/id") -async def delete_user_by_id(user_id: UUID, user: User = Depends(get_admin_user)): +async def delete_user_by_id(user_id: UUID, _: User = Depends(get_admin_user)): return {"deleted": await User.objects.delete(id=user_id)} @router.delete("/user/username") -async def delete_user_by_username(username: str, user: User = Depends(get_admin_user)): +async def delete_user_by_username(username: str, _: User = Depends(get_admin_user)): return {"deleted": await User.objects.delete(username=username)} @router.delete("/user/email") -async def delete_user_by_email(email: str, user: User = Depends(get_admin_user)): +async def delete_user_by_email(email: str, _: User = Depends(get_admin_user)): return {"deleted": await User.objects.delete(email=email)} diff --git a/classquiz/routers/avatar.py b/classquiz/routers/avatar.py index c66079c..ef07a4a 100644 --- a/classquiz/routers/avatar.py +++ b/classquiz/routers/avatar.py @@ -83,19 +83,6 @@ async def get_customized_avatar( clothe_graphic_type=clothe_graphic_type, ).render_svg() # skipcq: PY-W0069 - # print(f"skin_color: {len(AvatarItemsAsList.skin_color)},") - # print(f"hair_color: {len(AvatarItemsAsList.hair_color)},") - # print(f"facial_hair_type: {len(AvatarItemsAsList.facial_hair_type)},") - # print(f"facial_hair_color: {len(AvatarItemsAsList.facial_hair_color)},") - # print(f"top_type: {len(AvatarItemsAsList.top_type)},") - # print(f"hat_color: {len(AvatarItemsAsList.hat_color)},") - # print(f"mouth_type: {len(AvatarItemsAsList.mouth_type)},") - # print(f"eyebrow_type: {len(AvatarItemsAsList.eyebrow_type)},") - # print(f"nose_type: {len(AvatarItemsAsList.nose_type)},") - # print(f"accessories_type: {len(AvatarItemsAsList.accessories_type)},") - # print(f"clothe_type: {len(AvatarItemsAsList.clothe_type)},") - # print(f"clothe_color: {len(AvatarItemsAsList.clothe_color)},") - # print(f"clothe_graphic_type: {len(AvatarItemsAsList.clothe_graphic_type)},") return avatar diff --git a/classquiz/routers/community.py b/classquiz/routers/community.py index fcc080a..e50545b 100644 --- a/classquiz/routers/community.py +++ b/classquiz/routers/community.py @@ -17,18 +17,24 @@ router = APIRouter() # -@router.get("/user/{user_id}", response_model_include={"username", "created_at", "id"}, response_model=User) +@router.get( + "/user/{user_id}", + response_model_include={"username", "created_at", "id"}, + response_model=User, +) async def get_user_by_user_id(user_id: UUID): user = await User.objects.get_or_none(id=user_id) - # .select_related("quizs") - # print(user) if user is None: raise HTTPException(status_code=404, detail="user not found") else: return user -@router.get("/quizzes/{user_id}", response_model_exclude={"questions", "user_id"}, response_model=list[Quiz]) +@router.get( + "/quizzes/{user_id}", + response_model_exclude={"questions", "user_id"}, + response_model=list[Quiz], +) async def get_quizzes_from_user(user_id: UUID, imported: bool | None = None): if imported is None: quizzes = await Quiz.objects.all(user_id=user_id, public=True) @@ -68,7 +74,13 @@ async def rate_quiz(data: RateQuizInput, quiz_id: uuid.UUID, user: User = Depend quiz.likes -= 1 else: quiz.dislikes -= 1 - rating = Rating(id=uuid.uuid4(), user=user, positive=positive, quiz=quiz, created_at=datetime.now()) + rating = Rating( + id=uuid.uuid4(), + user=user, + positive=positive, + quiz=quiz, + created_at=datetime.now(), + ) await rating.save() if positive: quiz.likes += 1 diff --git a/classquiz/routers/eximport.py b/classquiz/routers/eximport.py index d14a236..bd9a112 100644 --- a/classquiz/routers/eximport.py +++ b/classquiz/routers/eximport.py @@ -37,7 +37,7 @@ class UUIDEncoder(json.JSONEncoder): @router.get("/{quiz_id}") -async def export_quiz(quiz_id: uuid.UUID, user: User = Depends(get_current_user)): +async def export_quiz(quiz_id: uuid.UUID, _: User = Depends(get_current_user)): try: quiz: Quiz = await Quiz.objects.filter(Quiz.id == quiz_id).first() except ormar.exceptions.NoMatch: @@ -138,7 +138,7 @@ async def import_quiz(file: UploadFile = File(), user: User = Depends(get_curren @router.get("/excel/{quiz_id}") -async def export_quiz_as_excel(quiz_id: uuid.UUID, user: User = Depends(get_current_user)): +async def export_quiz_as_excel(quiz_id: uuid.UUID, _: User = Depends(get_current_user)): try: quiz: Quiz = await Quiz.objects.filter(Quiz.id == quiz_id).first() except ormar.exceptions.NoMatch: diff --git a/classquiz/routers/users/__init__.py b/classquiz/routers/users/__init__.py index c086c92..afbd868 100644 --- a/classquiz/routers/users/__init__.py +++ b/classquiz/routers/users/__init__.py @@ -10,22 +10,18 @@ from datetime import datetime import ormar import pydantic from email_validator import validate_email, EmailNotValidError -from fastapi import APIRouter, Response, HTTPException, Request, Depends, status +from fastapi import APIRouter, Response, HTTPException, Request, Depends from fastapi.background import BackgroundTasks from fastapi.responses import JSONResponse, RedirectResponse, PlainTextResponse -from fastapi.security import OAuth2PasswordRequestForm -from jose import jwt, JWTError from classquiz import oauth from classquiz.helpers.avatar import gzipped_user_avatar import base64 -from classquiz.oauth.authenticate_user import rememberme_check, log_user_in from classquiz.auth import ( get_password_hash, verify_password, - authenticate_user, get_current_user, ) from classquiz.cache import clear_cache_for_account @@ -33,7 +29,7 @@ from classquiz.config import redis, settings, meilisearch import uuid import bleach from pydantic import BaseModel -from classquiz.db.models import User, UserSession, UpdatePassword, Token, Quiz, ApiKey +from classquiz.db.models import User, UserSession, UpdatePassword, Quiz, ApiKey from classquiz.emails import send_register_email, send_forgotten_password_email from classquiz.routers.users import webauthn, twofa @@ -66,7 +62,12 @@ 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.model_dump(), id=uuid.uuid4(), avatar=gzipped_user_avatar(), created_at=datetime.now()) + user: 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: @@ -91,31 +92,6 @@ async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User return user -@router.post("/token/cookie", response_model=Token, deprecated=True) -async def login_for_cookie_access_token( - request: Request, - response: Response, - form_data: OAuth2PasswordRequestForm = Depends(), -): - user = await authenticate_user(form_data.username, form_data.password) - user = await User.objects.select_related("fidocredentialss").get(id=user.id) - if not user or user.totp_secret is not None or user or len(user.fidocredentialss) != 0: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Incorrect username or password", - ) - - return await log_user_in(response=response, request=request, user=user) - - -@router.get("/token/rememberme") -async def rememberme_token(request: Request, response: Response): - rememberme_token_lol = request.cookies.get("rememberme_token") - if rememberme_token_lol is None: - raise HTTPException(status_code=400, detail="No rememberme cookie") - return await rememberme_check(rememberme_token=rememberme_token_lol, response=response) - - @router.get("/logout") async def logout(request: Request, response: Response): remember_token = request.cookies.get("rememberme_token") @@ -229,24 +205,26 @@ async def reset_password_with_token(reset_password: ResetPassword, response: Res return {"message": "Password updated successfully"} -@router.get("/sessions/list", response_model=list[UserSession], response_model_exclude={"user", "session_key", "quizs"}) +@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.model_dump() for session in sessions] @router.delete("/sessions/{session_id}") -async def delete_session(session_id: str, user: User = Depends(get_current_user)): - try: - session_id = uuid.UUID(session_id) - except ValueError: - raise HTTPException(status_code=400, detail="Invalid session id") +async def delete_session(session_id: uuid.UUID, user: User = Depends(get_current_user)): await UserSession.objects.filter(user=user, id=session_id).delete() return {"message": "Session deleted"} @router.get( - "/session", response_model=UserSession, response_model_exclude={"user": ..., "session_key": ..., "quizs": ...} + "/session", + response_model=UserSession, + response_model_exclude={"user": ..., "session_key": ..., "quizs": ...}, ) async def get_session(request: Request, user: User = Depends(get_current_user)): try: @@ -300,43 +278,10 @@ class InternalAuthData(BaseModel): jwt: str | None = None -@router.post("/auth/internal") -async def internal_auth(data: InternalAuthData, resp: Response): - try: - data.jwt = data.jwt.replace("Bearer ", "") - except AttributeError: - pass - if data.jwt is not None: - try: - payload = jwt.decode(data.jwt, settings.secret_key, algorithms=["HS256"]) - email: str = payload.get("sub") - if email is None: - resp.status_code = 401 - return resp - except JWTError: - resp.status_code = 401 - return resp - else: - return await rememberme_check(data.rememberme, resp) - - class GetEmailFromJWT(BaseModel): jwt: str -@router.post("/auth/internal/email") -async def get_email_from_jwt(data: GetEmailFromJWT): - try: - data.jwt = data.jwt.replace("Bearer ", "") - except AttributeError: - pass - try: - payload = jwt.decode(data.jwt, settings.secret_key, algorithms=["HS256"]) - return payload.get("sub") - except JWTError: - raise HTTPException(status_code=401) - - @router.post("/api_keys", response_model=ApiKey, response_model_include={"key"}) async def generate_api_key(user: User = Depends(get_current_user)): key = ApiKey(key=os.urandom(24).hex(), user=user) @@ -351,7 +296,7 @@ async def list_api_keys(user: User = Depends(get_current_user)): @router.delete("/api_keys") -async def delete_api_key(api_key: str, user: User = Depends(get_current_user)): +async def delete_api_key(api_key: str, _: User = Depends(get_current_user)): key = await ApiKey.objects.get_or_none(key=api_key) if key is None: raise HTTPException(status_code=404, detail="Key not found")