[FastAPI] 2. SQLAlchemy를 이용한 간단한 CRUD API 만들기

이번 글에서는 ORM에 대한 사용 방법에 대해 알아보도록 하겠습니다.

 

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

 

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

 

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

 

 

 

 

SQLAlchemy

SQLAlchemy는 본래 Flask에서 Flask용 SQLAlchemy 디펜던시가 있을 정도로 라이브러리 생태계가 굉장히 넓습니다. 그렇다고 해서 Flask에서만 사용할 수 있는 것은 아닙니다. JPA처럼 어느 프레임워크든, 애플리케이션이든 디펜던시를 올릴 수 있는 곳이라면 어떤 곳이든 사용이 가능하며 JPA와 마찬가지로 Connection Pool, Lazy loading 등을 지원합니다.

[tool.poetry]
name = "fastapiexample"
version = "0.1.0"
description = ""
authors = ["Neon K.I.D <contact@neonkid.xyz>"]

[tool.poetry.dependencies]
python = "^3.8.5"
fastapi = "^0.63.0"
uvicorn = "^0.13.2"
SQLAlchemy = "^1.3.22"
psycopg2-binary = "^2.8.6"

[tool.poetry.dev-dependencies]

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

우리는 이 글에서 PostgreSQL을 연동한다고 가정하고, SQLAlchemy와 Python용 Postgres 드라이버인 psycopg2-binary 디펜던시를 추가해보도록 합시다.

 

 

 

 

 

Create Model

필요한 디펜던시를 모두 설치했다면 이제 사용할 테이블을 만들어보도록 하죠. ORM이기 때문에 사용하고 싶은 테이블은 class로 만들어 정의할 수 있습니다.

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

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

class Memo(Base):
    __tablename__ = 'memos'

    id = Column(String(120), primary_key=True, default=lambda: str(uuid.uuid4()))
    title = Column(String(80), default='No title', nullable=False, index=True)
    content = Column(Text, nullable=True)
    is_favorite = Column(Boolean, nullable=False, default=False)

간단한 메모장 만들기 위해 Memo라는 Entity를 생성해보도록 하겠습니다. 메모장의 고유 ID와 제목, 내용 그리고 즐겨찾기 여부를 확인하는 is_favorite를 만들어줍니다.

 

그리고 앱 실행시 테이블을 자동으로 만들도록 create_all 함수를 호출해줍니다.

 

 

 

Create Connection

서버를 연결하기 위한 세션을 만들어보도록 하겠습니다. 

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


engine = create_engine('postgresql://{username}:{password}@{host}:{port}/{db_name}'.format(
    username='postgres', password='postgres', host='127.0.0.1', port='5432', db_name='nkmemo'
))
db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

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

 

autoCommit을 비활성화하는 이유는 데이터 변경 작업을 사용할 경우 여러 줄의 SQL 쿼리를 사용했을 때 한 번에 반영시키도록 하기 위함입니다. 가령 예를 들어 두 테이블이 동시에 변경 작업이 이루어진다고 했을 때, 하나의 ORM 코드에서 트랜잭션이 발생하더라도 코드에서 commit 함수를 호출할 때까지는 commit이 일어나지 않습니다.

 

autoFlush는 DB에서 flush를 자동화하겠냐는 여부인데, 이 역시 commit을 여러 줄로 발생시키기 위해 영속화를 자동화 시키지 않겠다는 조건으로 commit 함수가 발생했을 때만 영속화 하겠다는 옵션입니다.

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

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

 

 

 

 

Apply function

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

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from typing import Optional, List

app = FastAPI()


class ResponseMemo(BaseModel):
    id: str
    title: str
    content: Optional[str] = None
    is_favorite: bool

    class Config:
        orm_mode = True


@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: str
    title: str
    content: Optional[str] = None
    is_favorite: bool

    class Config:
        orm_mode = True


@app.post('/memos', response_model=ResponseMemo)
async def register_memo(req: RequestMemo, db: Session = Depends(get_db)):
    memo = Memo(**req.dict())

    db.add(memo)

    # 이 코드를 쓰지 않으면 DB에 반영되지 않음
    db.commit()

    return memo

id는 자동으로 생성되고, 사용자로부터 받아야할 내용은 제목과 내용입니다. 제목은 필수값으로 내용을 선택값으로 해줍니다.

 

POST 메소드 작성시, 우리는 위에서 autoCommit을 설정해주지 않았으므로 수동으로 commit 함수를 호출해야만 데이터베이스에 데이터가 영속됩니다. 따라서 commit 함수를 호출해주도록 합시다.

