[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(1) 회원가입 및 로그인 구현(JWT)

목차

 

FastAPI + SQLAlchemy + Pydantic을 이용한 게시판 만들기

 

[FastAPI]FastAPI 튜토리얼

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(1) 회원가입 및 로그인 구현(JWT)

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(1.5) 디테일 수정

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(2-1)Feed CRUD - Model, Service

[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-2)Feed CRUD - Routes, main.py

[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(3)S3 버킷을 이용한 사진 업로드

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(4-1)Comment CRUD - Models, Services

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(4-2)Comment CRUD - Routes, main.py

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(5)async

 

이번 글부터는 가장 인기 있다는 FastAPI + SQLAlchemy + Pydantic 패턴을 이용해

 

간단한 게시판 만들기를 구현하려고 한다.

 

시간이 나는 대로 틈틈이 진행할 거라 속도는 어떨지 모르지만,

 

일단 Flask는 내팽개쳐 두고 FastAPI로!

 

Hello, World!

 

시작하기 전에. 가상환경 설정과 필요한 패키지 설치를 진행하자.

 

아래의 세 줄을 입력하면 필요한 환경 설정이 마무리된다.

python3.11 -m venv venv
source venv/bin/activate  # Linux/macOS
pip install "fastapi[all]" fastapi sqlalchemy pydantic python-jwt

이어서 아래와 같은 패키지 구조를 만들고

 

main.py 파일을 아래와 같이 작성한다.

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"message": "Hello, World!"}

이어서 가상환경에서 아래와 같이 입력해 서버를 켠 뒤에

uvicorn main:app --reload

http://127.0.0.1:8000/ 주소로 접근하면 아래와 같은 화면을 만날 수 있다.

 

 

db.py

 

계속해서 프로젝트 구조를 아래와 같이 구성한다.

 

다음으로, 가장 기본이 되는 디비 연결 설정부터 하고 지나간다.

 

db.py에 아래와 같이 작성한다.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base: DeclarativeMeta = declarative_base()

디비 설정 자체는 Flask와 동일해서 설명을 생략하고 싶지만,

 

미래의 나를 위해 다시 한번 설명해 보자.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker

필요한 함수와 클래스를 가져온다. 각 역할은 아래와 같다.

 

• create_engine: DB 엔진 생성을 위한 함수
• DeclarativeMeta, declarative_base: SQLAlchemy의 선언형 모델을 사용하기 위한 클래스와 함수
• sessionmaker: DB 세션을 생성하고 관리하는 클래스

DATABASE_URL = "sqlite:///./test.db"

사용할 DB의 주소를 설정한다. test.db 부분은 마음대로 바꿔 설정하면 된다.

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

먼저 create_engine을 사용해 DB 엔진을 생성해 할당한다. 이 엔진은 SQL 쿼리를 DB에 전달하는 역할을 한다.

 

계속해서 sessionmaker를 사용해 세션 클래스를 생성한 뒤 할당한다. 이 클래스가 DB와의 모든 작업을 관리한다.

 

각 설정의 의미는 다음과 같다.

 

• autocommit=False: 자동 커밋을 비활성화한다.
• autoflush=False:자동 플러시를 비활성화한다.
• bind=engine: 생성한 세션 클래스가 사용할 DB 엔진을 지정한다.

Base: DeclarativeMeta = declarative_base()

declarative_base 함수를 사용해 모든 모델이 상속받을 기본 클래스 'Base'를 생성한다.

 

user.py

 

다음으로는 SQLAlchemy와 Pydantic을 이용해 사용자 모델 파일을 작성한다.

 

여기서 사용자는 이메일을 아이디(고유 식별자)로 갖고, 닉네임과 비밀번호만을 가진 간단한 모델이다.

from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel, validator
import re

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    email = Column(String, primary_key=True, index=True)
    password = Column(String())
    nickname = Column(String())


class UserCreate(BaseModel):
    email: str
    password: str
    nickname: str


class UserInDB(UserCreate):
    pass


class UserLogin(BaseModel):
    email: str
    password: str
    
    @validator("email")
    def validate_email(cls, value):
        if not re.match(r"[^@]+@[^@]+\.[^@]+", value):
            raise ValueError("Invalid Email Format")
        return value

처음 구현해 보는 코드니 하나씩 자세히 뜯어보자

from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel, validator
import re

Base = declarative_base()

먼저 SQLAlchemy에서 컬럼과 문자열 자료형을 가져온다.

 

그 이후 DB 모델을 정의할 때 사용되는 기본 클래스를 가져오고, Pydantic의 BaseModel, validator를 가져온다.

 

마지막으로 가져온 re는 파이썬의 정규 표현식 모듈이다.

 

다음 줄에선 SQLAlchemy의 declarative_base 함수를 호출해 'Base'클래스를 생성한다.

class User(Base):
    __tablename__ = "users"

    email = Column(String, primary_key=True, index=True)
    password = Column(String())
    nickname = Column(String())

먼저 'Base'를 상속받는 'User'라는 이름의 SQLAlchemy 모델 클래스를 정의하고 테이블 이름을 'users'로 지정한다.

 

다음으로 email을 기본키로 설정한 뒤 index=True를 통해 검색 성능을 향상한다.

 

다음 두 줄은 비밀번호와 닉네임을 저장하는 컬럼을 만드는 역할이다.

class UserCreate(BaseModel):
    email: str
    password: str
    nickname: str

이어서 세 개의 Pydantic 모델을 정의하는데, 위 모델은 회원 가입을 위한 모델로,

 

'BaseModel'을 상속받아 Pydantic의 유효성 검사 기능을 사용한다.

class UserInDB(UserCreate):
    pass

이 모델은 DB에 저장되는 사용자 정보를 나타낸다.

 

'UserCreate'모델을 상속받기 때문에 email, password, nickname 필드는 이미 있다.

 

추가적인 사용자 필드를 쉽게 추가할 수 있도록 구성된 코드지만, 여기서는 아무 역할이 없으므로 pass 해준다.

class UserLogin(BaseModel):
    email: str
    password: str
    
    @validator("email")
    def validate_email(cls, value):
        if not re.match(r"[^@]+@[^@]+\.[^@]+", value):
            raise ValueError("Invalid Email Format")
        return value

마지막 Pydantic 모델로, 로그인 시 필요한 정보를 담는 모델이다.

 

당연히 이메일과 비밀번호만 포함되며, 로그인 로직 처리 시 사용자로부터 받은 데이터의 유효성을 검사하는 데 사용된다.

 

이메일 형식이 맞는지 검증하기 위해 아래의 @validator 데코레이터를 사용하는데,

 

re.match() 함수는 문자열의 시작 부분부터 패턴이 일치하는지 확인하는 역할을 한다.

 

또한 cls는 클래스 메서드에 관례적으로 넣어주는 첫 번째 매개변수라고 한다.

 

 

auth.service.py

 

계속해서 서비스로직을 작성해 보자. JWT 사용을 위해서 아래의 패키지를 설치해 준다.

pip install "python-jose[cryptography]" passlib

이어서 서비스 로직을 간결하게 구현하면 아래와 같다.

from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from jose import JWTError, jwt
from passlib.context import CryptContext
from models.user import User

SECRET_KEY = "ThisIsTheSecretKeyOfFastAPIApplicationWithSQLAlchemyAndPydantic"
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def get_password_hash(password):
    return pwd_context.hash(password)


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=1)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt


