Removed deprecated typing classes and replaced with native Python dict, list, type, etc.

src/app/api/paginated.py

@@ -1,12 +1,12 @@
-from typing import Any, Dict, Generic, List, TypeVar
+from typing import Any, Generic, TypeVar
 
 from pydantic import BaseModel
 
 SchemaType = TypeVar("SchemaType", bound=BaseModel)
 
 
 class ListResponse(BaseModel, Generic[SchemaType]):
-    data: List[SchemaType]
+    data: list[SchemaType]
 
 
 class PaginatedListResponse(ListResponse[SchemaType]):
@@ -16,7 +16,7 @@ class PaginatedListResponse(ListResponse[SchemaType]):
     items_per_page: int | None = None
 
 
-def paginated_response(crud_data: dict, page: int, items_per_page: int) -> Dict[str, Any]:
+def paginated_response(crud_data: dict, page: int, items_per_page: int) -> dict[str, Any]:
     """
     Create a paginated response based on the provided data and pagination parameters.
 
@@ -31,7 +31,7 @@ def paginated_response(crud_data: dict, page: int, items_per_page: int) -> Dict[
 
     Returns
     -------
-    Dict[str, Any]
+    dict[str, Any]
         A structured paginated response dict containing the list of items, total count, pagination flags, and numbers.
 
     Note

src/app/api/v1/login.py

@@ -1,5 +1,5 @@
 from datetime import timedelta
-from typing import Annotated, Dict
+from typing import Annotated
 
 import fastapi
 from fastapi import Depends, Request, Response
@@ -26,7 +26,7 @@ async def login_for_access_token(
     response: Response,
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     db: Annotated[AsyncSession, Depends(async_get_db)],
-) -> Dict[str, str]:
+) -> dict[str, str]:
     user = await authenticate_user(username_or_email=form_data.username, password=form_data.password, db=db)
     if not user:
         raise UnauthorizedException("Wrong username, email or password.")
@@ -45,7 +45,7 @@ async def login_for_access_token(
 
 
 @router.post("/refresh")
-async def refresh_access_token(request: Request, db: AsyncSession = Depends(async_get_db)) -> Dict[str, str]:
+async def refresh_access_token(request: Request, db: AsyncSession = Depends(async_get_db)) -> dict[str, str]:
     refresh_token = request.cookies.get("refresh_token")
     if not refresh_token:
         raise UnauthorizedException("Refresh token missing.")

src/app/api/v1/logout.py

@@ -1,5 +1,3 @@
-from typing import Dict
-
 from fastapi import APIRouter, Depends, Response
 from jose import JWTError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,7 +12,7 @@
 @router.post("/logout")
 async def logout(
     response: Response, access_token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(async_get_db)
-) -> Dict[str, str]:
+) -> dict[str, str]:
     try:
         await blacklist_token(token=access_token, db=db)
         response.delete_cookie(key="refresh_token")

src/app/api/v1/posts.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Dict
+from typing import Annotated
 
 import fastapi
 from fastapi import Depends, Request
@@ -95,7 +95,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]:
+) -> 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")
@@ -119,7 +119,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]:
+) -> 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")
@@ -140,7 +140,7 @@ async def erase_post(
 @cache("{username}_post_cache", resource_id_name="id", to_invalidate_extra={"{username}_posts": "{username}"})
 async def erase_db_post(
     request: Request, username: str, id: int, db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> Dict[str, str]:
+) -> 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, Dict
+from typing import Annotated
 
 import fastapi
 from fastapi import Depends, Request
@@ -79,7 +79,7 @@ async def patch_rate_limit(
     id: int,
     values: RateLimitUpdate,
     db: Annotated[AsyncSession, Depends(async_get_db)],
-) -> Dict[str, str]:
+) -> dict[str, str]:
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if db_tier is None:
         raise NotFoundException("Tier not found")
@@ -103,7 +103,7 @@ async def patch_rate_limit(
 @router.delete("/tier/{tier_name}/rate_limit/{id}", dependencies=[Depends(get_current_superuser)])
 async def erase_rate_limit(
     request: Request, tier_name: str, id: int, db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> Dict[str, str]:
+) -> 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,4 +1,4 @@
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from arq.jobs import Job as ArqJob
 from fastapi import APIRouter, Depends
@@ -11,7 +11,7 @@
 
 
 @router.post("/task", response_model=Job, status_code=201, dependencies=[Depends(rate_limiter)])
-async def create_task(message: str) -> Dict[str, str]:
+async def create_task(message: str) -> dict[str, str]:
     """
     Create a new background task.
 
