FastAPI로 완벽한 블로그 만들기: 사용자 시스템 추가

이전 글에서 FastAPI를 사용하여 기본적인 개인 블로그를 만들고 성공적으로 배포했습니다.

하지만 이 블로그에는 심각한 보안 문제가 있습니다. 누구나 마음대로 글을 생성할 수 있다는 것이죠.

다음 튜토리얼에서는 이 블로그를 더 안전하게 만들기 위해 사용자 및 인증 시스템을 추가할 것입니다.

서론이 길었습니다. 시작해 봅시다.

인증 방식 소개

웹 개발에서 가장 일반적인 두 가지 인증 방식은 토큰 기반(예: JWT) 과 세션 기반(쿠키) 입니다.

JWT (JSON Web Tokens): 현재 가장 인기 있는 인증 방식입니다. 사용자가 로그인하면 서버는 토큰을 생성하여 클라이언트로 반환합니다. 클라이언트는 후속 요청에 이 토큰을 포함하고, 서버는 토큰을 검증하여 사용자 신원을 확인하기만 하면 됩니다. 서버는 사용자 상태를 저장할 필요가 없으므로, 분산되고 수평적으로 확장 가능한 대규모 애플리케이션에 매우 적합합니다.
세션-쿠키: 사용자가 로그인하면 서버는 세션을 생성하고 세션 ID를 쿠키를 통해 브라우저로 반환합니다. 브라우저는 후속 요청에 이 쿠키를 자동으로 포함합니다. 서버는 세션 ID를 기반으로 해당 세션 정보를 찾아 사용자를 식별합니다.

이 튜토리얼에서는 전통적인 세션-쿠키 방식을 선택할 것입니다. 블로그가 단순한 아키텍처를 가진 모놀리스이므로, 인증에 세션-쿠키를 사용하는 것이 가장 직접적이고 고전적이며 충분히 안전한 접근 방식입니다.

1단계: 사용자 모듈 생성

인증을 처리하기 전에 사용자 시스템이 필요합니다.

1. 사용자 데이터 모델 생성

models.py 파일을 열고 Post 클래스 위에 User 모델을 추가합니다.

# models.py
import uuid
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str = Field(unique=True, index=True)
    password: str # 저장되는 비밀번호는 암호화된 해시입니다.

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)

첫 번째 글에서 main.py의 create_db_and_tables 함수를 설정했으므로, 애플리케이션 시작 시 모든 SQLModel 모델을 자동으로 감지하고 해당 데이터베이스 테이블을 생성합니다. 따라서 SQL 문을 수동으로 실행할 필요가 없습니다.

SQL을 수동으로 실행해야 하고 데이터베이스가 Leapcell에서 생성되었다면,

그래픽 인터페이스를 사용하여 SQL 문을 쉽게 실행할 수 있습니다. 웹사이트의 데이터베이스 관리 페이지로 이동하여 위 문장들을 SQL 인터페이스에 붙여넣고 실행하기만 하면 됩니다.

2. 비밀번호 암호화 라이브러리 설치

보안을 위해 사용자 비밀번호는 절대로 일반 텍스트로 데이터베이스에 저장해서는 안 됩니다. 비밀번호를 해싱하기 위해 bcrypt 라이브러리를 사용할 것입니다.

먼저 requirements.txt 파일에 bcrypt를 추가합니다:

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

그런 다음 설치 명령을 실행합니다:

pip install -r requirements.txt

2단계: 사용자 등록 및 유효성 검사 로직 구현

다음으로 사용자 데이터를 처리하고 비밀번호를 유효성 검사하는 함수를 만들 것입니다.

프로젝트의 루트 디렉토리에 users_service.py라는 새 파일을 만들고 사용자 관련 비즈니스 로직을 저장합니다.

# users_service.py
import bcrypt
from sqlmodel import Session, select
from models import User

def get_user_by_username(username: str, session: Session) -> User | None:
    """사용자 이름으로 사용자 찾기"""
    statement = select(User).where(User.username == username)
    return session.exec(statement).first()

