Apply fixable lint settings with Ruff

Author: jb3

backend/__init__.py

@@ -27,7 +27,7 @@
     dsn=constants.FORMS_BACKEND_DSN,
     send_default_pii=True,
     release=SENTRY_RELEASE,
-    environment=SENTRY_RELEASE
+    environment=SENTRY_RELEASE,
 )
 
 middleware = [
@@ -36,10 +36,10 @@
         allow_origins=["https://forms.pythondiscord.com"],
         allow_origin_regex=ALLOW_ORIGIN_REGEX,
         allow_headers=[
-            "Content-Type"
+            "Content-Type",
         ],
         allow_methods=["*"],
-        allow_credentials=True
+        allow_credentials=True,
     ),
     Middleware(DatabaseMiddleware),
     Middleware(AuthenticationMiddleware, backend=JWTAuthenticationBackend()),

backend/authentication/backend.py

@@ -1,11 +1,9 @@
-import typing as t
-
 import jwt
 from starlette import authentication
 from starlette.requests import Request
 
-from backend import constants
-from backend import discord
+from backend import constants, discord
+
 # We must import user such way here to avoid circular imports
 from .user import User
 
@@ -19,20 +17,19 @@ def get_token_from_cookie(cookie: str) -> str:
         try:
             prefix, token = cookie.split()
         except ValueError:
-            raise authentication.AuthenticationError(
-                "Unable to split prefix and token from authorization cookie."
-            )
+            msg = "Unable to split prefix and token from authorization cookie."
+            raise authentication.AuthenticationError(msg)
 
         if prefix.upper() != "JWT":
-            raise authentication.AuthenticationError(
-                f"Invalid authorization cookie prefix '{prefix}'."
-            )
+            msg = f"Invalid authorization cookie prefix '{prefix}'."
+            raise authentication.AuthenticationError(msg)
 
         return token
 
     async def authenticate(
-        self, request: Request
-    ) -> t.Optional[tuple[authentication.AuthCredentials, authentication.BaseUser]]:
+        self,
+        request: Request,
+    ) -> tuple[authentication.AuthCredentials, authentication.BaseUser] | None:
         """Handles JWT authentication process."""
         cookie = request.cookies.get("token")
         if not cookie:
