[Python] FastAPI와 MySQL을 사용한 CRUD API 만들기

1. 개발 환경 준비

1) FastAPI 설치

• 이전 글을 참고해서 FastAPI를 설치

[FastAPI] FastAPI 설치 및 간단한 웹 API 만들기

 

2) MySQL 설치

• 이전 글을 참고해서 MySQL을 설치

[MySQL] MySQL 설치하기 (윈도우 / windows)

 

3) 패키지 설치

• SQLAlchemy: Python SQL toolkit이자 객체 관계형 매퍼(ORM)이다. SQLAlchemy는 테이블의 데이터를 객체와 연결시켜 주는 역할을 한다. (sqlalchemy-2.0.31)
• PyMySQL: Python에서 MySQL에 접하여 데이터베이스를 조작할 수 있도록 하는 라이브러리다. (pymysql-1.1.1)

pip install sqlalchemy 
pip install pymysql

• 추가적으로 환경 변수를 .env 파일에서 별도 관리할 수 있는 dotenv 라이브러리를 사용할 수 있다.

pip install python-dotenv

2. CRUD API 만들기

• 프로젝트 구조

 

1) 데이터베이스 엔진 생성

• 연결할 데이터베이스를 미리 생성한다.

create database test_db default character set utf8;

 

• 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")       # "mydb"

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()

• database.py는 데이터베이스 연결을 관리하는 파일이다.
• load_dotenv 함수를 통해 .env 파일에 있는 환경변수들을 가져올 수 있다.
  • dotenv 설치는 pip install python-dotenv 를 사용한다.
  • 이때 load_dotenv() 함수를 선언한 뒤,  os.getenv("변수명")로 변수를 가져온다.
  • 프로젝트 루트 디렉토리에 .env 파일을 생성하고 환경 변수를 DB_USER = root 형식으로 저장한다.
  • 환경변수가 변경된다면 load_dotenv(override=true) 로 설정해야 바뀐 변수가 적용된다.
  • 혹은 config.py 파일을 생성해 환경변수를 관리하는 방법이 있다. 
• DB_URL은 자신이 사용하는 데이터베이스에 맞는 주소를 작성한다.
  • MySQL : f'mysql+pymysql://{user}:{passwd}@{host}:{port}/{db}?charset=utf8'
  • Postgresql : f'postgresql://{user}:{passwd}@{host}:{port}/{db}'
  • SQLite : f'sqlite://{/dbname.db}'
    • SQLite는 create_engine(DB_URL, echo=True, connect_args={"check_same_thread":False})를 사용한다. SQLite는 한 번에 하나의 thread만 접근할 수 있기 때문이다.
• 필요에 따라 echo는 켜거나 끌 수 있다. sql 동작을 보고 싶다면 True 필요 없다면 지우면 된다.

 

2) 데이터베이스 모델 생성

• app/models.py를 만들어 사용할 테이블의 모델을 정의한다.

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')
    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")

• 예제는 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")로 작성하면 된다.
  • 만약 1:1 관계로 설정한다면, 부모 객체의 relationship 함수의 매개변수에 uselist=False를 추가한다.
• mapped_column 함수에서는 데이터 타입, primary_key, nullable, unique, default 등을 설정할 수 있다.
• Declarative 버전으로 작성한다면 아래와 같이 작성할 수 있다.

from sqlalchemy.orm import Mapped
from typing import List

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(unique=True)
    posts: Mapped[List["Post"]] = relationship(back_populates="owner")
    is_active: Mapped[bool] = mapped_column(Boolean,default=False)

• Tip! 대용량의 데이터를 다룰 때, User 테이블과 Post 테이블을 join하지 않고, 사용자 id를 통해 게시물을 쿼리하여 가져오기 때문에 속도가 느려질 수 있다. 이 때 join을 강제로 시켜주는 방법을 사용해 속도 개선이 가능하다.
  relationship("Post", lazy="joined")

 

3) Pydantic 스키마 생성

• app/schema.py를 생성하여 데이터 유효성 검증, 데이터 변환을 할 수 있도록 한다.

from pydantic import BaseModel
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:
        orm_mode = True

class UserBase(BaseModel):
    name: str
    email: str

class UserCreate(UserBase):
    pass 

class User(UserBase):
    id : int
    is_active : bool
    posts : list[Post] = []

    class Config:
        orm_model = True

• Pydantic 스키마는 데이터 유효성 검증과 함께 데이터 입력과 조회 시 필요한 필드만 선택할 수 있게 한다.
• Pydantic은 validation error를 일으켜 어떤 데이터 입력 형식이 잘못되었는지 확인 가능하다.
• 예제에서는 users 테이블에 데이터를 입력할 때, name과 email 이외의 필드는 제외하고 입력할 수 있도록 UserBase를 구성되었다.
• 또한 User 객체를 통해 응답 데이터의 범위를 조절할 수 있다. 즉 응답에서 제외하고 싶다면 스키마만 조정해도 된다. 예제에서는 User 객체를 통해서 id, name, email, is_active, posts 정보를 모두 확인할 수 있다. is_active를 제외하면 응답 데이터에 is_active가 노출되지 않는다.
• class Config: orm_model = True 는 모델의 항목들을 자동으로 스키마에 매핑해주는 역할을 한다.
• Django의 serializer.py와 유사하다.

 