@@ -22,15 +22,15 @@ async def create_task(message: str) -> Dict[str, str]:
 
     Returns
     -------
-    Dict[str, str]
+    dict[str, str]
         A dictionary containing the ID of the created task.
     """
     job = await queue.pool.enqueue_job("sample_background_task", message)  # type: ignore
     return {"id": job.job_id}
 
 
 @router.get("/task/{task_id}")
-async def get_task(task_id: str) -> Optional[Dict[str, Any]]:
+async def get_task(task_id: str) -> Optional[dict[str, Any]]:
     """
     Get information about a specific background task.
 
@@ -41,7 +41,7 @@ async def get_task(task_id: str) -> Optional[Dict[str, Any]]:
 
     Returns
     -------
-    Optional[Dict[str, Any]]
+    Optional[dict[str, Any]]
         A dictionary containing information about the task if found, or None otherwise.
     """
     job = ArqJob(task_id, queue.pool)

src/app/api/v1/tiers.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Dict
+from typing import Annotated
 
 import fastapi
 from fastapi import Depends, Request
@@ -50,7 +50,7 @@ async def read_tier(request: Request, name: str, db: Annotated[AsyncSession, Dep
 @router.patch("/tier/{name}", dependencies=[Depends(get_current_superuser)])
 async def patch_tier(
     request: Request, values: TierUpdate, name: str, db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> Dict[str, str]:
+) -> 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")
@@ -60,7 +60,7 @@ async def patch_tier(
 
 
 @router.delete("/tier/{name}", dependencies=[Depends(get_current_superuser)])
-async def erase_tier(request: Request, name: str, db: Annotated[AsyncSession, Depends(async_get_db)]) -> Dict[str, str]:
+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")

src/app/api/v1/users.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Any, Dict
+from typing import Annotated, Any
 
 import fastapi
 from fastapi import Depends, Request
@@ -75,7 +75,7 @@ async def patch_user(
     username: str,
     current_user: Annotated[UserRead, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)],
-) -> Dict[str, str]:
+) -> dict[str, str]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -104,7 +104,7 @@ async def erase_user(
     current_user: Annotated[UserRead, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)],
     token: str = Depends(oauth2_scheme),
-) -> Dict[str, str]:
+) -> dict[str, str]:
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username)
     if not db_user:
         raise NotFoundException("User not found")
@@ -123,7 +123,7 @@ async def erase_db_user(
     username: str,
     db: Annotated[AsyncSession, Depends(async_get_db)],
     token: str = Depends(oauth2_scheme),
-) -> Dict[str, str]:
+) -> dict[str, str]:
     db_user = await crud_users.exists(db=db, username=username)
     if not db_user:
         raise NotFoundException("User not found")
@@ -136,7 +136,7 @@ async def erase_db_user(
 @router.get("/user/{username}/rate_limits", dependencies=[Depends(get_current_superuser)])
 async def read_user_rate_limits(
     request: Request, username: str, db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     db_user: dict | None = await crud_users.get(db=db, username=username, schema_to_select=UserRead)
     if db_user is None:
         raise NotFoundException("User not found")
@@ -183,7 +183,7 @@ async def read_user_tier(
 @router.patch("/user/{username}/tier", dependencies=[Depends(get_current_superuser)])
 async def patch_user_tier(
     request: Request, username: str, values: UserTierUpdate, db: Annotated[AsyncSession, Depends(async_get_db)]
-) -> Dict[str, str]:
+) -> dict[str, str]:
     db_user = await crud_users.get(db=db, username=username, schema_to_select=UserRead)
     if db_user is None:
         raise NotFoundException("User not found")

src/app/core/security.py

@@ -1,5 +1,5 @@
 from datetime import UTC, datetime, timedelta
-from typing import Any, Dict, Literal, Union
+from typing import Any, Literal, Union
 
 import bcrypt
 from fastapi.security import OAuth2PasswordBearer
@@ -31,7 +31,7 @@ def get_password_hash(password: str) -> str:
 
 async def authenticate_user(
     username_or_email: str, password: str, db: AsyncSession
-) -> Union[Dict[str, Any], Literal[False]]:
+) -> Union[dict[str, Any], Literal[False]]:
     if "@" in username_or_email:
         db_user: dict | None = await crud_users.get(db=db, email=username_or_email, is_deleted=False)
     else:

src/app/core/setup.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Union
+from typing import Any, Union
 
 import anyio
 import fastapi
@@ -176,7 +176,7 @@ async def get_redoc_documentation() -> fastapi.responses.HTMLResponse:
                 return get_redoc_html(openapi_url="/openapi.json", title="docs")
 
             @docs_router.get("/openapi.json", include_in_schema=False)
-            async def openapi() -> Dict[str, Any]:
+            async def openapi() -> dict[str, Any]:
                 out: dict = get_openapi(title=application.title, version=application.version, routes=application.routes)
                 return out
 

src/app/core/utils/cache.py

@@ -2,7 +2,7 @@
 import json
 import re
 from collections.abc import Callable
-from typing import Any, Dict, List, Tuple, Union
+from typing import Any, Union
 
 from fastapi import Request, Response
 from fastapi.encoders import jsonable_encoder
@@ -14,7 +14,7 @@
 client: Redis | None = None
 
 
-def _infer_resource_id(kwargs: Dict[str, Any], resource_id_type: Union[type, Tuple[type, ...]]) -> int | str:
+def _infer_resource_id(kwargs: dict[str, Any], resource_id_type: Union[type, tuple[type, ...]]) -> int | str:
     """
     Infer the resource ID from a dictionary of keyword arguments.
 
