FastAPI로 완벽한 블로그 만들기: 인증 추가하기

이전 게시물에서는 FastAPI 블로그에 대한 사용자 등록 시스템과 기본 로그인 유효성 검사 로직을 성공적으로 구축했습니다. 사용자는 계정을 만들 수 있으며 애플리케이션은 사용자 이름과 비밀번호를 확인할 수 있습니다.

하지만 현재 로그인은 일회성 유효성 검사일 뿐이며, 서버는 사용자의 로그인 상태를 "기억"하지 않습니다. 페이지를 새로 고치거나 새 페이지를 방문할 때마다 사용자는 인증되지 않은 게스트로 돌아갑니다.

이 게시물에서는 미들웨어를 사용하여 블로그의 실제 사용자 로그인 상태 관리를 구현할 것입니다. 로그인해야 하는 페이지와 기능을 보호하고 사용자의 로그인 상태에 따라 인터페이스를 동적으로 업데이트하는 방법을 배울 것입니다.

세션 구성

FastAPI에서 세션 관리를 처리하기 위해 Starlette의 SessionMiddleware를 사용할 것입니다. Starlette는 FastAPI가 구축된 ASGI 프레임워크이며 SessionMiddleware는 세션 처리를 위한 공식 표준 도구입니다.

먼저 itsdangerous 라이브러리를 설치하세요. SessionMiddleware는 세션 데이터를 암호화하여 보안을 보장하기 위해 이 라이브러리에 의존합니다.

pip install itsdangerous

그런 다음 requirements.txt 파일에 추가하세요.

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt
itsdangerous

세션 저장에 Redis 사용

기본적으로 SessionMiddleware는 세션 데이터를 암호화하고 클라이언트 측 쿠키에 저장합니다. 이 접근 방식은 간단하며 백엔드 저장이 필요하지 않지만 쿠키 크기가 제한적(일반적으로 4KB)이라는 단점이 있어 많은 양의 데이터를 저장하는 데 적합하지 않습니다.

더 나은 확장성과 보안을 위해 Redis라는 고성능 인메모리 데이터베이스를 사용하여 서버 측에서 세션을 유지할 것입니다. 이를 통해 사용자가 브라우저를 닫거나 서버를 다시 시작해도 로그인 상태를 유지할 수 있습니다.

Redis가 없는 경우

Leapcell에서 Redis 인스턴스를 만들 수 있습니다. Leapcell은 백엔드 애플리케이션에 필요한 대부분의 도구를 제공합니다!

인터페이스에서 "Redis 생성" 버튼을 클릭하여 새 Redis 인스턴스를 생성합니다.

Redis 상세 페이지에는 Redis 명령을 직접 실행할 수 있는 온라인 CLI가 제공됩니다.

현재 Redis 서비스를 사용할 수 없는 경우 SessionMiddleware는 서명된 쿠키를 기본으로 사용합니다. 이 튜토리얼의 목적에는 기능에 영향을 미치지 않습니다.

Redis 관련 종속성을 설치합니다.

pip install redis

이제 main.py 파일을 열어 SessionMiddleware를 가져오고 구성합니다.

# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware # 미들웨어 가져오기
from dotenv import load_dotenv

from database import create_db_and_tables
from routers import posts, users, auth

# 환경 변수 로드
load_dotenv()

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating tables..")
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

# 환경 변수에서 비밀 키 읽기
# 'your-secret-key'를 진정으로 안전한 무작위 문자열로 바꾸는 것을 잊지 마세요. `openssl rand -hex 32`를 사용하여 생성할 수 있습니다.
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")

# SessionMiddleware 추가
app.add_middleware(
    SessionMiddleware,
    secret_key=SECRET_KEY,
    session_cookie="session_id", # 쿠키에 저장된 세션 ID의 이름
    max_age=60 * 60 * 24 * 7  # 세션은 7일 후에 만료됩니다.
)

# 정적 파일 디렉토리 마운트
app.mount("/static", StaticFiles(directory="public"), name="static")

# 라우터 포함
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)

참고: 보안을 위해 secret_key는 복잡하고 무작위로 생성된 문자열이어야 합니다. 데이터베이스 URL과 마찬가지로 하드코딩하는 대신 환경 변수를 통해 관리해야 합니다.

구성되면 SessionMiddleware는 각 요청을 자동으로 처리하여 요청 쿠키에서 세션 데이터를 구문 분석하고 이를 request.session 객체에 첨부하여 사용하도록 합니다.

실제 로그인 및 로그아웃 라우트 구현

다음으로 실제 로그인 및 로그아웃 로직을 처리하기 위해 routers/auth.py를 업데이트합니다.

# routers/auth.py
from fastapi import APIRouter, Request, Depends, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import auth_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/auth/login", response_class=HTMLResponse)
def show_login_form(request: Request):
    return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})