4) 데이터베이스 CRUD 함수 생성

• app/crud.py에는 create, read(get), update, delete 관련 함수들을 정의한다.

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

############################ USER ############################
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 get_users(db: Session, skip:int=0, limit:int=50):
    return db.query(models.User).offset(skip).limit(limit).all()

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_post(db: Session, post_id: int):
    return db.query(models.Post).filter(models.Post.id == post_id).first()

def get_posts(db: Session, skip:int=0, limit: int=50):
    return db.query(models.Post).offset(skip).limit(limit).all()

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()

• add: 인스턴스 객체를 데이터베이스 세션에 추가한다.
• commit: 데이터베이스에 변경 사항을 커밋한다.
• refresh: 인스턴스를 새로고침한다. (함께 입력하지 않아 새롭게 생성된 id와 같은 새 데이터를 포함하도록)
• sqlalchemy 2.0 버전에서도 query 함수를 사용할 수 있지만, 2.0 버전부터 지원 가능한 비동기 방식으로 개발하려면 execute와 select, update, delete 함수 등을 사용해야 한다.

[Python]

from sqlalchemy import select, update, delete
select_stmt = select(User).where(User.name == "spongebob")
update_stmt = (
    update(user_table)
    .where(user_table.c.name == "patrick")
    .values(fullname="Patrick the Star")
)
delete_stmt = (
    delete(user_table)
    .where(user_table.c.name == "patrick")
    .returning(user_table.c.id, user_table.c.name)
)
result = session.execute(stmt)

 

[SQL]

SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?

UPDATE user_account SET fullname=:fullname 
WHERE user_account.name = :name_1

DELETE FROM user_account
WHERE user_account.name = :name_1
RETURNING user_account.id, user_account.name

 

5) FastAPI의 main.py 생성

• app/main.py를 생성하여 FastAPI app을 생성하고, CRUD를 담당할 API의 Endpoint를 정의한다.

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

from . import crud,models, schema
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

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

####################### USER #######################
@app.get("/users/", response_model=list[schema.User])
def get_users(skip:int=0, limit:int=0, db:Session=Depends(get_db)):
    users = crud.get_users(db,skip=skip,limit=limit)
    return users

@app.get("/users/{user_id}/",response_model=schema.User)
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

@app.post("/users/",response_model=schema.User)
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=db,user=user)

@app.put("/users/{user_id}/",response_model=schema.User)
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")
    updated_user = crud.update_user(db, db_user, updated_user)
    return updated_user

@app.delete("/users/{user_id}/")
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"}

####################### POST #######################
@app.get("/posts/", response_model=list[schema.Post])
def get_posts(skip:int=0,limit:int=0,db:Session=Depends(get_db)):
    posts = crud.get_posts(db,skip=skip,limit=limit)
    return posts

@app.post("/users/{user_id}/posts/",response_model=schema.Post)
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)

@app.put("/posts/{post_id}/",response_model=schema.Post)
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")
    updated_post = crud.update_post(db, db_post, updated_post)
    return updated_post

@app.delete("/posts/{post_id}/")
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"}

• Dependency Injection(의존성 주입)은 반복되는 작업을 간편하게 사용할 수 있도록 하는 것을 말한다.
• 여기서는 의존성을 부여하 반복적으로 db의 Session을 선언하고 close하는 작업을 도와준다.
• @contextlib.contextmanager 이란 어노테이션을 get_db 함수에 붙이고 with get_db() as db: 와 사용해야 한다.

import contextlib
@contextlib.contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
@app.get("/users/", response_model=list[schema.User])
def get_users(skip:int=0, limit:int=0):
    with get_db() as db:
    	users = crud.get_users(db,skip=skip,limit=limit)
    return users

• 더 간단하게는 fastapi의 Depends를 통해 Session = Depends(get_db) 의 형태로 바로 사용 가능하다.

from fastapi import Depends

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

@app.get("/users/", response_model=list[schema.User])
def get_users(skip:int=0, limit:int=0, db:Session=Depends(get_db)):
    users = crud.get_users(db,skip=skip,limit=limit)
    return users

 

6) FastAPI 서버 실행

uvicorn app.main:app --reload

• app/main.py를 실행하기 위해 app.main으로 작성해서 서버를 실행한다.
• 서버가 실행된 뒤, http://127.0.0.1:8000/docs#/ 를 웹페이지에서 열면 FastAPI 문서를 볼 수 있다.

 

참고문헌

• https://medium.com/@iambkpl/setup-fastapi-and-sqlalchemy-mysql-986419dbffeb
• https://docs.sqlalchemy.org/en/20/