FastAPI에서 MySQL 연결

FastAPI에선 SQLAlchemy ORM 라이브러리로 DB를 조작할 수 있다. 먼저 MySQL과 연결하기 위해 sqlalchemy를 설치한다.

pip install sqlalchemy

 

✅ 설정 파일 추가하기

FastAPI에 ORM을 적용하려면 데이터베이스 설정이 필요하다. database.py 파일을 생성하고 아래와 같이 코드를 작성한다.

# app/database/conn.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from common.config import Settings

SQLALCHEMY_DATABASE_URL = Settings.DATABASE_URL
engine = create_engine(SQLALCHEMY_DATABASE_URL)

Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

SQLALCHEMY_DATABASE_URL는 데이터베이스 주소이다. DB주소를 깃헙에 노출시키지 않기 위해 common.config에 위치한 Settings 클래스에 정보를 숨겨놨다. Session은 데이터베이스에 접속하기 위해 필요한 클래스이다. create_engine, sessionmaker 등을 사용하는것은 SQLAlchemy 데이터베이스를 사용하기 위해 따라야 할 규칙이다.

 

sessionmaker에 autocommit=False는 commit이라는 사인을 주어야만 DB에 저장이 된다. True로 설정하면 commit 사인이 없어도 데이터베이스에 변경사항이 저장된다.

 

create_engine은 커넥션 풀을 생성한다. 컨넥션 풀이란 데이터베이스에 접속하는 객체를 일정 갯수만큼 만들어 놓고 돌려가며 사용하는 것을 말한다. 그리고 declarative_base 함수에 의해 반환된 Base 클래스는 향후 모델을 정의할 때 사용되는 클래스이다.

# app/conmmon/config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
	
    DB_USERNAME : str = os.getenv("USERNAME")
    DB_PASSWORD = os.getenv("PASSWORD")
    DB_HOST : str = os.getenv("HOST","localhost")
    DB_PORT : str = os.getenv("PORT",3306)
    DB_DATABASE : str = os.getenv("DATABASE")
	
    DATABASE_URL = f"mysql+pymysql://{DB_USERNAME}@{DB_HOST}:{DB_PORT}/{DB_DATABASE}"

settings = Settings()

DB에 대한 정보는 load_dotenv()를 이용해서 .env 파일에 숨겨두었다. 아래와 같이 .env파일에 정보를 넣어두면 key값으로 value값을 가져올 수 있다.

USERNAME="root"
HOST="localhost"
PORT="3306"
DATABASE="db"

 

✅ 모델 생성하기

먼저 데이터베이스에 데이터를 넣기 전에 모델을 정의해야 한다. Users와 같은 모델 클래스는 서두에 언급했던 Base 클래스를 상속하여 만들어야 한다. __tablename__은 모델에 의해 관리되는 테이블 이름을 뜻한다.

# app/database/model.py
from sqlalchemy import (
    Column,
    Integer,
    String,
    DateTime,
    func,
    Enum,
    Boolean,
    ForeignKey,
)
from database.conn import Base

class Users(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    status = Column(Enum("active", "deleted", "blocked"), default="active")
    email = Column(String(length=255), nullable=True)
    pw = Column(String(length=2000), nullable=True)
    name = Column(String(length=255), nullable=True)
    phone_number = Column(String(length=20), nullable=True, unique=True)
    profile_img = Column(String(length=1000), nullable=True)
    sns_type = Column(Enum("FB", "G", "K"), nullable=True)
    marketing_agree = Column(Boolean, nullable=True, default=True)

 

✅ 모델을 이용해 테이블을 자동으로 생성하기

 

모델을 구상하고 생성했으므로 SQLAlchemy의 alembic을 이용해 데이터베이스 테이블을 생성해 보자.

alembic은 SQLAlchemy로 작성한 모델을 기반으로 데이터베이스를 쉽게 관리할 수 있게 도와주는 도구이다. 예를들어 models.py 파일에 작성한 모델을 이용하여 테이블을 생성하고 변경할수 있다. 아래와 같이 alembic을 설치한다.

pip install alembic

설치 후 alembic 초기화 작업을 위해 init migrations를 해준다.

alembic init migrations

그러면 디렉토리 하위에 migrations와 alembic.ini 파일이 생성된다. alembic.ini 파일은 alembic의 환경설정 파일이다.

alembic.ini 파일을 열어서 DB url을 입력해준다.

sqlalchemy.url = mysql+pymysql://root@localhost:3306/db

그리고 migrations 디렉토리의 env.py도 아래과 같이 수정한다.

from database import models

target_metadata = models.Base.metadata

그리고 터미널 창에서 alembic revision --autogenerate를 입력해준다. 이어서 alembic upgrade head를 실행한다. 이렇게 할 경우 모델에 정의한 테이블이 생성된다.

 

✅ alembic 없이 테이블 생성

alembic없이도 테이블을 생성할 순 있다. main.py 파일에 다음의 문장을 삽입하면 FastAPI 실행시 필요한 테이블들이 모두 생성된다. 하지만 이렇게 할 경우 생성된 테이블에 대한 변경 관리를 할 수 없다는 단점이 존재한다.

from database import models
from database.models import Users
from database.conn import engine

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

 

MySQL에서 확인해보면 우리가 정의했던 모델에 맞춰 users테이블이 생성된 걸 볼 수 있다.

 

✅ 데이터 삽입

데이터가 잘 들어가는지 확인하기 위해 테스트 코드를 작성해본다.

from fastapi import Depends, FastAPI
from database.conn import Session
from database import models
from database.models import Users
from database.conn import engine

app = FastAPI()
session = Session()

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

@app.get("/")
def index():

    example = session.query(Users).all()

    return example

@app.post("/create")
def create_user():
    user = Users(id=1, name="haki_test")
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

swagger를 이용해서 create url로 POST요청을 보내보면 아래와 같이 응답이 온다.

생성한 데이터를 확인하기 위해 index url로 get요청을 보내게 되면 아래와 같이 모든 User에 대한 데이터 응답이 온다.