clean up type hints

This commit is contained in:
Micha R. Albert 2025-07-09 12:54:15 -04:00
parent 1a68c28802
commit d445a13646
Signed by: mra
SSH key fingerprint: SHA256:2JB0fGfy7m2HQXAzvSXXKm7wPTj9Z60MOjFOQGM2Y/E
4 changed files with 29 additions and 31 deletions

View file

@ -5,7 +5,7 @@ import datetime
import hashlib import hashlib
import logging import logging
import uuid import uuid
from typing import List, Optional from typing import Any
from aiocache import Cache, cached from aiocache import Cache, cached
from aiocache.serializers import PickleSerializer from aiocache.serializers import PickleSerializer
@ -17,7 +17,7 @@ from random_access.settings import settings
logger = logging.getLogger("uvicorn.error") logger = logging.getLogger("uvicorn.error")
# Global queue for write operations # 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 table schemas for validation
EXPECTED_SCHEMAS = { EXPECTED_SCHEMAS = {
@ -225,7 +225,7 @@ async def get_all_items(items_table):
f.__name__, *args, **kwargs 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).""" """Get session by token from Airtable (cached)."""
hashed = hashlib.sha256(token.encode("utf-8")).hexdigest() hashed = hashlib.sha256(token.encode("utf-8")).hexdigest()
session = sessions_table.first(formula=match({"Token": hashed})) 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 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).""" """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}") logger.info(f"Fetching user items from Item Instances table for user ID: {user_id}")
try: try:
@ -277,7 +277,7 @@ async def get_user_items(user_id: str, item_instances_table) -> List[dict]:
f.__name__, *args, **kwargs 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).""" """Get a specific item by ID (cached)."""
logger.info(f"Fetching item from Airtable for ID: {item_id}") logger.info(f"Fetching item from Airtable for ID: {item_id}")
try: try:

View file

@ -1,7 +1,7 @@
"""API routes for item endpoints.""" """API routes for item endpoints."""
import datetime import datetime
from typing import Dict, List, Optional from typing import Dict
from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials from fastapi.security import HTTPAuthorizationCredentials
@ -47,25 +47,25 @@ class UserItemResponse(BaseModel):
"""Response model for user's item (simplified flat structure).""" """Response model for user's item (simplified flat structure)."""
item_id: str = Field(..., description="Unique identifier for the item") item_id: str = Field(..., description="Unique identifier for the item")
name: Optional[str] = Field(None, description="Display name of the item") name: str | None = Field(None, description="Display name of the item")
type: Optional[str] = Field(None, description="Category or type of the item") type: str | None = Field(None, description="Category or type of the item")
level: Optional[int] = Field(None, description="Required level to use this item") level: int | None = Field(None, description="Required level to use this item")
rarity: Optional[str] = Field( rarity: str | None = Field(
None, description="Rarity classification (common, rare, epic, legendary, etc.)" 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" 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): class UserItemsResponse(BaseModel):
"""Response model for user's complete item collection.""" """Response model for user's complete item collection."""
user_id: str = Field(..., description="Unique identifier of the user") 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") 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" ..., description="List of all items owned by the user"
) )
@ -101,9 +101,9 @@ class ItemImageAttachment(BaseModel):
"""Model for item image attachment details.""" """Model for item image attachment details."""
url: str = Field(..., description="Direct URL to the PNG image file") url: str = Field(..., description="Direct URL to the PNG image file")
filename: Optional[str] = Field(None, description="Original filename of the uploaded image") filename: str | None = Field(None, description="Original filename of the uploaded image")
type: Optional[str] = Field(None, description="MIME type of the image file") type: str | None = Field(None, description="MIME type of the image file")
size: Optional[int] = Field(None, description="File size in bytes") size: int | None = Field(None, description="File size in bytes")
class DetailedItemResponse(BaseModel): class DetailedItemResponse(BaseModel):
@ -116,9 +116,9 @@ class DetailedItemResponse(BaseModel):
rarity: str = Field( rarity: str = Field(
..., description="Rarity classification (common, rare, epic, legendary, etc.)" ..., description="Rarity classification (common, rare, epic, legendary, etc.)"
) )
game_name: Optional[str] = Field(None, description="Name of the game this item belongs to") 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")
image: Optional[ItemImageAttachment] = Field(None, description="PNG image attachment details") image: ItemImageAttachment | None = Field(None, description="PNG image attachment details")
def create_items_router( def create_items_router(
@ -129,7 +129,7 @@ def create_items_router(
@router.get( @router.get(
"", "",
response_model=List[Dict[str, ItemResponse]], response_model=list[Dict[str, ItemResponse]],
summary="Get all available items", summary="Get all available items",
description="""Retrieve all items available in the system. description="""Retrieve all items available in the system.

View file

@ -1,7 +1,5 @@
"""API routes for user endpoints.""" """API routes for user endpoints."""
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials from fastapi.security import HTTPAuthorizationCredentials
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@ -20,10 +18,10 @@ class UserResponse(BaseModel):
"""Response model for user data.""" """Response model for user data."""
id: str = Field(..., description="Unique identifier for the user") id: str = Field(..., description="Unique identifier for the user")
display_name: Optional[str] = Field(None, description="User's display name") display_name: str | None = Field(None, description="User's display name")
slack_id: Optional[str] = Field(None, description="User's Slack ID") slack_id: str | None = Field(None, description="User's Slack ID")
email: Optional[str] = Field(None, description="User's email address") email: str | None = Field(None, description="User's email address")
created: Optional[str] = Field( created: str | None = Field(
None, description="ISO timestamp when the user account was created" None, description="ISO timestamp when the user account was created"
) )

View file

@ -2,7 +2,7 @@
import re import re
import secrets import secrets
from typing import Any, Dict, Optional from typing import Any
from fastapi import HTTPException, Request, status from fastapi import HTTPException, Request, status
@ -103,7 +103,7 @@ def get_client_ip(request: Request) -> str:
def create_safe_error_response( def create_safe_error_response(
error: Exception, user_message: str = "An error occurred" error: Exception, user_message: str = "An error occurred"
) -> Dict[str, Any]: ) -> dict[str, Any]:
"""Create safe error response that doesn't leak sensitive information.""" """Create safe error response that doesn't leak sensitive information."""
if settings.is_production: if settings.is_production:
return {"detail": user_message} 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.""" """Validate Bearer token format and extract token."""
if not authorization: if not authorization:
raise HTTPException( raise HTTPException(
@ -143,7 +143,7 @@ class SecurityHeaders:
"""Security headers middleware-like functionality.""" """Security headers middleware-like functionality."""
@staticmethod @staticmethod
def get_security_headers() -> Dict[str, str]: def get_security_headers() -> dict[str, str]:
"""Get recommended security headers.""" """Get recommended security headers."""
headers = { headers = {
"X-Content-Type-Options": "nosniff", "X-Content-Type-Options": "nosniff",