내장 ORM이 있는 Django와 다르게 Flask와 FastAPI는 SQLAlchemy 라는 ORM 라이브러리를 이용합니다.

ORM은 이전 포스트에서도 얘기를 했어서 짧게 간략하자면 Object Relation Mapping의 약자로 객체를 이용해서 데이터베이스 Entity에 접근하는 방법입니다. 보통 어플리케이션 레벨에서 DB에 접근할 때는 데이터베이스 드라이버를 이용하여 SQL Query를 던져 실행하지만 SQL Query는 소프트웨어 엔지니어에게 있어 러닝 커브를 증가시키고, 소프트웨어 코드 가독성을 저하시키는 원인이 되었습니다.

ORM을 이용하면 기본적인 CRUD를 포함해 간단한 쿼리에 대해 SQL Query를 프로그래밍 코드에 질의하지 않아돋 프로그래밍 코드 안에서 처리할 수 있는 이점을 얻을 수 있습니다.

Python 언어에서 대표적인 ORM 라이브러리로는 SQLAlchemy가 있습니다.

---

1. SQLAlchemy

Flask용 SQLAlchmy dependency가 있을 정도로 생태계가 큰 라이브러리입니다. JPA처럼 어느 프레임워크든 dependency를 올릴 수 있는 곳이라면 어디서든 사용이 가능합니다. JPA와 마찬가지로 Connection Pool, Lazy loading 등을 지원합니다. Postgresql을 파이썬에서 이용하기 위해서는 psycopg2-binary dependency를 추가하여야 합니다. (추가적으로 FastAPI, SQLAlchemy dependency 설치)

---

2. Create Model

필요한 dependncy를 설치했다면 사용할 테이블을 만들어 보겠습니다. ORM이므로 사용하고 싶은 테이블은 class로 정의할 수 있습니다. Base 테이블은 따로 만들어 default로 존재하는 id, created_at, updated_at 과 같은 속성을 추가해두고 써도 좋습니다.

from sqlalchemy import Column, Boolean, String, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import UUID

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

class Memo(Base):
	__tablename__ = 'memos'
    
    id = Column(UUID(as_uuid=True), primary_key=True)
    title = Column(String(250), nullable=False)
    content = Column(Text, nullable=True)
    is_favorite = Column(Boolean, nullable=False)

간단한 메모장을 만들기 위해 Memo라는 Entity를 생성하였습니다. create_all 함수는 앱 실행 시 자동으로 테이블을 만들어주는 함수입니ㅏㄷ.

---

3. Create Connection

데이터베이스 서버와 연결하기 위한 세션을 만들어보도록 하겠습니다.

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

engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{db_name})
session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

세션은 서버에서 DB에 요청을 보내기 위한 통로 역할을 합니다. 각각의 파라미터에 알맞는 데이터를 넣고 코드를 작성합니다.

scoped_session, autocommit, autoflush 들에 대해서도 아는 것도 중요하지만 데이터베이스관련 포스팅을 할 때 쓰도록 하겠습니다.

세션을 부르는 함수가 종료될 때 같이 종료될 수 있도록 의존 함수를 만들겠습니다.

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

FastAPI에서 각 API마다 DB에 종속되었을 때 함수별로 세션을 부여하고, 작업이 끝나면 close될 수 있도록 의존 함수를 만들어줍니다.

---

4. Create response model

API 함수를 만들어보도록 하겠습니다.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from typing import Optional, List
from pydantic import BaseModel
from uuid import UUID

app = FastAPI()

class ResponseMemo(BaseModel):
	id: UUID
    title: str
    content: Optional[str] = None  # python 3.10부터는 파이프라인으로 Optional 대체 가능
    is_favorite: bool
    
    class Config:
    	orm_mode = True

# 마찬가지로 List는 3.10부터 list로 사용 가능)
@app.get('/memos', response_model=List[ResponseMemo]) 
async def get_memos(db: Session = Depends(get_db)):
	memos = db.query(Memo).all()
    return memos