@router.post("/auth/login")
def login(
    request: Request, # 세션에 액세스하기 위해 Request 객체 주입
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    user = auth_service.validate_user(username, password, session)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    
    # 유효성 검사 성공, 세션에 사용자 정보 저장
    # SessionMiddleware는 후속 암호화 및 쿠키 설정을 자동으로 처리합니다.
    request.session["user"] = {"username": user.username, "id": str(user.id)}
    
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/auth/logout")
def logout(request: Request):
    # 세션 지우기
    request.session.clear()
    return RedirectResponse(url="/", status_code=302)

login 함수에서 사용자가 성공적으로 유효성이 검사된 후, 기본 사용자 정보가 포함된 딕셔너리를 request.session["user"]에 저장합니다. SessionMiddleware는 이 세션 데이터를 자동으로 암호화하고 서명하여 해당 데이터를 포함하는 쿠키를 브라우저에 설정합니다. 그러면 브라우저는 후속 모든 요청에 이 쿠키를 자동으로 포함하여 서버가 사용자의 로그인 상태를 인식할 수 있도록 합니다.

logout 함수에서 request.session.clear()를 호출하여 세션 데이터를 지우고 사용자 로그아웃을 효과적으로 처리합니다.

라우트 보호 및 UI 업데이트

이제 로그인 메커니즘이 있으므로 마지막 단계는 이를 사용하여 "게시물 만들기" 기능을 보호하고 로그인 상태에 따라 다른 UI 요소를 표시하는 것입니다.

인증 종속성 생성

FastAPI에서 라우트를 보호하는 가장 우아한 방법은 종속성 주입을 사용하는 것입니다. 사용자가 로그인했는지 확인하는 종속성 함수를 만들 것입니다.

프로젝트의 루트 디렉토리에 auth_dependencies.py라는 새 파일을 만듭니다.

# auth_dependencies.py
from fastapi import Request, Depends, HTTPException, status
from fastapi.responses import RedirectResponse

def login_required(request: Request):
    """
    사용자 로그인 여부를 확인하는 종속성.
    로그인하지 않은 경우 로그인 페이지로 리디렉션합니다.
    """
    if not request.session.get("user"):
        # HTTPException을 발생시킬 수도 있습니다.
        # raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
        return RedirectResponse(url="/auth/login", status_code=status.HTTP_302_FOUND)
    return request.session.get("user")

def get_user_from_session(request: Request) -> dict | None:
    """
    세션에서 사용자 정보 가져오기(존재하는 경우).
    이 종속성은 로그인을 강제하지 않으며 템플릿에서 사용자 정보를 편리하게 가져오는 데 사용됩니다.
    """
    return request.session.get("user")

첫 번째 함수인 login_required의 로직은 간단합니다. request.session에 user가 없으면 사용자에게 로그인 페이지로 리디렉션합니다. 존재하는 경우 사용자 정보를 반환하여 라우트 함수에서 직접 사용할 수 있도록 합니다.

종속성 적용

routers/posts.py를 열고 보호가 필요한 라우트에 login_required 종속성을 적용합니다.

# routers/posts.py
import uuid
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select

from database import get_session
from models import Post
from auth_dependencies import login_required # 종속성 가져오기

router = APIRouter()
templates = Jinja2Templates(directory="templates")

# ... 기타 라우트 ...

# 이 라우트를 보호하기 위해 종속성 적용
@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request, user: dict = Depends(login_required)):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post", "user": user})

# 이 라우트를 보호하기 위해 종속성 적용
@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session),
    user: dict = Depends(login_required) # 로그인한 사용자만 게시물을 만들 수 있도록 보장
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

# ... 기타 라우트 ...

이제 인증되지 않은 사용자가 /posts/new에 액세스하려고 하면 자동으로 로그인 페이지로 리디렉션됩니다.

프론트엔드 UI 업데이트

마지막으로 사용자 로그인 상태에 따라 다른 버튼을 표시하도록 UI를 업데이트합니다. get_user_from_session 종속성을 사용하여 사용자 정보를 검색하고 템플릿에 전달합니다.

templates/_header.html 수정:

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      {% if user %}
        <span class="welcome-msg">Welcome, {{ user.username }}</span>
        <a href="/posts/new" class="new-post-btn">New Post</a>
        <a href="/auth/logout" class="nav-link">Logout</a>
      {% else %}
        <a href="/users/register" class="nav-link">Register</a>
        <a href="/auth/login" class="nav-link">Login</a>
      {% endif %}
    </nav>
  </header>
  <main>

위 템플릿이 제대로 작동하려면 뷰에 user 정보를 전달하도록 뷰를 렌더링하는 모든 라우트를 업데이트해야 합니다.

routers/posts.py에서 뷰를 렌더링하는 모든 메서드를 수정합니다.

# routers/posts.py
# ... 가져오기 ...
from auth_dependencies import get_user_from_session, login_required # 새 종속성 가져오기

# ...

@router.get("/", response_class=HTMLResponse)
def root():
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(
    request: Request, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session) # 세션 사용자 정보 가져오기
):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    # 템플릿에 사용자 전달
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home", "user": user})

# ... new_post_form 라우트는 위에서 업데이트되었습니다 ...

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(
    request: Request, 
    post_id: uuid.UUID, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session) # 세션 사용자 정보 가져오기
):
    post = session.get(Post, post_id)
    # 템플릿에 사용자 전달
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title, "user": user})