WPS fixes

Author: vodkar

backend/app/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py

@@ -9,6 +9,11 @@
 import sqlalchemy as sa
 from alembic import op
 
+# Constants
+STRING_FIELD_LENGTH = 255
+USER_TABLE = "user"
+ITEM_TABLE = "item"
+
 # revision identifiers, used by Alembic.
 revision = "9c0a54914c78"
 down_revision = "e2412789c190"
@@ -20,37 +25,37 @@ def upgrade() -> None:
     """Upgrade database schema."""
     # Adjust the length of the email field in the User table
     op.alter_column(
-        "user",
+        USER_TABLE,
         "email",
         existing_type=sa.String(),
-        type_=sa.String(length=255),
+        type_=sa.String(length=STRING_FIELD_LENGTH),
         existing_nullable=False,
     )
 
     # Adjust the length of the full_name field in the User table
     op.alter_column(
-        "user",
+        USER_TABLE,
         "full_name",
         existing_type=sa.String(),
-        type_=sa.String(length=255),
+        type_=sa.String(length=STRING_FIELD_LENGTH),
         existing_nullable=True,
     )
 
     # Adjust the length of the title field in the Item table
     op.alter_column(
-        "item",
+        ITEM_TABLE,
         "title",
         existing_type=sa.String(),
-        type_=sa.String(length=255),
+        type_=sa.String(length=STRING_FIELD_LENGTH),
         existing_nullable=False,
     )
 
     # Adjust the length of the description field in the Item table
     op.alter_column(
-        "item",
+        ITEM_TABLE,
         "description",
         existing_type=sa.String(),
-        type_=sa.String(length=255),
+        type_=sa.String(length=STRING_FIELD_LENGTH),
         existing_nullable=True,
     )
 
@@ -59,36 +64,36 @@ def downgrade() -> None:
     """Downgrade database schema."""
     # Revert the length of the email field in the User table
     op.alter_column(
-        "user",
+        USER_TABLE,
         "email",
-        existing_type=sa.String(length=255),
+        existing_type=sa.String(length=STRING_FIELD_LENGTH),
         type_=sa.String(),
         existing_nullable=False,
     )
 
     # Revert the length of the full_name field in the User table
     op.alter_column(
-        "user",
+        USER_TABLE,
         "full_name",
-        existing_type=sa.String(length=255),
+        existing_type=sa.String(length=STRING_FIELD_LENGTH),
         type_=sa.String(),
         existing_nullable=True,
     )
 
     # Revert the length of the title field in the Item table
     op.alter_column(
-        "item",
+        ITEM_TABLE,
         "title",
-        existing_type=sa.String(length=255),
+        existing_type=sa.String(length=STRING_FIELD_LENGTH),
         type_=sa.String(),
         existing_nullable=False,
     )
 
     # Revert the length of the description field in the Item table
     op.alter_column(
-        "item",
+        ITEM_TABLE,
         "description",
-        existing_type=sa.String(length=255),
+        existing_type=sa.String(length=STRING_FIELD_LENGTH),
         type_=sa.String(),
         existing_nullable=True,
     )

backend/app/api/deps.py

@@ -10,45 +10,56 @@
 from pydantic import ValidationError
 from sqlmodel import Session
 
+from app import constants
 from app.core import security
-from app.core.config import settings
-from app.core.db import engine
+from app.core import config, db
 from app.models import TokenPayload, User
 
 reusable_oauth2 = OAuth2PasswordBearer(
-    tokenUrl=f"{settings.API_V1_STR}/login/access-token",
+    tokenUrl=f"{config.settings.API_V1_STR}/login/access-token",
 )
 
 
 def get_db() -> Generator[Session]:
     """Get database session."""
-    with Session(engine) as session:
+    with Session(db.engine) as session:
         yield session
 
 
 SessionDep = Annotated[Session, Depends(get_db)]
 TokenDep = Annotated[str, Depends(reusable_oauth2)]
 
 
-def get_current_user(session: SessionDep, token: TokenDep) -> User:
-    """Get current user from JWT token."""
+def _validate_token(token: str) -> TokenPayload:
+    """Validate JWT token and return payload."""
     try:
         payload = jwt.decode(
             token,
-            settings.SECRET_KEY,
+            config.settings.SECRET_KEY,
             algorithms=[security.ALGORITHM],
         )
