[FastAPI] CRUD API 구현 (SQLAlchemy, RESTful API)

FastAPI + RESTful API 구현

이전 회사에서 FastAPI로 개발을 하였으나, RESTful 하지 않게 설계도 하였고 정리도 해볼겸 간단한 CRUD API를 RESTful 하게 구현 해보려 한다.

 

▶ RESTful API 에 관련된 내용은 이전 블로그 참고

2025.05.20 - [WEB 개발] - REST, REST API, RESTful API

REST, REST API, RESTful API

---

프로젝트 세팅

pip install uvicorn fastapi

 

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
def hello():
    return {'message': 'Hello World'}

 

서버 실행

• uvicorn 실행 옵션

옵션	설명
--reload	코드 변경 시 자동으로 서버 재시작(개발모드)
--host	서버 호스트 주소 지정 (default: 127.0.0.1)
--port	포트 번호 지정 (default: 8000)
--workers	프로세스 수 지정(멀티 프로세싱, 운영용)

 

uvicorn main:app --reload

 

서버 실행 후 http://127.0.0.1:8000 으로 접속시 'Hello World'가 반환된다

 

---

DB 세팅

DB는 이전에 홈 서버에서 설치 해준 MySQL을 사용할 것이다. 지금은 간단한 기능 구현이라 SQLite를 써도 되나, 추후 비동기 처리도 구현하기 위해 MySQL을 사용한다. (SQLite는 단을 쓰레드로, 락 문제가 있다)

 

 

1. 데이터 베이스 생성

create database fastapi_crud default character set utf8;

 

2. .env 작성

# .env
DB_USER='root'
DB_PASSWD='1234'
DB_HOST=127.0.0.1
DB_PORT=3306
DB_NAME=fastapi_crud

3. 데이터베이스 엔진 설정

• app/database.py

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

from dotenv import load_dotenv
import os

load_dotenv()
user = os.getenv("DB_USER")     # "root"
passwd = os.getenv("DB_PASSWD") # "1234"
host = os.getenv("DB_HOST")     # "127.0.0.1"
port = os.getenv("DB_PORT")     # "3306"
db = os.getenv("DB_NAME")       # "fastapi_crud"

DB_URL = f'mysql+pymysql://{user}:{passwd}@{host}:{port}/{db}?charset=utf8'

engine = create_engine(DB_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False,autoflush=False, bind=engine)
Base = declarative_base()

# Dependency Injection 
def get_db():
    db = SessionLocal()
    try : 
        yield db
    finally:
        db.close()

 

4. 데이터베이스 모델 생성

• app/models.py
• 사용할 테이블의 모델을 정의 함.
• User객체와 Post 객체를 정의함으로써, users, 테이블과, posts 테이블을 생성함.
• User와 Post는 1:N의 관계를 갖음
  • 관계는 relationship 함수로 관리되며, ForeignKey를 사용하는 곳이 다수(N)로 설정됨
  • User의 posts는 자식(Post) 인스턴스를 참조하는 참조 변수를 의미
  • Post의 owner는 부모(User) 인스턴스를 참조하는 참조 변수를 의미
  • owner_id는 외래키로 User의 id를 참조하며, cascade='delete'를 통해 부모 인스턴스가 삭제될 시 자식 인스턴스도 함께 삭제 되도록 정의함.
  • 만약 자식 인스턴스는 남기고 부모 인스턴스만 삭제되어 해당 정보만 지워야 한다면 (owner_id -> None) 자식 객체에서 외래키를 ForeignKey("users.id", ondelete="CASCADE")로 작성하면 됨

from sqlalchemy import Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import mapped_column, relationship
from .database import Base

class User(Base):
    __tablename__ = "users"
    id = mapped_column(Integer, primary_key=True, autoincrement=True)
    name = mapped_column(String(255), nullable=False)
    email = mapped_column(String(255), unique=True, nullable=False)
    posts = relationship("Post",back_populates="owner", cascade='delete')
    hashed_pw = mapped_column(String(255), nullable=False)
    is_active = mapped_column(Boolean,default=False)

class Post(Base):
    __tablename__ = "posts"
    id = mapped_column(Integer, primary_key=True, autoincrement=True)
    title = mapped_column(String(255), nullable=False)
    description = mapped_column(String(255))
    owner_id = mapped_column(Integer, ForeignKey("users.id"))
    owner = relationship("User",back_populates="posts")

 

5. DB 초기화

• app/init_db.py
• 프로젝트에 정의한 모델을 기반으로 실제 데이터베이스에 테이블을 생성하려면 SQLAlchemy의 create_all() 메서드를 사용
• FastAPI 애플리케이션 실행과는 별도로 한 번만 실행

from .database import engine
from . import models

def create_tables():
    print("Creating tables...")
    models.Base.metadata.create_all(bind=engine)

if __name__ == "__main__":
    create_tables()

 

실행

python -m app.init_db

 

---

FastAPI 코드 작성

필요 패키지 설치

pip install fastapi uvicorn sqlalchemy python-dotenv passlib[bcrypt] python-jose pymysql pip install pydantic[email]

1. Pydantic 스키마

• app.schena.py
• 데이터 유효성 검증, 데이터 변환을 할 수 있도록 함

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class PostBase(BaseModel):
    title : str
    description : Optional[str] = None

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id : int
    owner_id  : int

    class Config:
        from_attributes = True

class UserBase(BaseModel):
    name: str = Field(..., example="홍길동")
    email: EmailStr = Field(..., title="이메일", description="유효한 이메일 형식", example="hong@naver.com")

