Add user management endpoints

src/api/app.py

@@ -13,6 +13,7 @@ from src.utils.logging import get_logger
 
 from .auth import router as auth_router
 from .sessions import router as sessions_router
+from .users import router as users_router
 
 log = get_logger(__name__)
 
@@ -52,6 +53,7 @@ app = FastAPI(
 )
 
 app.include_router(auth_router)
+app.include_router(users_router)
 app.include_router(sessions_router)
 
 

src/api/users.py

@@ -0,0 +1,164 @@
+from datetime import datetime
+from typing import Annotated, final
+
+from beanie import PydanticObjectId
+from fastapi import APIRouter, Body, HTTPException, Response, status
+from pydantic import BaseModel, EmailStr, StringConstraints
+from starlette.status import HTTP_200_OK
+
+from src.api.auth.dependencies import LoggedInDep
+from src.api.auth.passwords import check_hashed_password, create_password_hash
+from src.db.models.token import Token
+from src.db.models.user import User
+from src.utils.logging import get_logger
+
+from .auth import CurrentUserDep
+
+__all__ = ["router"]
+
+log = get_logger(__name__)
+
+public_router = APIRouter(tags=["Users"], prefix="/users")
+protected_router = APIRouter(tags=["Users"], prefix="/users", dependencies=[LoggedInDep])
+
+
+@final
+class RegisterData(BaseModel):
+    """Data necessary to register a new user."""
+
+    username: Annotated[str, StringConstraints(min_length=1, max_length=50, to_lower=True)]
+    password: Annotated[str, StringConstraints(min_length=1, max_length=50)]
+    email: EmailStr
+
+
+@final
+class UserData(BaseModel):
+    """Public data about a user."""
+
+    user_id: PydanticObjectId
+    username: str
+    email: EmailStr
+    created_at: datetime
+
+    @classmethod
+    def from_user(cls, user: User) -> "UserData":
+        """Construct UserData from a User instance."""
+        if user.id is None:
+            raise ValueError("Got a user without id")
+
+        return cls(
+            user_id=user.id,
+            username=user.username,
+            email=user.email,
+            created_at=user.created_at,
+        )
+
+
+@final
+class UserUpdatable(BaseModel):
+    """Data that can be modified about a user."""
+
+    username: Annotated[str, StringConstraints(min_length=1, max_length=50, to_lower=True)] | None = None
+    password: Annotated[str, StringConstraints(min_length=1, max_length=50)] | None = None
+    email: EmailStr | None = None
+
+    async def update_user(self, user: User) -> tuple[bool, User]:
+        """Update the user with data from this instance.
+
+        :return: Was the user updated?
+        """
+        updated = False
+        if self.username and self.username != user.username:
+            user.username = self.username
+            updated = True
+
+        if self.email and self.email != user.email:
+            user.email = self.email
+            updated = True
+
+        if self.password and check_hashed_password(self.password, user.password_hash):
+            user.password_hash = create_password_hash(self.password)
+            updated = True
+
+        if updated:
+            user = await user.replace()
+
+        return updated, user
+
+
+@public_router.post("/")
+async def register(reg_data: Annotated[RegisterData, Body()]) -> UserData:
+    """Register a new user."""
+    user = await User.find_one(User.username == reg_data.username)
+    if user is not None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Username already taken")
+
+    password_hash = create_password_hash(reg_data.password)
+    user = User(username=reg_data.username, email=reg_data.email, password_hash=password_hash)
+    user = await user.insert()
+    return UserData.from_user(user)
+
+
+@protected_router.get("/")
+async def get_users() -> list[UserData]:
+    """Get all registered users."""
+    users = await User.find_all().to_list()
+    return [UserData.from_user(user) for user in users]
+
+
+@protected_router.get("/{user_id}")
+async def get_user(user_id: PydanticObjectId) -> UserData:
+    """Get a specific user."""
+    user = await User.get(user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such user")
+    return UserData.from_user(user)
+
+
+@protected_router.delete("/{user_id}")
+async def delete_user(user_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Delete a specific user.
+
+    You can only delete your own account.
+    """
+    if user.id != user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You're not authorized to delete this user",
+        )
+
+    # First revoke all sessions of this user
+    tokens = await Token.find(Token.user == user).to_list()
+    for token in tokens:
+        token.revoked = True
+    await Token.replace_many(tokens)
+
+    _ = await user.delete()
+
+    return Response(status_code=HTTP_200_OK)
+
+
+@protected_router.patch("/{user_id}")
+async def patch_user(
+    user_id: PydanticObjectId,
+    user: CurrentUserDep,
+    data: Annotated[UserUpdatable, Body()],
+) -> UserData:
+    """Perform a partial update for a specific user.
+
+    You can only update your own account.
+    """
+    if user.id != user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You're not authorized to update this user",
+        )
+
+    _, user = await data.update_user(user)
+
+    return UserData.from_user(user)
+
+
+router = APIRouter()
+router.include_router(public_router)
+router.include_router(protected_router)