@@ -53,7 +53,7 @@ def _infer_resource_id(kwargs: Dict[str, Any], resource_id_type: Union[type, Tup
     return resource_id
 
 
-def _extract_data_inside_brackets(input_string: str) -> List[str]:
+def _extract_data_inside_brackets(input_string: str) -> list[str]:
     """
     Extract data inside curly brackets from a given string using regular expressions.
 
@@ -76,7 +76,7 @@ def _extract_data_inside_brackets(input_string: str) -> List[str]:
     return data_inside_brackets
 
 
-def _construct_data_dict(data_inside_brackets: List[str], kwargs: Dict[str, Any]) -> Dict[str, Any]:
+def _construct_data_dict(data_inside_brackets: list[str], kwargs: dict[str, Any]) -> dict[str, Any]:
     """
     Construct a dictionary based on data inside brackets and keyword arguments.
 
@@ -97,7 +97,7 @@ def _construct_data_dict(data_inside_brackets: List[str], kwargs: Dict[str, Any]
     return data_dict
 
 
-def _format_prefix(prefix: str, kwargs: Dict[str, Any]) -> str:
+def _format_prefix(prefix: str, kwargs: dict[str, Any]) -> str:
     """
     Format a prefix using keyword arguments.
 
@@ -118,7 +118,7 @@ def _format_prefix(prefix: str, kwargs: Dict[str, Any]) -> str:
     return formatted_prefix
 
 
-def _format_extra_data(to_invalidate_extra: Dict[str, str], kwargs: Dict[str, Any]) -> Dict[str, Any]:
+def _format_extra_data(to_invalidate_extra: dict[str, str], kwargs: dict[str, Any]) -> dict[str, Any]:
     """
     Format extra data based on provided templates and keyword arguments.
 
@@ -191,9 +191,9 @@ def cache(
     key_prefix: str,
     resource_id_name: Any = None,
     expiration: int = 3600,
-    resource_id_type: Union[type, Tuple[type, ...]] = int,
-    to_invalidate_extra: Dict[str, Any] | None = None,
-    pattern_to_invalidate_extra: List[str] | None = None,
+    resource_id_type: Union[type, tuple[type, ...]] = int,
+    to_invalidate_extra: dict[str, Any] | None = None,
+    pattern_to_invalidate_extra: list[str] | None = None,
 ) -> Callable:
     """
     Cache decorator for FastAPI endpoints.