class UserCreate(UserBase):
    password: str = Field(..., title="비밀번호", description="8~128자, 공백 없음", min_length=8, max_length=128, example="securePass123!")

class User(UserBase):
    id : int
    is_active : bool

    class Config:
        from_attributes = True

class LoginRequest(BaseModel):
    email: EmailStr = Field(..., example="user@example.com")
    password: str = Field(..., min_length=8, example="securepass123")

 

2. CRUD 함수 생성

• app/crud.py
• create, read, update, delete 관련 함수 정의

from sqlalchemy.orm import Session
from . import models, schema

############################ USER ############################
def get_users(db: Session, skip:int=0, limit:int=50):
    '''
    모든 사용자 정보 조회(페이징 처리)
    '''
    return db.query(models.User).offset(skip).limit(limit).all()

def get_user(db: Session, user_id: int):
    '''
    특정 사용자 조회
    '''
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    '''
    회원가입시 동일 이메일 존재 여부를 위한 조회
    '''
    return db.query(models.User).filter(models.User.email == email).first()


def create_user(db: Session, user:schema.UserCreate):
    '''
    신규 사용자 추가
    '''
    db_user = models.User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user: models.User, updated_user: schema.UserCreate):
    '''
    사용자 정보 수정
    '''
    for key, value in updated_user.model_dump().items():
        setattr(user, key, value)
    db.commit()
    db.refresh(user)
    return user

def delete_user(db: Session, user: models.User):
    '''
    사용자 제거
    '''
    db.delete(user)
    db.commit()

############################ POST ############################
def get_posts(db: Session, skip:int=0, limit: int=50):
    '''
    모든 게시물 조회(페이징 처리)
    '''
    return db.query(models.Post).offset(skip).limit(limit).all()

def get_post(db: Session, post_id: int):
    '''
    특정 게시물 조회
    '''
    return db.query(models.Post).filter(models.Post.id == post_id).first()


def create_user_post(db:Session, post:schema.PostCreate, user_id : int):
    '''
    특정 사용자의 게시물 생성
    '''
    db_post = models.Post(**post.model_dump(), owner_id=user_id )
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def update_post(db: Session, post: models.Post, updated_post: schema.PostCreate):
    '''
    게시물 수정
    '''
    for key, value in updated_post.model_dump().items():
        setattr(post, key, value)
    db.commit()
    db.refresh(post)
    return post

def delete_post(db: Session, post: models.Post):
    '''
    게시물 삭제
    '''
    db.delete(post)
    db.commit()

 

• create_user 함수 수행시 비밀번호 해시화
• app/utils/security.py

from passlib.context import CryptContext

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

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

3. main.py 수정

• app/main.py로 이동

from fastapi import FastAPI
from .routers import user, post

app = FastAPI()

# 라우터 등록
app.include_router(user.router)
app.include_router(post.router)

 

4. route 생성

• app/routers/user.py
• 사용자 정보 관련 API

from fastapi import Depends, HTTPException, APIRouter
from sqlalchemy.orm import Session

from .. import crud, schema
from ..database import get_db

router = APIRouter(prefix="/users", tags=["users"])

@router.get("", response_model=list[schema.User],summary="모든 사용자 정보 조회")
def get_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

@router.get("/{user_id}", response_model=schema.User, summary="특정 사용자 정보 조회")
def get_user(user_id: int, db: Session = Depends(get_db)):
    db_user =  crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@router.post("", response_model=schema.User, summary="회원가입")
def post_user(user: schema.UserCreate, db: Session = Depends(get_db)):
    db_user =  crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db, user=user)

@router.put("/{user_id}", response_model=schema.User, summary="기존 사용자 정보 수정")
def update_user(user_id: int, updated_user: schema.UserCreate, db: Session = Depends(get_db)):
    db_user =  crud.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return crud.update_user(db, db_user, updated_user)

@router.delete("/{user_id}", summary="사용자 삭제")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user =  crud.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    crud.delete_user(db, db_user)
    return {"message": "User deleted successfully"}

 

• app/routers/post.py
• 게시물 관련 API

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from .. import crud, schema
from ..database import get_db

router = APIRouter(prefix="/post", tags=["post"])

@router.get("", response_model=list[schema.Post], summary="모든 게시글 목록 조회")
def get_posts(skip: int = 0, limit: int = 50, db: Session = Depends(get_db)):
    return crud.get_posts(db, skip=skip, limit=limit)


@router.post("/{user_id}", response_model=schema.Post, summary="특정 사용자의 게시글 생성")
def post_post_for_user(user_id: int, post: schema.PostCreate, db: Session = Depends(get_db)):
    return crud.create_user_post(db=db, user_id=user_id, post=post)

@router.put("/{post_id}", response_model=schema.Post, summary="기존 게시글 수정")
def update_post(post_id: int, updated_post: schema.PostCreate, db: Session = Depends(get_db)):
    db_post = crud.get_post(db, post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    return crud.update_post(db, db_post, updated_post)

@router.delete("/{post_id}", summary="게시글 삭제")
def delete_post(post_id: int, db: Session = Depends(get_db)):
    db_post = crud.get_post(db, post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    crud.delete_post(db, db_post)
    return {"message": "Post deleted successfully"}

 

---

FastAPI 실행

uvicorn app.main:app --reload

 

API 서버 실행 후 http://127.0.0.1:8000/docs 접속

 

회원가입 테스트

• 이메일 조건 불충족
• 패스워드 조건 불충족

 

결과

• 422 code 리턴 및 email, password 조건 불충족 에러 메시지 반환

 

만족 조건으로 테스트

 

결과

• 200 코드 반환 및 추가된 정보 반환