def create_user(db: Session, user: User):
    existing_user = get_user_by_email(db, user.email)
    if existing_user:
        raise ValueError("Email already registered")

    hashed_password = get_password_hash(user.password)
    db_user = User(email=user.email, password=hashed_password, nickname=user.nickname)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    return db_user


def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()


def authenticate_user(db: Session, email: str, password: str):
    user = get_user_by_email(db, email)
    if not user:
        raise ValueError("Invalid Credentials")
    if not verify_password(password, user.password):
        raise ValueError("Invalid Credentials")
    return user

코드를 뜯어보자.

from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from jose import JWTError, jwt
from passlib.context import CryptContext
from models.user import User

필요한 함수와 라이브러리를 가져온다.

 

여기서 datetime은 JWT의 만료시간을 설정할 때 사용하며, jose는 JWT 생성 및 검증에 사용된다.

SECRET_KEY = "ThisIsTheSecretKeyOfFastAPIApplicationWithSQLAlchemyAndPydantic"
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

JWT를 생성할 때 사용될 시크릿 키와 알고리즘을 정하고,

 

비밀번호 암호화 및 검증을 위한 설정을 해준다.

def get_password_hash(password):
    return pwd_context.hash(password)

전달받은 비밀번호를 해시한다. 반환 타입은 str이다.

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

해시된 비밀번호와 원래 비밀번호를 비교해 검증한다. 반환 타입은 bool이다.

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=1)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt

주어진 데이터로 JWT를 생성한다. 반환 타입은 str이다.

def create_user(db: Session, user: User):
    existing_user = get_user_by_email(db, user.email)
    if existing_user:
        raise ValueError("Email already registered")

    hashed_password = get_password_hash(user.password)
    db_user = User(email=user.email, password=hashed_password, nickname=user.nickname)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    return db_user

새로운 사용자를 생성한다. 중복된 이메일이 있으면(이미 가입되어 있으면) 예외를 발생시킨다.

 

반환 타입은 User이다.

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

이메일로 사용자를 조회한다. 반환 타입은 User이다.

def authenticate_user(db: Session, email: str, password: str):
    user = get_user_by_email(db, email)
    if not user:
        raise ValueError("Invalid Credentials")
    if not verify_password(password, user.password):
        raise ValueError("Invalid Credentials")
    return user

사용자의 이메일과 비밀번호를 검증해 해당 User를 반환하거나 ValueError를 반환한다.

 

auth.py

 

계속해서 라우트 파일을 구현한다.

 

이 라우팅 계층은 MVC의 컨트롤러와 거의 같은 역할을 한다고 보면 된다.

 

다만 FastAPI의 라우팅 계층은 스프링 MVC의 컨트롤러에 비해 훨씬 편하게 쓸 수 있는데,

 

