fix for a few bugs

Author: igorbenav

src/app/api/dependencies.py

@@ -9,6 +9,7 @@
 from app.core.database import async_get_db
 from app.crud.crud_users import get_user_by_email, get_user_by_username
 from app.core.models import TokenData
+from app.models.user import User
 
 async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(async_get_db)]):
     credentials_exception = HTTPException(
@@ -38,3 +39,10 @@ async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: An
         raise HTTPException(status_code=400, detai="User deleted")
 
     return user
+
+def get_current_superuser(current_user: Annotated[User, Depends(get_current_user)]) -> User:
+    if not current_user.is_superuser:
+        raise HTTPException(
+            status_code=403, detail="The user doesn't have enough privileges"
+        )
+    return current_user

src/app/api/v1/users.py

@@ -5,7 +5,7 @@
 import fastapi
 
 from app.schemas.user import UserCreate, UserUpdate, UserRead, UserBase
-from app.api.dependencies import get_current_user
+from app.api.dependencies import get_current_user, get_current_superuser
 from app.core.database import async_get_db
 from ...crud.crud_users import (
     get_user, 
@@ -98,3 +98,17 @@ async def erase_user(
 
     db_user = await delete_user(db=db, id=id, user=db_user)
     return db_user
+
+
+@router.delete("/db_user/{id}")
+async def erase_db_user(
+    id: int, 
+    current_superuser: Annotated[UserRead, Depends(get_current_superuser)],
+    db: AsyncSession = Depends(async_get_db)
+):
+    db_user = await get_user(db=db, id=id)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    db_user = await delete_user(db=db, id=id, user=db_user)
+    return db_user

src/app/crud/crud_users.py

@@ -1,10 +1,12 @@
 from datetime import datetime
+from typing import Union, Dict, Any
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, delete, update
 from passlib.context import CryptContext
 from sqlalchemy import and_
 
+
 from app.schemas.user import (
     UserCreate, 
     UserUpdate, 
@@ -56,23 +58,23 @@ async def get_users(db: AsyncSession):
 async def update_user(
         db: AsyncSession, 
         id: int, 
-        values: UserUpdate, 
+        values: Union[UserUpdate, Dict[str, Any]], 
         user: User | None = None
 ):
     user = user or await get_user(db=db, id=id)
     if user is not None:
-        internal_values = UserUpdateInternal(
-            name=values.name,
-            username=values.username,
-            email=values.email,
-            profile_image_url=values.profile_image_url,
-            updated_at=datetime.utcnow()
-        )
-        query = update(User).where(User.id == id).values(internal_values.dict(exclude_unset=True))
-        await db.execute(query)
-        await db.commit()
-        await db.refresh(user)
-
+        if isinstance(values, dict):
+            update_data = values
+        else:
+            update_data = values.model_dump(exclude_unset=True)
+        
+        update_data.update({"updated_at": datetime.utcnow()})        
+        for field in user.__dict__:
+            if field in update_data:
+                setattr(user, field, update_data[field])
+            db.add(user)
+            await db.commit()
+   
     return user
 
 async def delete_user(
@@ -86,7 +88,7 @@ async def delete_user(
             is_deleted=True,
             deleted_at=datetime.utcnow()
         )
-        query = update(User).where(User.id == id).values(values.dict(exclude_unset=True))
+        query = update(User).where(User.id == id).values(values.model_dump())
         await db.execute(query)
         await db.commit()
         await db.refresh(user)

src/app/models/user.py

@@ -18,7 +18,7 @@ class User(Base):
     email: Mapped[str] = mapped_column(String(50), unique=True, index=True)
     hashed_password: Mapped[str] = mapped_column(String)
 
-    profile_image_url: Mapped[str] = mapped_column(String, default=None)
+    profile_image_url: Mapped[str] = mapped_column(String, default="https://profileimageurl.com")
     uuid: Mapped[uuid_pkg.UUID] = mapped_column(
         default_factory=uuid_pkg.uuid4, primary_key=True, unique=True
     )

src/app/schemas/user.py

@@ -1,7 +1,7 @@
-from typing import Annotated
+from typing import Annotated, Optional
 from datetime import datetime
 
-from pydantic import BaseModel, EmailStr, StringConstraints, Field, HttpUrl
+from pydantic import BaseModel, EmailStr, Field, HttpUrl, ConfigDict
 
 from app.core.models import UUIDModel, TimestampModel, PersistentDeletion
 
@@ -27,11 +27,16 @@ class UserBase(BaseModel):
 
 
 class User(TimestampModel, UserBase, UUIDModel, PersistentDeletion):
-    profile_image_url: Annotated[HttpUrl, Field(default="profileimageurl.com")]
+    profile_image_url: Annotated[
+        str, 
+        Field(default="https://profileimageurl.com")
+    ]
     hashed_password: str
+    is_superuser: bool = False
 
 
 class UserRead(BaseModel):
+    id: int
     name: Annotated[
         str, 
         Field(
@@ -44,10 +49,12 @@ class UserRead(BaseModel):
             min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", examples=["userson"]
         )
     ]
-    profile_image_url: Annotated[HttpUrl | None, Field()]
+    profile_image_url: str
 
 
 class UserCreate(UserBase):
+    model_config = ConfigDict(extra='forbid')
+
     password: Annotated[
         str, 
         Field(
@@ -57,32 +64,51 @@ class UserCreate(UserBase):
 
 
 class UserUpdate(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
     name: Annotated[
-        str, 
+        Optional[str], 
         Field(
-            min_length=2, max_length=30, examples=["User Userson"], default=None
+            min_length=2, 
+            max_length=30, 
+            examples=["User Userberg"],
+            default=None
         )
     ]
     username: Annotated[
-        str, 
+        Optional[str], 
         Field(
-            min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", examples=["userson"], default=None
+            min_length=2, 
+            max_length=20, 
+            pattern=r"^[a-z0-9]+$", 
+            examples=["userberg"],
+            default=None
         )
     ]
     email: Annotated[
-        EmailStr, 
+        Optional[EmailStr],
         Field(
-            examples=["[email protected]"], default=None
+            examples=["[email protected]"],
+            default=None
+        )
+    ]
+    profile_image_url: Annotated[
+        Optional[str],
+        Field(
+            pattern=r"^((http|https)://)[-a-zA-Z0-9@:%._\\+~#?&//=]{2,256}\\.[a-z]{2,6}\\b([-a-zA-Z0-9@:%._\\+~#?&//=]*)$",
+            examples=["https://profileimageurl.com"],
+            default=None
         )
     ]
-    profile_image_url: Annotated[HttpUrl | None, Field(default=None)]
 
 
 class UserUpdateInternal(UserUpdate):
     updated_at: datetime
 
 
 class UserDelete(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
     is_deleted: bool
     deleted_at: datetime