[FastAPI] AWS DB - MySQL 연결하기

https://fastapi.tiangolo.com/tutorial/sql-databases/#create-the-relationships

SQL (Relational) Databases - FastAPI

처음에는 하나의 DB (MongoDB)를 사용하여 작업을 하고자 구상했지만

연결된 데이터 관리와 관계형 데이터베이스 사용 경험을 위해 AWS에서 DB를 하나 생성했다.

가장 먼저, SQL 사용을 위한 패키지를 설치한다.

python3 -m pip install sqlalchemy

초기 설정은 가이드 문서 그대로 따라하되, 연결될 DB는  SQLite가 아닌 AWS의 MySQL이므로 연결 정보 수정이 필요했다.

이 과정에서 pymysql 이라는 패키지도 추가로 설치해야 한다.

python3 -m pip install pymysql

해당 패키지가 없으면 "mysql + pymysql" 부분에서 에러가 발생한다.

 

1. database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv
import os

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
load_dotenv(os.path.join(BASE_DIR, ".env"))

SQL_ADDRESS = f'mysql+pymysql://{os.environ["sql_db_user"]}:{os.environ["sql_db_password"]}@{os.environ["sql_db_address"]}:{os.environ["sql_db_port"]}/{os.environ["sql_db_connect"]}'

engine = create_engine(
    SQL_ADDRESS
)

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

Base = declarative_base()

 

가이드에서 설명하는 디렉토리 구조는 아래와 같다.

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

위의 구조에서 database.py 는 위의 데이터 베이스가 연결되고, 실행하는 정보를 저장하는 역할을 하고

schemas.py는 클라이언트로부터 데이터를 받기 위해 필요한 정보를 저장하는 역할을 담당한다.

models.py는 DB에서 테이블에 대한 정의와 관련된 정보를 저장하고

마지막으로 crud.py는 서버가 DB에 접근해서 데이터를 입출력하는 방법을 정의한다.

database.py는 위에서 살펴봤으니, 다음으로는 models.py를 정의하자.

 

2. models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

from .database import Base

class Board(Base):
    __tablename__ = "Boards"
    
    id = Column(Integer, primary_key = True, index = True)
    writer = Column(String(255))
    title = Column(String(255))
    content = Column(String(255))
    like = Column(Integer, default=0)
    created_datetime = Column(DateTime, default = func.now())
    edited_datetime = Column(DateTime, default = func.now())
    comment = relationship("Comment", back_populates= "owner")
    
class Comment(Base):
    __tablename__ = "Comments"
    
    id = Column(Integer, primary_key = True, index = True)
    writer = Column(String(255))
    content = Column(String(255))
    created_datetime = Column(String(255))
    edited_datetime = Column(String(255))
    
    board_id = Column(Integer, ForeignKey("Boards.id"))
    owner = relationship("Board", back_populates= "comment")

DB를 이용해서 게시글을 입력하고, 게시글마다 답을을 입력하기 위해 위와 같이 코드를 작성했다.

입력받기 위한 데이터의 형태와 범위(한도)를 지정, 기본값이 필요할때는 default 를 사용한다.

create_datetime / edited_datatime 필드 참고

Board 테이블은 Comment와 연관을 가질수 있도록 relationship을 사용한다.

하나의 Board는 여러개의 Comment를 가질 수 있으므로 back_populates 값을 owner로 지정했고

Comment는 Board와 관련이 있다는 내용을 입력한다.

 

3. schemas.py

from pydantic import BaseModel
from typing import Union
from datetime import datetime

class Create_Board(BaseModel):
    writer : str
    title : str
    content : str

class Edit_Board(Create_Board):
    edited_datetime : Union[str, None] = datetime.now()

클라이언트로부터 데이터를 받기 위한 데이터를 명시하자.

Create_Baord는 글 작성 시 3가지의 데이터를 요구하도록 구성하였고,

Edit_Board 는 Create_Board를 상속받아서 같은 3가지 정보 외 수정된 시간을 받아올 수 있도록 구성했다.

 

4. crud.py

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

def create_board(db: Session, board: schemas.Create_Board):
    board_db_create = models.Board(writer = board.writer,
                                   title = board.title,
                                   content = board.content)
    
    db.add(board_db_create)
    db.commit()
    db.refresh(board_db_create)
    return board_db_create

클라이언트로부터 데이터를 받아 각 정보를 넣어주는 역할, 항목별로 데이터를 받아서 데이터를 만들고 저장한다.

이 때 없는 데이터들 ( 생성, 수정시간)은 저장 당시 시간으로 저장된다.

 

5. routers/boards.py

from fastapi import Depends, HTTPException, APIRouter
from sqlalchemy.orm import Session
from sql_app import crud, models, schemas
from sql_app.database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)
router = APIRouter(prefix="/api/board", tags=["boards"])

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

@router.post("/create")
def create_board(board: schemas.Create_Board, db: Session = Depends(get_db)):
    
    return crud.create_board(db = db, board = board)

위 경로로 요청이 들어오면 데이터를 생성해주고 반환한다.