diff --git a/src/random_access/database.py b/src/random_access/database.py index f002b31..6b57ca1 100644 --- a/src/random_access/database.py +++ b/src/random_access/database.py @@ -5,7 +5,7 @@ import datetime import hashlib import logging import uuid -from typing import List, Optional +from typing import Any from aiocache import Cache, cached from aiocache.serializers import PickleSerializer @@ -17,7 +17,7 @@ from random_access.settings import settings logger = logging.getLogger("uvicorn.error") # Global queue for write operations -write_queue: asyncio.Queue = asyncio.Queue() +write_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() # Expected table schemas for validation EXPECTED_SCHEMAS = { @@ -225,7 +225,7 @@ async def get_all_items(items_table): f.__name__, *args, **kwargs ), ) -async def get_session_by_token_cached(token: str, sessions_table) -> Optional[dict]: +async def get_session_by_token_cached(token: str, sessions_table) -> dict | None: """Get session by token from Airtable (cached).""" hashed = hashlib.sha256(token.encode("utf-8")).hexdigest() session = sessions_table.first(formula=match({"Token": hashed})) @@ -245,7 +245,7 @@ async def get_session_by_token_cached(token: str, sessions_table) -> Optional[di f.__name__, *args, **kwargs ), ) -async def get_user_items(user_id: str, item_instances_table) -> List[dict]: +async def get_user_items(user_id: str, item_instances_table) -> list[dict]: """Get all items for a user from the Item Instances table (cached).""" logger.info(f"Fetching user items from Item Instances table for user ID: {user_id}") try: @@ -277,7 +277,7 @@ async def get_user_items(user_id: str, item_instances_table) -> List[dict]: f.__name__, *args, **kwargs ), ) -async def get_item_by_id(item_id: str, items_table) -> Optional[dict]: +async def get_item_by_id(item_id: str, items_table) -> dict | None: """Get a specific item by ID (cached).""" logger.info(f"Fetching item from Airtable for ID: {item_id}") try: diff --git a/src/random_access/routes/items.py b/src/random_access/routes/items.py index b145faf..22f1f0a 100644 --- a/src/random_access/routes/items.py +++ b/src/random_access/routes/items.py @@ -1,7 +1,7 @@ """API routes for item endpoints.""" import datetime -from typing import Dict, List, Optional +from typing import Dict from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials @@ -47,25 +47,25 @@ class UserItemResponse(BaseModel): """Response model for user's item (simplified flat structure).""" item_id: str = Field(..., description="Unique identifier for the item") - name: Optional[str] = Field(None, description="Display name of the item") - type: Optional[str] = Field(None, description="Category or type of the item") - level: Optional[int] = Field(None, description="Required level to use this item") - rarity: Optional[str] = Field( + name: str | None = Field(None, description="Display name of the item") + type: str | None = Field(None, description="Category or type of the item") + level: int | None = Field(None, description="Required level to use this item") + rarity: str | None = Field( None, description="Rarity classification (common, rare, epic, legendary, etc.)" ) - game_name: Optional[str] = Field( + game_name: str | None = Field( None, description="Name of the game this item belongs to" ) - description: Optional[str] = Field(None, description="Description of the item") + description: str | None = Field(None, description="Description of the item") class UserItemsResponse(BaseModel): """Response model for user's complete item collection.""" user_id: str = Field(..., description="Unique identifier of the user") - user_name: Optional[str] = Field(None, description="Display name of the user") + user_name: str | None = Field(None, description="Display name of the user") total_items: int = Field(..., description="Total number of items owned by the user") - items: List[UserItemResponse] = Field( + items: list[UserItemResponse] = Field( ..., description="List of all items owned by the user" ) @@ -101,9 +101,9 @@ class ItemImageAttachment(BaseModel): """Model for item image attachment details.""" url: str = Field(..., description="Direct URL to the PNG image file") - filename: Optional[str] = Field(None, description="Original filename of the uploaded image") - type: Optional[str] = Field(None, description="MIME type of the image file") - size: Optional[int] = Field(None, description="File size in bytes") + filename: str | None = Field(None, description="Original filename of the uploaded image") + type: str | None = Field(None, description="MIME type of the image file") + size: int | None = Field(None, description="File size in bytes") class DetailedItemResponse(BaseModel): @@ -116,9 +116,9 @@ class DetailedItemResponse(BaseModel): rarity: str = Field( ..., description="Rarity classification (common, rare, epic, legendary, etc.)" ) - game_name: Optional[str] = Field(None, description="Name of the game this item belongs to") - description: Optional[str] = Field(None, description="Description of the item") - image: Optional[ItemImageAttachment] = Field(None, description="PNG image attachment details") + game_name: str | None = Field(None, description="Name of the game this item belongs to") + description: str | None = Field(None, description="Description of the item") + image: ItemImageAttachment | None = Field(None, description="PNG image attachment details") def create_items_router( @@ -129,7 +129,7 @@ def create_items_router( @router.get( "", - response_model=List[Dict[str, ItemResponse]], + response_model=list[Dict[str, ItemResponse]], summary="Get all available items", description="""Retrieve all items available in the system. diff --git a/src/random_access/routes/users.py b/src/random_access/routes/users.py index 01a9804..55eb63f 100644 --- a/src/random_access/routes/users.py +++ b/src/random_access/routes/users.py @@ -1,7 +1,5 @@ """API routes for user endpoints.""" -from typing import Optional - from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials from pydantic import BaseModel, Field @@ -20,10 +18,10 @@ class UserResponse(BaseModel): """Response model for user data.""" id: str = Field(..., description="Unique identifier for the user") - display_name: Optional[str] = Field(None, description="User's display name") - slack_id: Optional[str] = Field(None, description="User's Slack ID") - email: Optional[str] = Field(None, description="User's email address") - created: Optional[str] = Field( + display_name: str | None = Field(None, description="User's display name") + slack_id: str | None = Field(None, description="User's Slack ID") + email: str | None = Field(None, description="User's email address") + created: str | None = Field( None, description="ISO timestamp when the user account was created" ) diff --git a/src/random_access/security.py b/src/random_access/security.py index f725c14..778fd87 100644 --- a/src/random_access/security.py +++ b/src/random_access/security.py @@ -2,7 +2,7 @@ import re import secrets -from typing import Any, Dict, Optional +from typing import Any from fastapi import HTTPException, Request, status @@ -103,7 +103,7 @@ def get_client_ip(request: Request) -> str: def create_safe_error_response( error: Exception, user_message: str = "An error occurred" -) -> Dict[str, Any]: +) -> dict[str, Any]: """Create safe error response that doesn't leak sensitive information.""" if settings.is_production: return {"detail": user_message} @@ -116,7 +116,7 @@ def create_safe_error_response( } -def validate_bearer_token_format(authorization: Optional[str]) -> str: +def validate_bearer_token_format(authorization: str | None) -> str: """Validate Bearer token format and extract token.""" if not authorization: raise HTTPException( @@ -143,7 +143,7 @@ class SecurityHeaders: """Security headers middleware-like functionality.""" @staticmethod - def get_security_headers() -> Dict[str, str]: + def get_security_headers() -> dict[str, str]: """Get recommended security headers.""" headers = { "X-Content-Type-Options": "nosniff",