FastAPI에서 ORM을 통한 객체 테이블 만들기
순서
1. DB 생성하기
2. DB 연동하기
3. 테이블 만들기
4. 테이블에 데이터 입력
5. DB 확인
데이터베이스에 직접 접근하여 스키마를 생성하고, 쿼리문을 통해 테이블을 생성하지 않고 프로젝트 파일 내에서 객체형태로 테이블의 컬럼과 제약조건을 설정하고 DB를 간접적으로 관리하는 방법이 있다.
이것을 ORM(Object Relational Mapping)이라고 하는데 FastAPI 공식문서에서는 "sqlalchemy"를 통해 ORM을 설정하는 법을 안내하고 있다.
ORM없이는 쿼리문을 통해 DB를 관리해야 되니까 쿼리문을 알아야할 뿐더러 쿼리문과 코딩을 번갈아가면서 사용해야한다.
또한 DB는 공통적인 쿼리문이 있지만 미세하게 DB사이의 쿼리문 차이가 존재한다.
DB만 연결한다면 이런 고민을 제거하고 프로젝트 내에서 기능구현에만 집중 할 수 있다.
물론 DB 관리로 부터 완전한 분리가 되진않는다. 테이블간 관계가 복잡해질 경우 ORM 설계 난이도 또한 올라가고, 성능 개선을 위해 결국 쿼리문을 사용하여 코드를 구현해야 할 수도 있다.
나중에 회원가입, 로그인 기능을 위한 User 테이블을 만들고, 사용자들이 글을 작성하는 기능을 위한 Post 테이블을 만들어 보자.
파일트리는 DB에 접속하는 설정이 있는 config.py, 요청을 주고 받고 데이터를 생성하는 main.py, 테이블을 정의하는 model.py, DB에 데이터 저장 요청시 저장 전 올바른 데이터 저장인지 유효성을 평가하는 validation,py로 이루어져 있다.
DB는 다른 것을 사용해도 무관하나 여기서는 Mysql을 사용한다.
일단 Workbench를 통해서 관리자를 생성하고 스키마를 생성해야 아래 코드를 따라할 수 있다.
관리자명: adimin
비밀번호: 1234
DB 경로: localhost
DB 포트: 3306
DB 이름: fast-orm
MySQL과 FastAPI를 연동
(sqlalchemy 추가 설치 필요)
#config.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
URL_DATABASE = 'mysql+pymysql://admin:1234@localhost:3306/fast-orm'
engine = create_engine(URL_DATABASE, echo=True)
SessionLocal = sessionmaker(autocommit = False, autoflush= False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()URL_DATABASE : DB종류, DB의 경로에 DB 관리자 비밀번호 포트, DB명을 명시해준다.
engine: DB 사용을 위한 engine 객체를 생성한다. echo=True를 추가 할 경우 쿼리문 요청시 터미널에 출력해준다.
SessionLocal: DB를 사용하는 세션생성
세션이란 뭘까?
DB 사용 시작과 끝을 묶은 시점 단위라고 생각하면 될 것 같다.
한 세션단위로 DB를 관리해야 데이터 문제가 발생하였을 때 문제 발생 시점이 명확해지고 롤백을 통해 되돌릴 수 있다.
autocommit은 commit()함수를 자동적으로 실행시켜주나 문제가 발생할 경우 rollback 할 수 없다.
autoflush는 flush()을 자동적으로 실행하여 트렌젝션을 DB에 전송한다.
Base: declarative_base() 함수를 사용하여 모델을 생성하기 위한 클래스를 정의한다.
model.py에서는 테이블을 정의하고 자료형이나, 제약조건을 설정한다.
#model.py
from sqlalchemy import Column, String, Integer
from config import Base
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, index=True)
user_id = Column(String(50), unique=True)
user_name = Column(String(50))
password = Column(String(100))유저 데이터 생성시 id가 자동생성되고 기본키로 설정되도록 했다.
원래 따로 명시하지 않더라도 자동적용된다.
또한 다른 필드를 기본키로 적용하고 싶다면 primary_key=True를 다른 필드에 적어주면 된다.
그리고 user_id는 중복이 되지 않도록 unique=True(고유키)로 설정하였다.
validation.py에서는 데이터 생성 할때 타입 검증을 하기 위한 코드이다.
공식문서에서는 pydantic을 사용한 방법을 소개하고 있다.
#validation.py
from pydantic import BaseModel
class UserBase(BaseModel):
user_id: str
user_name: str
password: strmain.py에서는 주소별 요청을 처리한다.
/user/sign-up에서 user_id, user_name, password를 post요청 보내고 정상적으로 처리되면 status code가 201로 반환되도록 하였다.
#main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import Annotated
# config.py에서 Base와 engine 가져오기
from config import get_db, Base, engine
import model
import validation
app = FastAPI()
# 테이블을 삭제한 후 다시 생성하는 코드
Base.metadata.drop_all(bind=engine) # 모든 테이블 삭제
Base.metadata.create_all(bind=engine) # 테이블 다시 생성
db_dependency = Annotated[Session, Depends(get_db)]
@app.get("/")
def main():
return {"Hello FastAPI!"}
#유저 생성
@app.post("/user/sign-up", status_code=status.HTTP_201_CREATED)
async def create_user(user: validation.UserBase, db:db_dependency):
db_user = model.User(**user.dict())
db.add(db_user)
db.commit()http://localhost:8000/docs에 접속하여 Swagger UI를 통해 유저 생성 테스트를 하고 workbench에서 select * from user 쿼리를 통해 데이터가 저장되었는지 확인 할 수 있다.
이제 나머지 DB 조회 수정 삭제를 추가해보자
#유저 id로 정보조회
@app.post("/user/{user_id}", status_code=status.HTTP_200_OK)
async def read_user(user_id: str, db: db_dependency):
user = db.query(model.User).filter(model.User.user_id == user_id).first()
if user is None:
raise HTTPException(status_code=404, detail='User not found')
return user
#유저 정보 수정
@app.put("/user/{user_id}", status_code=status.HTTP_200_OK)
async def update_user(user_id: str, user_update: validation.UserBase, db: db_dependency):
# user_id에 해당하는 유저를 DB에서 검색
user = db.query(model.User).filter(model.User.user_id == user_id).first()
# 유저가 존재하지 않을 경우 예외 발생
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# 유저 정보 업데이트
user.user_id = user_update.user_id
user.user_name = user_update.user_name
user.password = user_update.password
# 데이터베이스에 변경 사항 반영
db.commit()
db.refresh(user)
return {"message": "User updated successfully", "user": user}
#유저 삭제
@app.delete("/user/{user_id}", status_code=status.HTTP_200_OK)
async def delete_user(user_id:str, db: db_dependency):
user = db.query(model.User).filter(model.User.user_id == user_id).first()
if user is None:
raise HTTPException(status_code=404, detail='User was not found')
db.delete(user)
db.commit()
이제 게시글 테이블을 추가해 보자
연관관계를 매핑 할때 User 클래스를 매핑할때 user_id를 쓰게 되는데 User 클래스의 user_id와 혼동 할 우려가 있어 custom_id로 변경 하였다.
config를 제외한 테이블 추가, 유효성, CRUD 코드를 추가하면 된다.
연관관계를 설정하기 위해
ForeignKey 클래스를 추가해주고 각 객체클래스에 관계 정의를 추가했다.
#model.py
from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship
from config import Base
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, index=True)
custom_id = Column(String(50), unique=True)
user_name = Column(String(50))
password = Column(String(100))
# Post와의 관계 정의 (1:N)
posts = relationship("Post", back_populates="user")
class Post(Base):
__tablename__ = 'post'
id = Column(Integer, primary_key=True, index=True)
title = Column(String(50))
content = Column(String(100))
custom_id=Column(String(50), ForeignKey('user.custom_id'))
# User와의 관계 정의
user = relationship("User", back_populates="posts")유효성 검사도 타입을 통일 시켰다.
#validation.py
from pydantic import BaseModel
class UserBase(BaseModel):
custom_id: str
user_name: str
password: str
class PostBase(BaseModel):
custom_id: str
title: str
content: str# 게시글 작성
@app.post("/post", status_code=status.HTTP_201_CREATED)
async def create_post(post: validation.PostBase, db: db_dependency):
db_post = model.Post(**post.dict())
db.add(db_post)
db.commit()