fixed for more strict

Author: igorbenav

mypy.ini

@@ -1,5 +1,8 @@
 [mypy]
+python_version = 3.11
+warn_return_any = True
+warn_unused_configs = True
 ignore_missing_imports = True
-strict_optional = True
-warn_redundant_casts = True
-warn_unused_ignores = True
+
+[mypy-src.app.*]
+disallow_untyped_defs = True

src/app/api/dependencies.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Union, Any
 
 from app.core.security import SECRET_KEY, ALGORITHM, oauth2_scheme
 from app.core.config import settings
@@ -31,13 +31,13 @@
 async def get_current_user(
     token: Annotated[str, Depends(oauth2_scheme)],
     db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> dict:
+) -> Union[dict[str, Any], None]:
     token_data = await verify_token(token, db)
     if token_data is None:
         raise UnauthorizedException("User not authenticated.")
 
     if "@" in token_data.username_or_email:
-        user = await crud_users.get(db=db, email=token_data.username_or_email, is_deleted=False)
+        user: dict = await crud_users.get(db=db, email=token_data.username_or_email, is_deleted=False)
     else: 
         user = await crud_users.get(db=db, username=token_data.username_or_email, is_deleted=False)
 
@@ -76,7 +76,7 @@ async def get_optional_user(
         return None
 
 
-async def get_current_superuser(current_user: Annotated[User, Depends(get_current_user)]) -> dict:
+async def get_current_superuser(current_user: Annotated[dict, Depends(get_current_user)]) -> dict:
     if not current_user["is_superuser"]:
         raise ForbiddenException("You do not have enough privileges.")
 
@@ -87,7 +87,7 @@ async def rate_limiter(
     request: Request,
     db: Annotated[AsyncSession, Depends(async_get_db)],
     user: User | None = Depends(get_optional_user)
-):
+) -> None:
     path = sanitize_path(request.url.path)
     if user:
         user_id = user["id"]

src/app/api/v1/login.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Dict
 from datetime import timedelta
 
 from fastapi import Depends
@@ -17,7 +17,7 @@
 async def login_for_access_token(
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     user = await authenticate_user(
         username_or_email=form_data.username, 
         password=form_data.password, 

src/app/api/v1/logout.py

@@ -1,6 +1,8 @@
+from typing import Dict
+
 from datetime import datetime
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 from jose import jwt, JWTError
 
@@ -16,7 +18,7 @@
 async def logout(
     token: str = Depends(oauth2_scheme),
     db: AsyncSession = Depends(async_get_db)
-):
+) -> Dict[str, str]:
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
         expires_at = datetime.fromtimestamp(payload.get("exp"))

src/app/api/v1/posts.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Union, Dict, Any
 
 from fastapi import Request, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -23,7 +23,7 @@ async def write_post(
     post: PostCreate, 
     current_user: Annotated[UserRead, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> PostRead:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -50,7 +50,7 @@ async def read_posts(
     db: Annotated[AsyncSession, Depends(async_get_db)],
     page: int = 1,
     items_per_page: int = 10
-):
+) -> PaginatedListResponse[PostRead]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if not db_user:
         raise NotFoundException("User not found")
@@ -78,7 +78,7 @@ async def read_post(
     username: str,
     id: int, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> PostRead:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -103,7 +103,7 @@ async def patch_post(
     values: PostUpdate,
     current_user: Annotated[UserRead, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -131,7 +131,7 @@ async def erase_post(
     id: int,
     current_user: Annotated[UserRead, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -159,7 +159,7 @@ async def erase_db_post(
     username: str,
     id: int,
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
         raise NotFoundException("User not found")

src/app/api/v1/rate_limits.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Dict
 
 from fastapi import Request, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -25,7 +25,7 @@ async def write_rate_limit(
     tier_name: str,
     rate_limit: RateLimitCreate, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> RateLimitRead:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise NotFoundException("Tier not found")
@@ -48,7 +48,7 @@ async def read_rate_limits(
     db: Annotated[AsyncSession, Depends(async_get_db)],
     page: int = 1,
     items_per_page: int = 10
-):
+) -> PaginatedListResponse[RateLimitRead]:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise NotFoundException("Tier not found")
@@ -74,7 +74,7 @@ async def read_rate_limit(
     tier_name: str,
     id: int,
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> RateLimitRead:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise NotFoundException("Tier not found")
@@ -98,7 +98,7 @@ async def patch_rate_limit(
     id: int,
     values: RateLimitUpdate,
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if db_tier is None:
         raise NotFoundException("Tier not found")
@@ -134,7 +134,7 @@ async def erase_rate_limit(
     tier_name: str,
     id: int,
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise NotFoundException("Tier not found")

src/app/api/v1/tasks.py

@@ -1,5 +1,7 @@
+from typing import Dict, Optional, Any
+
 from arq.jobs import Job as ArqJob
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 
 from app.core.utils import queue
 from app.schemas.job import Job
@@ -8,12 +10,39 @@
 router = APIRouter(prefix="/tasks", tags=["tasks"])
 
 @router.post("/task", response_model=Job, status_code=201, dependencies=[Depends(rate_limiter)])
-async def create_task(message: str):
+async def create_task(message: str) -> Dict[str, str]:
+    """
+    Create a new background task.
+
+    Parameters
+    ----------
+    message: str
+        The message or data to be processed by the task.
+
+    Returns
+    -------
+    Dict[str, str]
+        A dictionary containing the ID of the created task.
+    """
     job = await queue.pool.enqueue_job("sample_background_task", message)
     return {"id": job.job_id}
 
 
 @router.get("/task/{task_id}")
-async def get_task(task_id: str):
+async def get_task(task_id: str) -> Optional[Dict[str, Any]]:
+    """
+    Get information about a specific background task.
+
+    Parameters
+    ----------
+    task_id: str
+        The ID of the task.
+
+    Returns
+    -------
+    Optional[Dict[str, Any]]
+        A dictionary containing information about the task if found, or None otherwise.
+    """
     job = ArqJob(task_id, queue.pool)
-    return await job.info()
+    job_info: dict = await job.info()
+    return job_info

src/app/api/v1/tiers.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Dict
 
 from fastapi import Request, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -23,7 +23,7 @@ async def write_tier(
     request: Request, 
     tier: TierCreate, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):  
+) -> TierRead:
     tier_internal_dict = tier.model_dump()
     db_tier = await crud_tiers.exists(db=db, name=tier_internal_dict["name"])
     if db_tier:
@@ -39,7 +39,7 @@ async def read_tiers(
     db: Annotated[AsyncSession, Depends(async_get_db)],
     page: int = 1,
     items_per_page: int = 10
-):
+) -> PaginatedListResponse[TierRead]:
     tiers_data = await crud_tiers.get_multi(
         db=db,
         offset=compute_offset(page, items_per_page),
@@ -59,7 +59,7 @@ async def read_tier(
     request: Request,
     name: str, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> TierRead:
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
         raise NotFoundException("Tier not found")
@@ -73,7 +73,7 @@ async def patch_tier(
     values: TierUpdate,
     name: str,
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
         raise NotFoundException("Tier not found")
@@ -87,7 +87,7 @@ async def erase_tier(
     request: Request,
     name: str, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
-):
+) -> Dict[str, str]:
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
         raise NotFoundException("Tier not found")