def create_user(user_data: dict, session: Session) -> User:
    """새 사용자를 생성하고 비밀번호를 해싱합니다."""
    # 일반 텍스트 비밀번호를 바이트로 변환
    password_bytes = user_data["password"].encode('utf-8')
    # 솔트(salt)를 생성하고 비밀번호를 해싱
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(password_bytes, salt)
    
    new_user = User(
        username=user_data["username"],
        # 데이터베이스에 저장하기 위해 해시된 비밀번호(바이트)를 문자열로 디코딩
        password=hashed_password.decode('utf-8')
    )
    
    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

그런 다음 사용자 인증을 처리하기 위해 auth_service.py 파일을 만듭니다.

# auth_service.py
import bcrypt
from sqlmodel import Session
from models import User
from users_service import get_user_by_username

def validate_user(username: str, plain_password: str, session: Session) -> User | None:
    """사용자 이름과 비밀번호가 일치하는지 확인합니다."""
    user = get_user_by_username(username, session)
    if not user:
        return None
        
    # 입력된 일반 텍스트 비밀번호와 저장된 해시 비밀번호를 모두 바이트로 인코딩
    plain_password_bytes = plain_password.encode('utf-8')
    hashed_password_bytes = user.password.encode('utf-8')
    
    # 비교를 위해 bcrypt.checkpw 사용
    if bcrypt.checkpw(plain_password_bytes, hashed_password_bytes):
        return user # 유효성 검사 성공, 사용자 정보 반환
        
    return None # 유효성 검사 실패

3단계: 로그인 및 등록 페이지 생성

사용자가 등록하고 로그인할 수 있는 인터페이스를 제공해야 합니다. templates 폴더에 login.html과 register.html 파일을 만듭니다.

register.html

{% include "_header.html" %}
<form action="/users/register" method="POST" class="post-form">
  <h2>Register</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Register</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  이미 계정이 있으신가요? <a href="/auth/login">여기를 클릭하여 로그인</a>.
</p>
{% include "_footer.html" %}

login.html

{% include "_header.html" %}
<form action="/auth/login" method="POST" class="post-form">
  <h2>Login</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Login</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  계정이 없으신가요? <a href="/users/register">여기를 클릭하여 등록</a>.
</p>
{% include "_footer.html" %}

동시에, 상단 오른쪽 코너에 등록 및 로그인 링크를 추가하기 위해 _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>
      <a href="/posts/new" class="new-post-btn">New Post</a>
      <a href="/users/register" class="nav-link">Register</a>
      <a href="/auth/login" class="nav-link">Login</a>
    </nav>
  </header>
  <main>

4단계: 라우팅 및 컨트롤러 로직 구현

이제 프로젝트 구조를 더 명확하게 하기 위해 라우팅 로직을 다른 파일로 분할할 것입니다.

프로젝트 루트 디렉토리에 routers 폴더를 생성합니다.

Post 관련 라우트(@app.get("/posts", ...) 등)를 모두 main.py에서 잘라내어 routers/posts.py 파일에 붙여넣습니다.

# 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

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

@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)):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home"})

@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post"})

@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session)
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(request: Request, post_id: uuid.UUID, session: Session = Depends(get_session)):
    post = session.get(Post, post_id)
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title})

참고: @app 데코레이터 대신 @router를 사용합니다.

routers 폴더에 사용자 등록을 처리할 users.py를 생성합니다.

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

from database import get_session
import users_service

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

@router.get("/users/register", response_class=HTMLResponse)
def show_register_form(request: Request):
    return templates.TemplateResponse("register.html", {"request": request, "title": "Register"})

@router.post("/users/register")
def register_user(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    # 편의상 복잡한 유효성 검사는 없음
    users_service.create_user({"username": username, "password": password}, session)
    return RedirectResponse(url="/auth/login", status_code=302)

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(
    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")
    
    # 유효성 검사 성공
    return RedirectResponse(url="/posts", status_code=302)

마지막으로 main.py를 업데이트하여 이전 라우트를 제거하고 새 라우터 파일을 포함합니다.

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

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

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

app = FastAPI(lifespan=lifespan)

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

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