-        token_data = TokenPayload(**payload)
-    except (InvalidTokenError, ValidationError):
+    except InvalidTokenError:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Could not validate credentials",
         ) from None
+    try:
+        return TokenPayload(**payload)
+    except ValidationError:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Could not validate credentials",
+        ) from None
+
+
+def get_current_user(session: SessionDep, token: TokenDep) -> User:
+    """Get current user from JWT token."""
+    token_data = _validate_token(token)
     user = session.get(User, token_data.sub)
     if not user:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise HTTPException(status_code=constants.NOT_FOUND_CODE, detail="User not found")
     if not user.is_active:
-        raise HTTPException(status_code=400, detail="Inactive user")
+        raise HTTPException(status_code=constants.BAD_REQUEST_CODE, detail="Inactive user")
     return user
 
 
@@ -59,7 +70,7 @@ def get_current_active_superuser(current_user: CurrentUser) -> User:
     """Get current active superuser."""
     if not current_user.is_superuser:
         raise HTTPException(
-            status_code=403,
+            status_code=constants.FORBIDDEN_CODE,
             detail="The user doesn't have enough privileges",
         )
     return current_user

backend/app/api/main.py

@@ -2,13 +2,13 @@
 
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+from app.api.routes import items, login, misc, private, users
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
-api_router.include_router(utils.router)
+api_router.include_router(misc.router)
 api_router.include_router(items.router)
 
 

backend/app/api/routes/items.py

@@ -7,6 +7,7 @@
 from sqlmodel import func, select
 
 from app.api.deps import CurrentUser, SessionDep
+from app.constants import BAD_REQUEST_CODE, NOT_FOUND_CODE
 from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message
 
 router = APIRouter(prefix="/items", tags=["items"])