위에서 만들었던 종속 함수를 사용하여 FastAPI에서 제공하는 Depends 함수를 이용해 API가 호출될 때마다 DB 세션을 생성하여 사용할 수 있도록 구현할 수 있습니다.

이렇게 구현하면 API가 호출될 때 세션이 생성되고, 사용이 끝나면 세션이 자동으로 해제됩니다.

덧붙여서 pydantic에서 제공하는 orm_mode를 이용하여 반환 모델을 만들게 되면 ORM JSONEncoder에 의해 자동으로 json으로 변환시켜주어 별도로 JSONResponse 등의 객체를 이용할 필요가 없습니다.

마지막으로 전체의 메모 내용을 가져오기 때문에 response_model에 List를 이용합니다.

from pydantic import BaseModel
from typing import Optional

class RequestMemo(BaseModel):
	title: str
    content: Optional[str] = None
    is_favorite: Optional[bool] = False
    

class ResponseMemo(BaseModel):
	id: UUID
    title: str
    content: Optional[str] = None  # python 3.10부터는 파이프라인으로 Optional 대체 가능
    is_favorite: bool
    
    class Config:
    	orm_mode = True
        
        
@app.post('/memos', response_model=ResponseModel)
async def register_memo(req: RequestMemo, db: Session = Dependsd(get_db)):
	memo = Memo(**req.dict())
    
    db.add(memo)
    
    # autoCommit을 False로 해두었기 때문에 commit을 사용하지않으면 반영되지 않는다. 
    db.commit()
    
    return memo

id는 자동으로 생성되고, 사용자로부터 받아야할 내용은 제목과 내용입니다. POST 메소드 작성 시, 우리는 위에서 autoCommit을 설정해주지 않았으므로 수동으로 commit 함수를 호출해야만 데이터베이스에 데이터가 영속될 수 있습니다.

app.put('/memos/{item_id}', response_model=ResponseMemo)
async def update_memo(item_id: UUID, req: RequestMemo, db: Session = Depends(get_db)):
	memo = db.query(Memo).filter(Memo.id == item_id).first()
    req_dict = req.dict()
    req_dict['id'] = item_id
    
    for key, value in req_dict.items():
    	setattr(memo, key, value)
        
    db.commit()
    
    return memo

PUT 메소드를 작성할때는 기존 메모 데이터가 있는지를 filter 메소드를 통해 데이터를 검색한 후 진행하도록 합니다.

from starlette.responses import Response
from starlette.status import HTTP_204_NO_CONTENT


@app.delete('memos/{item_id}')
async def delete_memo(item_id: str, db: Session = Depends(get_db)):
	memo = db.query(Memo).filter(Memo.id == item_id).first()
   
   db.delete(memo)
   db.commit()
   
   return Response(status_code=HTTP_204_NO_CONTENT)
   

Session에서 제공하는 delete 함수를 이용해 데이터를 삭제할 수 있습니다. 데이터를 삭제할 때는 filter 를 통해 모델 데이터를 받아 진행합니다.

---

SQLAlchemy의 내용은 오늘 다룬 것 이외에도 엄청 많이 있습니다. 두 개 이상의 복합 테이블을 이용한다면 설계를 할 때 심오하고 빠른 최적화를 위한 설계가 필요합니다.

Marshmallow를 사용할 때는 orm_mode가 제공되지 않아 JSON이나 dict으로 변환하기 위한 함수를 별도로 재구현해야했습니다.

하지만 Pydantic은 JSON, dict 변환에 대한 로직을 개발자가 별도로 구현하지 않아도 SQLAlchemy가 제공하는 객체에 맞게 자동으로 코드 한 줄만 작성해서 반환 모델을 JSON으로 변환해주는 로직을 제공해주는 점은 굉장히 편했습니다.