code for token refresh created

Author: igorbenav

src/app/api/v1/login.py

@@ -1,20 +1,28 @@
 from typing import Annotated, Dict
-from datetime import timedelta
+from datetime import timedelta, datetime, timezone
 
-from fastapi import Depends
+from fastapi import Response, Request, Depends
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 import fastapi
 
+from app.core.config import settings
 from app.core.db.database import async_get_db
-from app.core.schemas import Token
-from app.core.security import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, authenticate_user
 from app.core.exceptions.http_exceptions import UnauthorizedException
+from app.core.schemas import Token
+from app.core.security import (
+    ACCESS_TOKEN_EXPIRE_MINUTES, 
+    create_access_token, 
+    authenticate_user, 
+    create_refresh_token,
+    verify_token
+)
 
 router = fastapi.APIRouter(tags=["login"])
 
 @router.post("/login", response_model=Token)
 async def login_for_access_token(
+    response: Response,
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> Dict[str, str]:
@@ -30,5 +38,37 @@ async def login_for_access_token(
     access_token = await create_access_token(
         data={"sub": user["username"]}, expires_delta=access_token_expires
     )
+
+    refresh_token = await create_refresh_token(data={"sub": user["username"]})
+    max_age = settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60
+
+    response.set_cookie(
+        key="refresh_token",
+        value=refresh_token,
+        httponly=True,
+        secure=True,
+        samesite='Lax',
+        max_age=max_age
+    )
 
-    return {"access_token": access_token, "token_type": "bearer"}
+    return {
+        "access_token": access_token, 
+        "token_type": "bearer"
+    }
+
+
+@router.post("/refresh")
+async def refresh_access_token(
+    request: Request,
+    db: AsyncSession = Depends(async_get_db)
+) -> Dict[str, str]:
+    refresh_token = request.cookies.get("refresh_token")
+    if not refresh_token:
+        raise UnauthorizedException("Refresh token missing.")
+
+    user_data = await verify_token(refresh_token, db)
+    if not user_data:
+        raise UnauthorizedException("Invalid refresh token.")
+
+    new_access_token = await create_access_token(data={"sub": user_data.username_or_email})
+    return {"access_token": new_access_token, "token_type": "bearer"}

src/app/api/v1/logout.py

@@ -1,8 +1,6 @@
 from typing import Dict
 
-from datetime import datetime
-
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Response, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 from jose import JWTError
 
@@ -14,11 +12,14 @@
 
 @router.post("/logout")
 async def logout(
-    token: str = Depends(oauth2_scheme),
+    response: Response,
+    access_token: str = Depends(oauth2_scheme),
     db: AsyncSession = Depends(async_get_db)
 ) -> Dict[str, str]:
     try:
-        await blacklist_token(token=token, db=db)
+        await blacklist_token(token=access_token, db=db)
+        response.delete_cookie(key="refresh_token")
+
         return {"message": "Logged out successfully"}
 
     except JWTError:

src/app/core/config.py

@@ -16,8 +16,9 @@ class AppSettings(BaseSettings):
 
 class CryptSettings(BaseSettings):
     SECRET_KEY: str = config("SECRET_KEY")
-    ALGORITHM: str = config("ALGORITHM")
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = config("ACCESS_TOKEN_EXPIRE_MINUTES")
+    ALGORITHM: str = config("ALGORITHM", default="HS256")
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = config("ACCESS_TOKEN_EXPIRE_MINUTES", default=30)
+    REFRESH_TOKEN_EXPIRE_DAYS: int = config("REFRESH_TOKEN_EXPIRE_DAYS", default=7)
 
 
 class DatabaseSettings(BaseSettings):

src/app/core/db/models.py

@@ -11,6 +11,7 @@ class TimestampMixin:
     created_at: datetime = Column(DateTime, default=datetime.utcnow, server_default=text("current_timestamp(0)"))
     updated_at: datetime = Column(DateTime, nullable=True, onupdate=datetime.utcnow, server_default=text("current_timestamp(0)"))
 
+
 class SoftDeleteMixin:
     deleted_at: datetime = Column(DateTime, nullable=True)
     is_deleted: bool = Column(Boolean, default=False)

src/app/core/security.py

@@ -14,6 +14,7 @@
 SECRET_KEY = settings.SECRET_KEY
 ALGORITHM = settings.ALGORITHM
 ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES
+REFRESH_TOKEN_EXPIRE_DAYS = settings.REFRESH_TOKEN_EXPIRE_DAYS
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/login")
 crypt_context = CryptContext(schemes=["sha256_crypt"])
@@ -50,6 +51,16 @@ async def create_access_token(data: dict[str, Any], expires_delta: timedelta | N
     encoded_jwt: str = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     return encoded_jwt
 
+async def create_refresh_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str:
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+    to_encode.update({"exp": expire})
+    encoded_jwt: str = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
 async def verify_token(token: str, db: AsyncSession) -> TokenData | None:
     """
     Verify a JWT token and return TokenData if valid.