@@ -48,21 +45,25 @@ async def authenticate(
         scopes = ["authenticated"]
 
         if not payload.get("token"):
-            raise authentication.AuthenticationError("Token is missing from JWT.")
+            msg = "Token is missing from JWT."
+            raise authentication.AuthenticationError(msg)
         if not payload.get("refresh"):
-            raise authentication.AuthenticationError(
-                "Refresh token is missing from JWT."
-            )
+            msg = "Refresh token is missing from JWT."
+            raise authentication.AuthenticationError(msg)
 
         try:
             user_details = payload.get("user_details")
             if not user_details or not user_details.get("id"):
-                raise authentication.AuthenticationError("Improper user details.")
-        except Exception:
-            raise authentication.AuthenticationError("Could not parse user details.")
+                msg = "Improper user details."
+                raise authentication.AuthenticationError(msg)  # noqa: TRY301
+        except Exception:  # noqa: BLE001
+            msg = "Could not parse user details."
+            raise authentication.AuthenticationError(msg)
 
         user = User(
-            token, user_details, await discord.get_member(request.state.db, user_details["id"])
+            token,
+            user_details,
+            await discord.get_member(request.state.db, user_details["id"]),
         )
         if await user.fetch_admin_status(request.state.db):
             scopes.append("admin")

backend/authentication/user.py

@@ -1,4 +1,3 @@
-import typing
 import typing as t
 
 import jwt
@@ -16,7 +15,7 @@ def __init__(
         self,
         token: str,
         payload: dict[str, t.Any],
-        member: typing.Optional[models.DiscordMember],
+        member: models.DiscordMember | None,
     ) -> None:
         self.token = token
         self.payload = payload
@@ -31,11 +30,11 @@ def is_authenticated(self) -> bool:
     @property
     def display_name(self) -> str:
         """Return username and discriminator as display name."""
-        return f"{self.payload['username']}#{self.payload['discriminator']}"
+        return f"{self.payload["username"]}#{self.payload["discriminator"]}"
 
     @property
     def discord_mention(self) -> str:
-        return f"<@{self.payload['id']}>"
+        return f"<@{self.payload["id"]}>"
 
     @property
     def user_id(self) -> str:
@@ -61,9 +60,10 @@ async def get_user_roles(self, database: Database) -> list[str]:
         return roles
 
     async def fetch_admin_status(self, database: Database) -> bool:
-        self.admin = await database.admins.find_one(
-            {"_id": self.payload["id"]}
-        ) is not None
+        query = {"_id": self.payload["id"]}
+        found_admin = await database.admins.find_one(query)
+
+        self.admin = found_admin is not None
 
         return self.admin
 

backend/constants.py

@@ -18,7 +18,8 @@
 OAUTH2_CLIENT_ID = os.getenv("OAUTH2_CLIENT_ID")
 OAUTH2_CLIENT_SECRET = os.getenv("OAUTH2_CLIENT_SECRET")
 OAUTH2_REDIRECT_URI = os.getenv(
-    "OAUTH2_REDIRECT_URI", "https://forms.pythondiscord.com/callback"
+    "OAUTH2_REDIRECT_URI",
+    "https://forms.pythondiscord.com/callback",
 )
 
 GIT_SHA = os.getenv("GIT_SHA", "dev")
@@ -28,7 +29,7 @@
 
 SECRET_KEY = os.getenv("SECRET_KEY", binascii.hexlify(os.urandom(30)).decode())
 DISCORD_BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN")
-DISCORD_GUILD = os.getenv("DISCORD_GUILD", 267624335836053506)
+DISCORD_GUILD = os.getenv("DISCORD_GUILD", "267624335836053506")
 
 HCAPTCHA_API_SECRET = os.getenv("HCAPTCHA_API_SECRET")
 

backend/discord.py

@@ -2,7 +2,6 @@
 
 import datetime
 import json
-import typing
 
 import httpx
 import starlette.requests
@@ -17,7 +16,7 @@ async def fetch_bearer_token(code: str, redirect: str, *, refresh: bool) -> dict
         data = {
             "client_id": constants.OAUTH2_CLIENT_ID,
             "client_secret": constants.OAUTH2_CLIENT_SECRET,
-            "redirect_uri": f"{redirect}/callback"
+            "redirect_uri": f"{redirect}/callback",
         }
 
         if refresh:
@@ -27,9 +26,13 @@ async def fetch_bearer_token(code: str, redirect: str, *, refresh: bool) -> dict
             data["grant_type"] = "authorization_code"
             data["code"] = code
 
-        r = await client.post(f"{constants.DISCORD_API_BASE_URL}/oauth2/token", headers={
-            "Content-Type": "application/x-www-form-urlencoded"
-        }, data=data)
+        r = await client.post(
+            f"{constants.DISCORD_API_BASE_URL}/oauth2/token",
+            headers={
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+            data=data,
+        )
 
         r.raise_for_status()
 
@@ -38,9 +41,12 @@ async def fetch_bearer_token(code: str, redirect: str, *, refresh: bool) -> dict
 
 async def fetch_user_details(bearer_token: str) -> dict:
     async with httpx.AsyncClient() as client:
-        r = await client.get(f"{constants.DISCORD_API_BASE_URL}/users/@me", headers={
-            "Authorization": f"Bearer {bearer_token}"
-        })
+        r = await client.get(
+            f"{constants.DISCORD_API_BASE_URL}/users/@me",
+            headers={
+                "Authorization": f"Bearer {bearer_token}",
+            },
+        )
 
         r.raise_for_status()
 
@@ -52,15 +58,17 @@ async def _get_role_info() -> list[models.DiscordRole]:
     async with httpx.AsyncClient() as client:
         r = await client.get(
             f"{constants.DISCORD_API_BASE_URL}/guilds/{constants.DISCORD_GUILD}/roles",
-            headers={"Authorization": f"Bot {constants.DISCORD_BOT_TOKEN}"}
+            headers={"Authorization": f"Bot {constants.DISCORD_BOT_TOKEN}"},
         )
 
         r.raise_for_status()
         return [models.DiscordRole(**role) for role in r.json()]
 
 
 async def get_roles(
-    database: Database, *, force_refresh: bool = False
+    database: Database,
+    *,
+    force_refresh: bool = False,
 ) -> list[models.DiscordRole]:
     """
     Get a list of all roles from the cache, or discord API if not available.
@@ -86,23 +94,26 @@ async def get_roles(
     if len(roles) == 0:
         # Fetch roles from the API and insert into the database
         roles = await _get_role_info()
-        await collection.insert_many({
-            "name": role.name,
-            "id": role.id,
-            "data": role.json(),
-            "inserted_at": datetime.datetime.now(tz=datetime.timezone.utc),
-        } for role in roles)
+        await collection.insert_many(
+            {
+                "name": role.name,
+                "id": role.id,
+                "data": role.json(),
+                "inserted_at": datetime.datetime.now(tz=datetime.UTC),
+            }
+            for role in roles
+        )
 
     return roles
 
 
-async def _fetch_member_api(member_id: str) -> typing.Optional[models.DiscordMember]:
+async def _fetch_member_api(member_id: str) -> models.DiscordMember | None:
     """Get a member by ID from the configured guild using the discord API."""
     async with httpx.AsyncClient() as client:
         r = await client.get(
             f"{constants.DISCORD_API_BASE_URL}/guilds/{constants.DISCORD_GUILD}"
             f"/members/{member_id}",
-            headers={"Authorization": f"Bot {constants.DISCORD_BOT_TOKEN}"}
+            headers={"Authorization": f"Bot {constants.DISCORD_BOT_TOKEN}"},
         )
 
         if r.status_code == 404:
@@ -113,8 +124,11 @@ async def _fetch_member_api(member_id: str) -> typing.Optional[models.DiscordMem
 
 
 async def get_member(
-    database: Database, user_id: str, *, force_refresh: bool = False
-) -> typing.Optional[models.DiscordMember]:
+    database: Database,
+    user_id: str,
+    *,
+    force_refresh: bool = False,
+) -> models.DiscordMember | None:
     """
     Get a member from the cache, or from the discord API.
 
@@ -147,7 +161,7 @@ async def get_member(
     await collection.insert_one({
         "user": user_id,
         "data": member.json(),
-        "inserted_at": datetime.datetime.now(tz=datetime.timezone.utc),
+        "inserted_at": datetime.datetime.now(tz=datetime.UTC),
     })
     return member
 
@@ -161,7 +175,9 @@ class UnauthorizedError(exceptions.HTTPException):
 
 
 async def _verify_access_helper(
-    form_id: str, request: starlette.requests.Request, attribute: str
+    form_id: str,
+    request: starlette.requests.Request,
+    attribute: str,
 ) -> None:
     """A low level helper to validate access to a form resource based on the user's scopes."""
     form = await request.state.db.forms.find_one({"_id": form_id})

backend/middleware.py

@@ -7,22 +7,20 @@
 
 
 class DatabaseMiddleware:
-
     def __init__(self, app: ASGIApp) -> None:
         self._app = app
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         client: AsyncIOMotorClient = AsyncIOMotorClient(
             DATABASE_URL,
-            tlsAllowInvalidCertificates=True
+            tlsAllowInvalidCertificates=True,
         )
         db = client[MONGO_DATABASE]
         Request(scope).state.db = db
         await self._app(scope, receive, send)
 
 
 class ProtectedDocsMiddleware:
-
     def __init__(self, app: ASGIApp) -> None:
         self._app = app
 

backend/models/__init__.py

@@ -7,13 +7,13 @@
 
 __all__ = [
     "AntiSpam",
+    "CodeQuestion",
+    "DiscordMember",
     "DiscordRole",
     "DiscordUser",
-    "DiscordMember",
     "Form",
+    "FormList",
     "FormResponse",
-    "CodeQuestion",
     "Question",
-    "FormList",
-    "ResponseList"
+    "ResponseList",
 ]

backend/models/discord_role.py

@@ -1,13 +1,11 @@
-import typing
-
 from pydantic import BaseModel
 
 
 class RoleTags(BaseModel):
     """Meta information about a discord role."""
 
-    bot_id: typing.Optional[str]
-    integration_id: typing.Optional[str]
+    bot_id: str | None
+    integration_id: str | None
     premium_subscriber: bool
 
     def __init__(self, **data) -> None:
@@ -20,7 +18,7 @@ def __init__(self, **data) -> None:
         We manually parse the raw data to determine if the field exists, and give it a useful
         bool value.
         """
-        data["premium_subscriber"] = "premium_subscriber" in data.keys()
+        data["premium_subscriber"] = "premium_subscriber" in data
         super().__init__(**data)
 
 
@@ -31,10 +29,10 @@ class DiscordRole(BaseModel):
     name: str
     color: int
     hoist: bool
-    icon: typing.Optional[str]
-    unicode_emoji: typing.Optional[str]
+    icon: str | None
+    unicode_emoji: str | None
     position: int
     permissions: str
     managed: bool
     mentionable: bool
-    tags: typing.Optional[RoleTags]
+    tags: RoleTags | None