@@ -24,7 +25,7 @@ def read_items(
         count_statement = select(func.count()).select_from(Item)
         count = session.exec(count_statement).one()
         statement = select(Item).offset(skip).limit(limit)
-        items = session.exec(statement).all()
+        item_list = session.exec(statement).all()
     else:
         count_statement = (
             select(func.count())
@@ -38,22 +39,22 @@ def read_items(
             .offset(skip)
             .limit(limit)
         )
-        items = session.exec(statement).all()
+        item_list = session.exec(statement).all()
 
-    return ItemsPublic(data=items, count=count)
+    return ItemsPublic(item_data=item_list, count=count)
 
 
 @router.get("/{item_id}")
 def read_item(
     session: SessionDep, current_user: CurrentUser, item_id: uuid.UUID,
 ) -> ItemPublic:
     """Get item by ID."""
-    item = session.get(Item, item_id)
-    if not item:
-        raise HTTPException(status_code=404, detail="Item not found")
-    if not current_user.is_superuser and (item.owner_id != current_user.id):
-        raise HTTPException(status_code=400, detail="Not enough permissions")
-    return ItemPublic.model_validate(item)
+    db_item = session.get(Item, item_id)
+    if not db_item:
+        raise HTTPException(status_code=NOT_FOUND_CODE, detail="Item not found")
+    if not current_user.is_superuser and (db_item.owner_id != current_user.id):
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Not enough permissions")
+    return ItemPublic.model_validate(db_item)
 
 
 @router.post("/")
@@ -64,11 +65,11 @@ def create_item(
     item_in: ItemCreate,
 ) -> ItemPublic:
     """Create new item."""
-    item = Item.model_validate(item_in, update={"owner_id": current_user.id})
-    session.add(item)
+    db_item = Item.model_validate(item_in, update={"owner_id": current_user.id})
+    session.add(db_item)
     session.commit()
-    session.refresh(item)
-    return ItemPublic.model_validate(item)
+    session.refresh(db_item)
+    return ItemPublic.model_validate(db_item)
 
 
 @router.put("/{item_id}")
@@ -80,17 +81,17 @@ def update_item(
     item_in: ItemUpdate,
 ) -> ItemPublic:
     """Update an item."""
-    item = session.get(Item, item_id)
-    if not item:
-        raise HTTPException(status_code=404, detail="Item not found")
-    if not current_user.is_superuser and (item.owner_id != current_user.id):
-        raise HTTPException(status_code=400, detail="Not enough permissions")
+    db_item = session.get(Item, item_id)
+    if not db_item:
+        raise HTTPException(status_code=NOT_FOUND_CODE, detail="Item not found")
+    if not current_user.is_superuser and (db_item.owner_id != current_user.id):
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Not enough permissions")
     update_dict = item_in.model_dump(exclude_unset=True)
-    item.sqlmodel_update(update_dict)
-    session.add(item)
+    db_item.sqlmodel_update(update_dict)
+    session.add(db_item)
     session.commit()
-    session.refresh(item)
-    return ItemPublic.model_validate(item)
+    session.refresh(db_item)
+    return ItemPublic.model_validate(db_item)
 
 
 @router.delete("/{item_id}")
@@ -100,11 +101,11 @@ def delete_item(
     item_id: uuid.UUID,
 ) -> Message:
     """Delete an item."""
-    item = session.get(Item, item_id)
-    if not item:
-        raise HTTPException(status_code=404, detail="Item not found")
-    if not current_user.is_superuser and (item.owner_id != current_user.id):
-        raise HTTPException(status_code=400, detail="Not enough permissions")
-    session.delete(item)
+    db_item = session.get(Item, item_id)
+    if not db_item:
+        raise HTTPException(status_code=NOT_FOUND_CODE, detail="Item not found")
+    if not current_user.is_superuser and (db_item.owner_id != current_user.id):
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Not enough permissions")
+    session.delete(db_item)
     session.commit()
     return Message(message="Item deleted successfully")

backend/app/api/routes/login.py

@@ -9,11 +9,11 @@
 
 from app import crud
 from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
+from app.constants import BAD_REQUEST_CODE, NOT_FOUND_CODE
 from app.core import security
 from app.core.config import settings
-from app.core.security import get_password_hash
 from app.models import Message, NewPassword, Token, UserPublic
-from app.utils import (
+from app.email_utils import (
     generate_password_reset_token,
     generate_reset_password_email,
     send_email,
@@ -35,9 +35,9 @@ def login_access_token(
         password=form_data.password,
     )
     if not user:
-        raise HTTPException(status_code=400, detail="Incorrect email or password")
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Incorrect email or password")
     if not user.is_active:
-        raise HTTPException(status_code=400, detail="Inactive user")
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Inactive user")
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     return Token(
         access_token=security.create_access_token(
@@ -60,7 +60,7 @@ def recover_password(email: str, session: SessionDep) -> Message:
 
     if not user:
         raise HTTPException(
-            status_code=404,
+            status_code=NOT_FOUND_CODE,
             detail="The user with this email does not exist in the system.",
         )
     password_reset_token = generate_password_reset_token(email=email)
@@ -82,16 +82,16 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
     """Reset password."""
     email = verify_password_reset_token(token=body.token)
     if not email:
-        raise HTTPException(status_code=400, detail="Invalid token")
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Invalid token")
     user = crud.get_user_by_email(session=session, email=email)
     if not user:
         raise HTTPException(
-            status_code=404,
+            status_code=NOT_FOUND_CODE,
             detail="The user with this email does not exist in the system.",
         )
     if not user.is_active:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    hashed_password = get_password_hash(password=body.new_password)
+        raise HTTPException(status_code=BAD_REQUEST_CODE, detail="Inactive user")
+    hashed_password = security.get_password_hash(password=body.new_password)
     user.hashed_password = hashed_password
     session.add(user)
     session.commit()
@@ -109,7 +109,7 @@ def recover_password_html_content(email: str, session: SessionDep) -> HTMLRespon
 
     if not user:
         raise HTTPException(
-            status_code=404,
+            status_code=NOT_FOUND_CODE,
             detail="The user with this username does not exist in the system.",
         )
     password_reset_token = generate_password_reset_token(email=email)

backend/app/api/routes/utils.py renamed to backend/app/api/routes/misc.py

@@ -4,16 +4,17 @@
 from pydantic.networks import EmailStr
 
 from app.api.deps import get_current_active_superuser
+from app.constants import CREATED_CODE
 from app.models import Message
-from app.utils import generate_test_email, send_email
+from app.email_utils import generate_test_email, send_email
 
 router = APIRouter(prefix="/utils", tags=["utils"])
 
 
 @router.post(
     "/test-email/",
     dependencies=[Depends(get_current_active_superuser)],
-    status_code=201,
+    status_code=CREATED_CODE,
 )
 def test_email(email_to: EmailStr) -> Message:
     """Test emails."""