대략 아래와 같은 차이점이 두드러진다.

 

• 의존성 주입: FastAPI는 'Depends' 의존성 주입 메커니즘을 직접 제공한다. 
• 타입 및 유효성 검사: Pydantic 모델과 함께 사용하면 요청 및 응답에 대해 자동으로 유효성 검사를 할 수 있다.
• 문서 자동화: swagger를 기반으로 한 API 문서를 자동으로 만들어준다.
• 비동기 지원: 파이썬의 async/await을 사용해 비동기 처리를 쉽게 할 수 있다.

전부 MVC 컨트롤러에서는 수동으로 작성해주어야만 하는 기능이다.

 

실제로 얼마나 편한지는 앞으로 알아가 보도록 하자.

 

우선 여기선 아래와 같이 구현한다.

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from config import db as config
from models.user import UserCreate, UserLogin
from services import auth_service

router = APIRouter()

@router.post("/signup", response_model=UserCreate)
def signup(user: UserCreate, db: Session = Depends(config.get_db)):
    try:
        return auth_service.create_user(db, user)
    except ValueError:
        raise HTTPException(status_code=400, detail="Email Already Registered")
    
@router.post("/login")
def login(user: UserLogin, db: Session = Depends(config.get_db)):
    try:
        db_user = auth_service.authenticate_user(db, user.email, user.password)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid Credentials")

    access_token = auth_service.create_access_token({"sub": user.email})
    return {"access_token": access_token, "email": db_user.email, "nickname": db_user.nickname}

코드를 뜯어서 읽어보면 다음과 같다.

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from config import db as config
from models.user import UserCreate
from services import auth_service

router = APIRouter()

필요한 모듈과 패키지를 가져온 뒤, FastApi의 APIRouter를 사용해 라우터 객체를 생성한다.

@router.post("/signup", response_model=UserCreate)
def signup(user: UserCreate, db: Session = Depends(config.get_db)):
    try:
        return auth_service.create_user(db, user)
    except ValueError:
        raise HTTPException(status_code=400, detail="Email Already Registered")

회원가입을 처리하는 라우트이다. 입력으로 UserCreate 모델을 받는다.

 

이어서 Depends를 이용해 SQLAlchemy 세션을 주입받고

 

auth_service.create.user를 호출해 사용자를 생성한다.

 

만약 이미 존재하는 이메일이라면 ValueError가 발생해 400의 상태코드와 함께 메시지를 반환한다.

 

이 경우 필요한 JSON 바디는 아래와 같으며,

{
  "email": "example@email.com",
  "password": "your_password",
  "nickname": "your_nickname"
}

서비스 레이어에 이미 로직이 구성되어 있기 때문에 이걸로 끝이다.

@router.post("/login")
def login(user: UserCreate, db: Session = Depends(config.get_db)):
    try:
        db_user = auth_service.authenticate_user(db, user.email, user.password)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid Credentials")

    access_token = auth_service.create_access_token({"sub": user.email})
    return {"access_token": access_token, "email": db_user.email, "nickname": db_user.nickname}

다음으로는 로그인 라우트이다. 마찬가지로 UserCreate 모델을 입력받는다.

이어서 Depends를 이용해 SQLAlchemy 세션을 주입받고 auth_service.authenticate_user를 호출한다.

 

여기서 인증에 실패하면 401 상태코드와 함께 "Invalid Credential 메시지를 반환하고,

 

인증에 성공하면 액세스 토큰을 생성하고 회원 정보와 함께 반환한다.

 

main.py

 

마지막으로 main.py 파일에 막 작성한 라우터를 등록하면 된다.

from fastapi import FastAPI
from sqlalchemy import create_engine
from models.user import Base
from routers import auth as auth_router

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)

app = FastAPI()

app.include_router(auth_router.router, prefix="/auth", tags=["auth"])

@app.get("/")
def read_root():
    return {"message": "Hello, World!"}

앞부분을 보면

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)

와 같이 입력해 서버 시작 시 테이블을 생성하도록 해주었다.

 

이 코드는 한 번만 생성되고, 만약 해당 테이블이 있다면 실행되지 않기 때문에

 

초기화될 염려는 하지 않아도 된다.

app.include_router(auth_router.router, prefix="/api/auth", tags=["auth"])

계속 보면, 여기서 app.include_router는 FastAPI 앱에 라우터를 포함시키는 역할을 한다.

 

이를 통해 특정 라우터 파일에서 정의한 여러 API 엔드포인트를 메인 앱에 등록할 수 있다.

 

이후의 prefix는 auth_router에 정의된 모든 경로 앞에 /auth를 붙인다.

 

마지막으로 tags 옵션은 Swagger UI 같은 문서에 이 라우터의 엔드포인트를 auth 태그로 그룹화하게 해 준다.

 

스프링에서 스웨거를 쓰려면 코드가 지저분해져서 싫었는데, 여기는 굉장히 깔끔하게 끝난다.

 

이렇게 해서 회원가입과 로그인을 진행하면 어렵지 않게 성공하는 것을 확인할 수 있다.

 

글이 매우 길어졌지만, FastAPI 첫 글, 끝!