Use current user information when creating a new quiz

Author: matinone

api_design.md

@@ -67,7 +67,6 @@
     * email (String, Unique): User's email address.
     * password_hash (String): Hashed password for security.
     * created_at (DateTime): Date and time of registration.
-    * last_login (DateTime): Date and time of last login.
 
 ## Quizzes Table
 * Table Name: **quizzes**

app/api/dependencies.py

@@ -0,0 +1,26 @@
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+
+from app.core.security import decode_token
+from app.core.settings import Settings, get_settings
+from app.models.database import AsyncSessionDep
+from app.models.user import User
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/tokens")
+
+
+async def get_current_user(
+    db: AsyncSessionDep,
+    token: Annotated[str, Depends(oauth2_scheme)],
+    settings: Annotated[Settings, Depends(get_settings)],
+):
+    token_data = decode_token(token=token, settings=settings)
+    user = await User.get(db=db, id=int(token_data.sub))
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
+        )
+
+    return user

app/api/endpoints/login.py

@@ -17,15 +17,15 @@
     status_code=status.HTTP_201_CREATED,
     summary="Get a new access token",
 )
-def get_access_token_from_username(
+async def get_access_token_from_username(
     db: AsyncSessionDep,
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
 ) -> Any:
     """
     Get an OAuth2 access token from a user logging in with a username and password,
     to use in future requests as an authenticated user.
     """
-    user = models.User.get_by_username(db=db, username=form_data.username)
+    user = await models.User.get_by_username(db=db, username=form_data.username)
     if not user:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,

app/api/endpoints/quiz.py

@@ -5,6 +5,7 @@
 
 import app.models as models
 import app.schemas as schemas
+from app.api.dependencies import get_current_user
 from app.models.database import AsyncSessionDep
 
 router = APIRouter(prefix="/quizzes", tags=["quiz"])
@@ -27,7 +28,12 @@ async def get_quiz_from_id(quiz_id: int, db: AsyncSessionDep) -> models.Quiz:
     summary="Create a new quiz",
     response_description="The new created quiz",
 )
-async def create_quiz(db: AsyncSessionDep, quiz: schemas.QuizCreate) -> Any:
+async def create_quiz(
+    db: AsyncSessionDep,
+    quiz: schemas.QuizCreate,
+    current_user: Annotated[models.User, Depends(get_current_user)],
+) -> Any:
+    quiz.created_by = current_user.id
     new_quiz = await models.Quiz.create(db=db, quiz=quiz)
     return new_quiz
 

app/core/custom_logging.py

@@ -22,13 +22,13 @@ def configure_logger():
 
     # logger.level("INFO", color="<green>")
 
-    logging.basicConfig(handlers=[InterceptHandler()], level=0)
-    logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]
-    for _log in ["uvicorn", "uvicorn.error", "fastapi"]:
-        _logger = logging.getLogger(_log)
-        _logger.handlers = [InterceptHandler()]
+    # logging.basicConfig(handlers=[InterceptHandler()], level=0)
+    # logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]
+    # for _log in ["uvicorn", "uvicorn.error", "fastapi"]:
+    #     _logger = logging.getLogger(_log)
+    #     _logger.handlers = [InterceptHandler()]
 
-    logger.bind(request_id=None, method=None)
+    # logger.bind(request_id=None, method=None)
 
     return logger
 

app/core/security.py

@@ -1,21 +1,16 @@
 from datetime import datetime, timedelta
-from typing import Annotated
 
-from fastapi import Depends, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer
+from fastapi import HTTPException, status
 from jose import ExpiredSignatureError, JWTError, jwt
 from passlib.context import CryptContext
 from pydantic import ValidationError
 
-import app.models as models
 from app.core.settings import Settings, get_settings
-from app.models.database import AsyncSessionDep
 from app.schemas import TokenPayload
 
 settings = get_settings()
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/tokens")
 
 
 def decode_token(token: str, settings: Settings) -> TokenPayload:
@@ -63,18 +58,3 @@ def get_password_hash(password: str) -> str:
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
     return pwd_context.verify(plain_password, hashed_password)
-
-
-async def get_current_user(
-    db: AsyncSessionDep,
-    token: Annotated[str, Depends(oauth2_scheme)],
-    settings: Annotated[str, Depends(get_settings)],
-):
-    token_data = decode_token(token=token, settings=settings)
-    user = await models.User.get(db=db, id=token_data.sub)
-    if not user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    return user

app/models/quiz.py

@@ -23,7 +23,7 @@ class Quiz(Base):
         DateTime(timezone=True), server_default=func.now()
     )
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
-    created_by: Mapped[str] = mapped_column(ForeignKey("users.id"))
+    created_by: Mapped[int] = mapped_column(ForeignKey("users.id"))
 
     # have to use "Question" to avoid circular dependencies
     questions: Mapped[list["Question"]] = relationship(

app/models/user.py

@@ -27,7 +27,6 @@ class User(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
-    last_login: Mapped[datetime] = mapped_column(DateTime(timezone=True))
 
     # have to use "Quiz" to avoid circular dependencies
     quizzes: Mapped[list["Quiz"]] = relationship("Quiz", back_populates="user")

app/schemas/quiz.py

@@ -8,7 +8,7 @@
 class QuizBase(BaseModel):
     title: str
     description: str | None = None
-    created_by: str | None = None
+    created_by: int | None = None
 
     model_config = ConfigDict(from_attributes=True)
 

app/schemas/token.py

@@ -7,4 +7,4 @@ class Token(BaseModel):
 
 
 class TokenPayload(BaseModel):
-    sub: str | None = None
+    sub: str