Add ruff linter

pyproject.toml

@@ -5,12 +5,12 @@ description = "A fully Async FastAPI boilerplate using SQLAlchemy and Pydantic 2
 authors = ["Igor Magalhaes <igor.magalhaes.r@gmail.com>"]
 license = "MIT"
 readme = "README.md"
-packages = [{include = "fastapi_boilerplate"}]
+packages = [{ include = "fastapi_boilerplate" }]
 
 [tool.poetry.dependencies]
 python = "^3.11"
 python-dotenv = "^1.0.0"
-pydantic = {extras = ["email"], version = "^2.4.1"}
+pydantic = { extras = ["email"], version = "^2.4.1" }
 fastapi = "^0.103.1"
 uvicorn = "^0.23.2"
 uvloop = "^0.17.0"
@@ -35,3 +35,42 @@ bcrypt = "^4.1.1"
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.ruff]
+target-version = "py311"
+line-length = 120
+fix = true
+select = [
+    # https://github.com/charliermarsh/ruff#pyflakes-f
+    "F", # Pyflakes
+    # https://github.com/charliermarsh/ruff#pycodestyle-e-w
+    "E", # pycodestyle
+    "W", # Warning
+    # https://github.com/charliermarsh/ruff#flake8-comprehensions-c4
+    # https://github.com/charliermarsh/ruff#mccabe-c90
+    "C", # Complexity (mccabe+) & comprehensions
+    # https://github.com/charliermarsh/ruff#pyupgrade-up
+    "UP", # pyupgrade
+    # https://github.com/charliermarsh/ruff#isort-i
+    "I", # isort
+]
+ignore = [
+    # https://github.com/charliermarsh/ruff#pycodestyle-e-w
+    "E402", # module level import not at top of file
+    # https://github.com/charliermarsh/ruff#pyupgrade-up
+    "UP006", # use-pep585-annotation
+    "UP007", # use-pep604-annotation
+    "E741",  # Ambiguous variable name
+    # "UP035", # deprecated-assertion
+]
+[tool.ruff.per-file-ignores]
+"__init__.py" = [
+    "F401", # unused import
+    "F403", # star imports
+]
+
+[tool.ruff.mccabe]
+max-complexity = 24
+
+[tool.ruff.pydocstyle]
+convention = "numpy"

src/app/api/dependencies.py

@@ -1,19 +1,14 @@
-from typing import Annotated, Union, Any
+from typing import Annotated, Any, Union
 
+from fastapi import Depends, HTTPException, Request
 from sqlalchemy.ext.asyncio import AsyncSession
-from fastapi import (
-    Depends, 
-    HTTPException, 
-    Request
-)
 
-from ..core.security import oauth2_scheme
 from ..core.config import settings
-from ..core.exceptions.http_exceptions import UnauthorizedException, ForbiddenException, RateLimitException
 from ..core.db.database import async_get_db
+from ..core.exceptions.http_exceptions import ForbiddenException, RateLimitException, UnauthorizedException
 from ..core.logger import logging
+from ..core.security import oauth2_scheme, verify_token
 from ..core.utils.rate_limit import is_rate_limited
-from ..core.security import verify_token
 from ..crud.crud_rate_limit import crud_rate_limits
 from ..crud.crud_tier import crud_tiers
 from ..crud.crud_users import crud_users
@@ -25,49 +20,46 @@
 DEFAULT_LIMIT = settings.DEFAULT_RATE_LIMIT_LIMIT
 DEFAULT_PERIOD = settings.DEFAULT_RATE_LIMIT_PERIOD
 
