chore: apply new ruff rules

Author: CarlosAndreo

app/core/jwt.py

@@ -1,16 +1,14 @@
-from datetime import datetime, timedelta, timezone
-from typing import Any, Dict
+from datetime import UTC, datetime, timedelta
+from typing import Any
 
 import jwt
 
 from core.config import settings
 
 
-def create_access_token(data: Dict):
+def create_access_token(data: dict):
     to_encode = data.copy()
-    expire = datetime.now(timezone.utc) + timedelta(
-        minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
-    )
+    expire = datetime.now(UTC) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     to_encode.update({"exp": expire, "type": "access"})
     encoded_jwt = jwt.encode(
         to_encode,
@@ -20,11 +18,9 @@ def create_access_token(data: Dict):
     return encoded_jwt
 
 
-def create_refresh_token(data: Dict[str, str]) -> str:
-    to_encode: Dict[str, Any] = data.copy()
-    expire = datetime.now(timezone.utc) + timedelta(
-        days=settings.REFRESH_TOKEN_EXPIRE_DAYS
-    )
+def create_refresh_token(data: dict[str, str]) -> str:
+    to_encode: dict[str, Any] = data.copy()
+    expire = datetime.now(UTC) + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
     to_encode.update({"exp": expire, "type": "refresh"})
     encoded_jwt = jwt.encode(
         to_encode,
@@ -38,7 +34,7 @@ def decode_access_token(token: str):
     return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
 
 
-def verify_refresh_token(token: str) -> Dict | None:
+def verify_refresh_token(token: str) -> dict | None:
     payload = jwt.decode(
         token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]
     )

app/core/logger.py

@@ -2,9 +2,19 @@
 import logging
 import logging.config
 import sys
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
+
+_EXTRA_LOG_FIELDS = (
+    "request_id",
+    "method",
+    "path",
+    "client_host",
+    "query_params",
+    "status_code",
+    "duration_ms",
+)
 
 
 class JsonFormatter(logging.Formatter):
@@ -23,8 +33,8 @@ def format(self, record: logging.LogRecord) -> str:
         Returns:
             JSON-formatted string with log information.
         """
-        log_data: Dict[str, Any] = {
-            "timestamp": datetime.now(timezone.utc).isoformat(),
+        log_data: dict[str, Any] = {
+            "timestamp": datetime.now(UTC).isoformat(),
             "level": record.levelname,
             "logger": record.name,
             "module": record.module,
@@ -34,22 +44,10 @@ def format(self, record: logging.LogRecord) -> str:
         }
         if record.exc_info:
             log_data["exception"] = self.formatException(record.exc_info)
-        # core/logging.py
-        if hasattr(record, "request_id"):
-            log_data["request_id"] = getattr(record, "request_id")
-        if hasattr(record, "method"):
-            log_data["method"] = getattr(record, "method")
-        if hasattr(record, "path"):
-            log_data["path"] = getattr(record, "path")
-        if hasattr(record, "client_host"):
-            log_data["client_host"] = getattr(record, "client_host")
-        if hasattr(record, "query_params"):
-            log_data["query_params"] = getattr(record, "query_params")
-        if hasattr(record, "status_code"):
-            log_data["status_code"] = getattr(record, "status_code")
-        if hasattr(record, "duration_ms"):
-            log_data["duration_ms"] = getattr(record, "duration_ms")
-        # all routers
+        for field in _EXTRA_LOG_FIELDS:
+            value = getattr(record, field, None)
+            if value is not None:
+                log_data[field] = value
         return json.dumps(log_data, default=str)
 
 

app/middlewares/logging.py

@@ -1,6 +1,6 @@
 import time
 import uuid
-from typing import Callable
+from collections.abc import Callable
 
 from core.logger import get_logger
 from fastapi import Request, Response

app/services/auth.py

@@ -48,8 +48,8 @@ async def get_current_user(
         username = payload.get("sub")
         if not username:
             raise credentials_exception
-    except InvalidTokenError:
-        raise credentials_exception
+    except InvalidTokenError as err:
+        raise credentials_exception from err
     user = await get_user(username=username)
     if not user:
         raise credentials_exception

app/services/user.py

@@ -27,7 +27,11 @@ async def create_user(user: UserCreate) -> User | None:
         **user.model_dump(exclude={"password"}), hashed_password=hashed_password
     )
     created_user = await insert_user(user=user_in_db)
-    return User(**created_user.model_dump(exclude={"hashed_password"})) if created_user else None
+    return (
+        User(**created_user.model_dump(exclude={"hashed_password"}))
+        if created_user
+        else None
+    )
 
 
 async def change_password(user: User, new_password: str) -> User | None:
@@ -39,4 +43,8 @@ async def change_password(user: User, new_password: str) -> User | None:
         **user.model_dump(exclude={"password"}), hashed_password=hashed_password
     )
     updated_user = await update_user(user=user_in_db)
-    return User(**updated_user.model_dump(exclude={"hashed_password"})) if updated_user else None
+    return (
+        User(**updated_user.model_dump(exclude={"hashed_password"}))
+        if updated_user
+        else None
+    )