@app.put('/memos/{item_id}', response_model=ResponseMemo)
async def mod_memo(item_id: str, req: RequestMemo, db: Session = Depends(get_db)):
    memo = db.query(Memo).filter_by(id=item_id)
    req_dict = req.dict()
    req_dict['id'] = item_id
    
    req = {k: v for k, v in req_dict.items()}

    for key, value in req.items():
        setattr(memo, key, value)

    db.commit()

    return memo

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

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


@app.delete('/memos/{item_id}')
async def del_memo(item_id: str, db: Session = Depends(get_db)):
    memo = db.query(Memo).filter_by(id=item_id).first()

    db.delete(memo)
    db.commit()

    return Response(status_code=HTTP_204_NO_CONTENT)

Session에서 제공하는 delete 함수를 이용해 데이터를 삭제할 수 있습니다. 데이터를 삭제할 때는 반드시 filter, filter_by 등을 통해 모델 데이터를 받은 다음 진행합니다.

 

 

 

 

Test

테스트를 진행하기 전에, 데이터베이스에서 프로그래밍 코드에 입력한 데이터베이스를 생성한 후 진행해주세요. 테스트에는 Postman 애플리케이션을 사용해봤습니다.

 

PyCharm Professional을 사용하신다면 Database 기능을 이용하시는 것도 좋을 것 같네요.

 

애플리케이션을 실행하면 위와 같이 주소와 포트를 알려주며,

 

위와 같이 테이블을 자동으로 만들어줍니다.

 

먼저 GET 메소드를 호출하면 데이터베이스에 아무런 내용이 없으므로 아무것도 출력하지 않습니다.

 

메모를 하나 등록하기 위해 POST 메소드를 호출하고 결과를 보면 아래와 같이 내용이 출력되고 메모에 대한 ID가 부여됩니다.

 

데이터베이스에도 데이터가 들어간 모습을 볼 수 있습니다.

 

PUT 메소드를 이용하여 내용을 수정할 때 역시 잘 동작하면 위와 같이 결과가 출력됩니다.

 

마지막으로 DELETE 메소드의 경우 아까 전달받은 ID를 주소 뒤에 넣고 실행했을 때 아무런 문제가 없다면 204 No Content 코드와 함께 아무런 내용없이 출력하게 됩니다.

 

데이터베이스에도 잘 삭제가 되는 모습이죠.

 

 

 

 

 

 

마치며...

여기까지 FastAPI와 SQLAlchemy를 이용하여 DB 연동 후 API를 만드는 방법에 대해 알아봤습니다. 

 

SQLAlchemy의 내용은 이보다 더 방대합니다. 지금은 간단하게 하나의 테이블을 이용하였지만 두 개 이상의 복합 테이블을 이용한다면 설계를 할 때 좀 더 심오하고 빠른 최적화를 위한 설계가 필요합니다.

 

개인적으로 Marshmallow를 사용했을 때는 orm_mode가 별도로 제공되어 있지 않아 JSON이나 dict로 변환하기 위한 추상 함수를 별도로 재구현 해야 했습니다. 혹은 JSONEncoder를 별도로 만드는 것도 하나의 방법이죠.

 

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

 

또한 Depends 함수를 제공해주어 DB 세션에 대한 관리가 좀 더 편했는데요. 미들웨어를 이용하거나 decorator 등을 이용해서도 비슷하게 구현할 수 있지만 애플리케이션 개발자가 App context의 라이프 사이클에 맞춰서 고려해야 하는 점이 늘 큰 과제였는데, 이를 프레임워크에서 함수로 제공해주는 것은 개발자가 애플리케이션 레벨 개발에서만 신경쓰도록 노력했다는 점에 눈에 띄었습니다.

 

한 가지 더 바라는 점이 있다면 아직까지 관계형 데이터베이스에 대한 처리에 대해 비동기 처리가 미흡하다는 점이 있습니다. 물론 ORM 레벨에서도 그 지원이 미흡하다는 점이 있지만 데이터베이스 인스턴스에서도 지원하지 않는다는 것은 참 아쉬운 이야기 입니다. Java 진영에서는 R2DBC라는 커넥션을 이용해 데이터베이스 레벨에서 비동기 처리를 지원하지만 아직까지 데이터베이스 미들웨어에서 지원하지 않아 많은 관계형 데이터베이스에서 이점을 누릴 수 없습니다.

 

Python 진영에서도 SQLAlchemy가 1.4부터 비동기 지원을 하고 있고, 현재 베타 버전이 릴리즈된 상태입니다. 비동기 프레임워크가 많이 보편화 되었긴 했지만 가장 많이 사용되는 데이터베이스와의 연결에서 비동기가 지원되지 않는다는 점에 있어서 아쉬운 점이 많이 있는데요. 차후 데이터베이스에서도 비동기 처리가 안정적으로 될 수 있는 날이 왔으면 좋겠습니다.