+
 async def get_current_user(
-    token: Annotated[str, Depends(oauth2_scheme)],
-    db: Annotated[AsyncSession, Depends(async_get_db)]
+    token: Annotated[str, Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> 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: dict | None = await crud_users.get(db=db, email=token_data.username_or_email, is_deleted=False)
-    else: 
+    else:
         user = await crud_users.get(db=db, username=token_data.username_or_email, is_deleted=False)
-    
+
     if user:
         return user
 
     raise UnauthorizedException("User not authenticated.")
 
 
-async def get_optional_user(
-    request: Request,
-    db: AsyncSession = Depends(async_get_db)
-) -> dict | None:
+async def get_optional_user(request: Request, db: AsyncSession = Depends(async_get_db)) -> dict | None:
     token = request.headers.get("Authorization")
     if not token:
         return None
 
     try:
-        token_type, _, token_value = token.partition(' ')
-        if token_type.lower() != 'bearer' or not token_value:
+        token_type, _, token_value = token.partition(" ")
+        if token_type.lower() != "bearer" or not token_value:
             return None
 
         token_data = await verify_token(token_value, db)
         if token_data is None:
             return None
 
         return await get_current_user(token_value, db=db)
-    
+
     except HTTPException as http_exc:
         if http_exc.status_code != 401:
             logger.error(f"Unexpected HTTPException in get_optional_user: {http_exc.detail}")
         return None
-    
+
     except Exception as exc:
         logger.error(f"Unexpected error in get_optional_user: {exc}")
         return None
@@ -76,29 +68,26 @@ async def get_optional_user(
 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.")
-    
+
     return current_user
 
 
 async def rate_limiter(
-    request: Request,
-    db: Annotated[AsyncSession, Depends(async_get_db)],
-    user: User | None = Depends(get_optional_user)
+    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"]
         tier = await crud_tiers.get(db, id=user["tier_id"])
         if tier:
-            rate_limit = await crud_rate_limits.get(
-                db=db,
-                tier_id=tier["id"],
-                path=path
-            )
+            rate_limit = await crud_rate_limits.get(db=db, tier_id=tier["id"], path=path)
             if rate_limit:
                 limit, period = rate_limit["limit"], rate_limit["period"]
             else:
-                logger.warning(f"User {user_id} with tier '{tier['name']}' has no specific rate limit for path '{path}'. Applying default rate limit.")
+                logger.warning(
+                    f"User {user_id} with tier '{tier['name']}' has no specific rate limit for path '{path}'. \
+                        Applying default rate limit."
+                )
                 limit, period = DEFAULT_LIMIT, DEFAULT_PERIOD
         else:
             logger.warning(f"User {user_id} has no assigned tier. Applying default rate limit.")
@@ -107,12 +96,6 @@ async def rate_limiter(
         user_id = request.client.host
         limit, period = DEFAULT_LIMIT, DEFAULT_PERIOD
 
-    is_limited = await is_rate_limited(
-        db=db,
-        user_id=user_id,
-        path=path,
-        limit=limit,
-        period=period
-    )
+    is_limited = await is_rate_limited(db=db, user_id=user_id, path=path, limit=limit, period=period)
     if is_limited:
         raise RateLimitException("Rate limit exceeded.")

src/app/api/paginated.py

@@ -1,4 +1,4 @@
-from typing import TypeVar, Generic, List, Dict, Any
+from typing import Any, Dict, Generic, List, TypeVar
 
 from pydantic import BaseModel
 
@@ -76,4 +76,4 @@ def compute_offset(page: int, items_per_page: int) -> int:
     >>> offset(3, 10)
     20
     """
-    return (page - 1) * items_per_page
+    return (page - 1) * items_per_page

src/app/api/v1/__init__.py

@@ -2,11 +2,11 @@
 
 from .login import router as login_router
 from .logout import router as logout_router
-from .users import router as users_router
 from .posts import router as posts_router
+from .rate_limits import router as rate_limits_router
 from .tasks import router as tasks_router
 from .tiers import router as tiers_router
-from .rate_limits import router as rate_limits_router
+from .users import router as users_router
 
 router = APIRouter(prefix="/v1")
 router.include_router(login_router)

src/app/api/v1/login.py

@@ -1,21 +1,21 @@
-from typing import Annotated, Dict
 from datetime import timedelta
+from typing import Annotated, Dict
 
-from fastapi import Response, Request, Depends
+import fastapi
+from fastapi import Depends, Request, Response
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
-import fastapi
 
 from ...core.config import settings
 from ...core.db.database import async_get_db
 from ...core.exceptions.http_exceptions import UnauthorizedException
 from ...core.schemas import Token
 from ...core.security import (
-    ACCESS_TOKEN_EXPIRE_MINUTES, 
-    create_access_token, 
-    authenticate_user, 
+    ACCESS_TOKEN_EXPIRE_MINUTES,
+    authenticate_user,
+    create_access_token,
     create_refresh_token,
-    verify_token
+    verify_token,
 )
 
 router = fastapi.APIRouter(tags=["login"])

src/app/api/v1/logout.py

@@ -1,12 +1,12 @@
 from typing import Dict
 
-from fastapi import APIRouter, Response, Depends
-from sqlalchemy.ext.asyncio import AsyncSession
+from fastapi import APIRouter, Depends, Response
 from jose import JWTError
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...core.security import oauth2_scheme, blacklist_token
 from ...core.db.database import async_get_db
 from ...core.exceptions.http_exceptions import UnauthorizedException
+from ...core.security import blacklist_token, oauth2_scheme
 
 router = APIRouter(tags=["login"])