Settings

docker run --name cj-mysql \
	-e MYSQL_ROOT_PASSWORD=1234 \ 
    -d mysql:8.0 \
    --character-set-server=utf8mb4 \
    --collation-server=utf8mb4_unicode_ci

pip install PyMySQL

• python과 mysql 연결을 위한 드라이버 설치

App Code

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


engine = create_engine("mysql+pymysql://admin:1234@0.0.0.0:3306/dev")
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
)

Base = declarative_base()

• database.py
• create_engine
  • pymysql을 사용하므로 프로토콜은 저 위처럼 지정해줘야한다.
  • user 이름이 admin, paswword까 1234다.
• Base는 models.py에서 사용

from sqlalchemy import Boolean, Column, Integer, String

from .database import 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)

• models.py
• database의 Base를 상속

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


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

    class Config:
        orm_mode = True

• schemas.py
• sqlalchemy와 FASTAPI의 pydantic Model이 겹치므로 FastAPI에서는 schemas.py로 정의하도록 권장하고 있다.

from typing import List

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

from . import models, schemas
from .database import SessionLocal, engine

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


app = FastAPI()


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


@app.post("/users", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    existed_user = db.query(models.User).filter_by(email=user.email).first()

    if existed_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(email=user.email, password=user.password)
    db.add(user)
    db.commit()

    return user


@app.get("/users", response_model=List[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return db.query(models.User).all()

• main.py
• models.Base.metadata.create_all(bind=engine)
  • models에서 정의한 내용을 생성하겠다는 의미다.

pip install sqlalchemy
pip install cryptography

• 위를 설치해서 유저를 생성하는 API, 얻는 API를 테스트해본다.

uvicorn app.main:app --host 0.0.0.0 --port 8000 --realod

• 실행 !

• 테이블 생성이 잘되는지 확인해보자.

docker exec -it cj-db mysql -uadmin -p

• uadmin에서 admin은 id
• passward를 치면 들어가진다.

• show databases 명령어를 통해 데이터베이스가 생성되었음을 확인할 수 있다.
• 위에서 create_all() 함수를 통해 생성을 해준 것이다.

use dev -> dev를 database를 선택한 것이다.
show tables; -> table을 확인해본다.

• 쿼리해보면 역시 아무 것도 없다 ! 유저를 만들어주지 않았기 때문 !

• docs에서 create API를 이용해보자.

• 아주 달달하게 등록되었음을 DB에서도 확인할 수 있었다.

• user 조회 API를 사용해봤을 때에도 정상 작동한다.

@app.get("/users", response_model=List[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return db.query(models.User).all()

• 해당 API가 작동하는 함수인데, 어떻게 sqlalchemy orm임에도 불구하고 pydantic 로직의 모습을 따올 수 있는 것일까 ?

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


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

    class Config:
        orm_mode = True

• 이는 schemas.py 에서 정의한 pydantic 모델에서 orm_mode를 True로 해주었기 때문이다.
  • 별도의 변환작업 없이 위 기능을 사용하면 강력하게 변환을 줘서 스키마를 적용할 수 있게된다.