[FastAPI] 6. Database & CRUD

앞서 포스팅에서 DB에 CRUD 를 하기 위한 SQLAlchemy 모델과
데이터를 검증하기 위한 Pydantic 모델을 다루었는데요,
이어서 CRUD 메소드 중 Create 와 Read 를 구현해 보겠습니다.

Create

app/model/crud.py

번역 언어 정보 및 문장 쌍을 DB에 저장하는 메소드를 만들어 보겠습니다.
메소드 동작 순서는 다음과 같습니다.

1. SQLAlchemy 모델 인스턴스를 만들고 데이터를 넣습니다.
   (1.5. db = MysqlSession : 저는 db 세션을 여기서 지정했습니다.)
2. add : query라는 쿼리를 db 세션에 추가합니다. 이는 데이터베이스에 새로운 데이터를 추가하거나 업데이트하는 작업을 수행할 수 있습니다.
3. commit : 데이터베이스에 대한 변경 사항을 저장합니다.
4. refresh : query에 대한 데이터베이스의 최신 상태를 다시 가져옵니다. 이는 변경된 데이터가 올바르게 반영되었는지 확인하기 위해 사용될 수 있습니다.

async def create_translate(translate: schemas.TranslateCreate):
    # SQLAlchemy 모델 인스턴스를 만들고 데이터를 넣습니다.
    query = models.Translate(src_lang=translate.sl,
                                    src_text=translate.text,
                                    tgt_lang=translate.tl,
                                    mt_text=translate.mt
                                    )
    try:
        # db 세션 지정
        db = MysqlSession()
        db.add(query)
        db.commit()
        db.refresh(query)
    # 기본키에 어긋나는 데이터가 DB에 들어가면 발생
    except exc.IntegrityError:
        raise
	except exc.OperationalError:
        db.rollback()
        raise

Read

DB에 저장된 번역 이력을 가져오는 API 를 만들어 보겠습니다.

저는 추후 기능 추가를 고려하여 번역 (translate) 와 조회 (history) 라우터를 따로 만들었습니다.

main.py

• 라우터 추가

app.include_router(history_router, prefix="/history")

app/routes/history.py

• db dependency 추가
• DB 에 저장된 번역 결과를 가져와서 schema 에 맞춰서 리턴

from app.model.database import MysqlSession, engine
from app.model.models import Base

Base.metadata.create_all(bind=engine)

# Create routing method
router = APIRouter()

# Dependency
def get_db():
    db = MysqlSession()
    try:
        yield db
    finally:
        db.close()

# DB 조회 결과를 Translate schema 에 맞춰서 리턴합니다.
@router.get("/", response_model=list[schemas.Translate])
# @router.get("/")
async def read_history(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    histories = await history.get_history(db, skip=skip, limit=limit)
    return histories

app/models/schemas.py

class Translate(BaseModel):
    id: int
    src_lang: str
    src_text: str
    tgt_lang: str
    mt_text: str
    created_at: datetime

	# relation data 포함하여 리턴
    class Config:
        orm_mode = True

app/service/history.py

• Read API 호출

async def get_history(db: Session, skip: int = 0, limit: int = 100):
    return await crud.get_history(db=db, skip=skip, limit=limit)

app/model/crud.py

• model.Translate 즉 translate 테이블 데이터 읽기

async def get_history(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Translate).offset(skip).limit(limit).all()

Result

Create

Excute

DB

Read