[FastAPI] (7) MySQL 연결 및 API Test

 

MySQL 설치 및 DB 생성

1-1. MySql 설치 (로컬 pc에 직접 설치)

https://www.mysql.com/downloads/

MySQL :: MySQL Downloads

 

1-2. Docker MySQL 설치

docker run -e MYSQL_ROOT_PASSWORD=<YOUR PASSWORD> -d mysql:8.0 --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci

 

2. mysql workbench 접속해서 Create Schema 

mysql workbench 설치 완료시, root 계정으로 접속해서 schema (=database) 를 생성

 

 

FastAPI 코드 작성

파일들을 나눠서 작성하는 것이 훨씬 좋지만, 한눈에 보기 위해 예제에서는 mysql_ex.py 라는 이름으로 한번에 모아서 작성하였다. 아래 나눠둔 파일 구분으로 따로 작성하여도 된다.

 

3-1. database, model 작성

database : DB 연결 Connection 담당

models : 정의된 DB 모델을 뜻한다.

from typing import List
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import Boolean, Column, Integer, String
from pydantic import BaseModel

# database.py
engine = create_engine("mysql+pymysql://root:rlatjsals1!@localhost:3306/my_db?charset=utf8mb4")
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
)

# models.py
Base = declarative_base()
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, index=True)
    password = Column(String(255))
    is_active = Column(Boolean, default=True)

Base.metadata.create_all(bind=engine) # Model 정보를 DB에 생성

 

3-2. dto 작성

dto : 실제 FastAPI CRUD Method를 위한 dto (data transfer object) 정의.

# dto.py - pydantic
class UserBase(BaseModel):
    email: str

class UserOut(UserBase):
    id: int
    email: str
    is_active: bool

    class Config:
        orm_mode = True

class UserCreate(UserBase):
    password: str

 

3-3. main 작성

main : 프로그램 진입문 및 API 정의

# main.py
app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
        
@app.get("/users", response_model=List[UserOut])
def read_users(db: Session = Depends(get_db)):
    return db.query(User).all()

@app.post("/users", response_model=UserOut)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    user = User(email=user.email, password=user.password)
    db.add(user)
    db.commit()
    return user

 

API Test

다음과 같은 Swagger 화면이 나오고 API를 TEST 를 해볼 수 있다.

 

http://127.0.0.1:8000/docs

 

curl 기반 POST 테스트 => 200 응답

curl -X 'POST' \
  'http://127.0.0.1:8000/users' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "email": "yubi5050@naver.com",
  "password": "rlatjsals1!"
}'

 

WorkBench에서 Query문을 돌려보면 정상적으로 생성 된 것을 확인 할 수 있다.

workbench POST 데이터 생성 결과