Flask SQLAlchemy vs SQLAlchemy

 

파이썬을 서비스를 개발할 때 ORM은 SQLAlchemy 말고 딱히 다른 옵션이 없다. 그런데 Flask 개발자는 일반 SQLAlchemy와 Flask에 최적화된 Flask-SQLAlchemy 중 한가지를 골라야한다. 

 

사실 이 두가지는 완전히 다른 라이브러리는 아니다. Flask SQLAlchemy는 기존 SQLAlchemy에서 일부를 가져와 Flask에 맞게 바꾼 형태다. 세부적으로 말하면 Flask-SQLAlchemy는 세션 관리를 app 컨텍스트에 맞춰 관리해주기 때문에 편하다. 반면 SQLAlchemy를 사용하는 경우에는 세션 관리를 개발자가 직접 해야한다. 

 

SQLAlchemy를 사용하는 경우 코드는 아래처럼 짜게 된다. 

 

SQLALCHEMY_DATABASE_URL = "postgresql://common:common@localhost/common"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False)
SessionLocal = sessionmaker(autoflush=False, bind=engine, autocommit=False)

Base = declarative_base()

def close_db(e=None):
    SessionLocal.remove()

def init_app(app: Flask):
    app.teardown_appcontext(close_db)
    Base.metadata.create_all(bind=engine)
    
    SessionLocal.query(SomeEntity).get({id: 3})

 

일반 SQLAlchemy를 사용하는 경우 먼저 엔진을 만들고 세션을 만든다. 개발 단계에선 비즈니스로직에서 세션을 임포트하고 사용해도 동작에 큰 문제 없이 잘돌아간다. 

 

그런데 이 코드는 함정이 있다. 멀티 쓰레드 환경에서 바라보는 Session은 모두 동일한 객체다. 작업을 처리할 때 모두 동일한 세션을 가지고 처리하게 된다. 테스트 단계에선 여러 요청이 동시에 들어오는 경우가 없기 때문에 하나의 세션으로 동시에 여러개의 작업을 실행하는 경우가 없어 문제가 없을지도 모른다.

 

그런데 프로덕션 단계에서 여러 요청이 동시에 들어오는 경우라면 어떨까? Flask의 쓰레드 A, B가 각각 별개의 데이터베이스 업데이트 명령을 처리한다고 하자. 만약 쓰레드 A에서 트랜잭션을 처리하는 중에 에러가 발생해 롤백을 하게 되는 경우가 생긴다면? 쓰레드 B가 작업한 내용들도 모두 날라가게 된다. 동일한 세션을 바라보는 경우 이런 문제가 발생한다. 

 

SQLAlchemy에서도 이런 문제점을 인식하고 scoped_session이라는 함수를 이용해 세션의 범위를 지정할 수 있게 만들었다.

 

from my_web_framework import get_current_request, on_request_end
from sqlalchemy.orm import scoped_session, sessionmaker

Session = scoped_session(sessionmaker(bind=some_engine), scopefunc=get_current_request)

@on_request_end
def remove_session(req):
    Session.remove()

 

함수 인자 scopefunc 로 현재 요청의 범위를 설정해주는 값을 넣는다면 세션의 범위를 지정해줄 수 있다. 매 요청마다 다른 세션을 갖는것이 바람직하므로 SQLAlchemy에서 제공하는 소스코드에선 get_current_request라는 함수를 쓸 것을 권장한다. 

 

Flask에서도 get_current_request와 비슷한 함수를 넣으면 세션 관리가 자연스럽게 해결 될 것이다. 실제로 스택오버플로우에서는 아래 코드를 이용해서 세션관리를 직접 한다고 한다.  스택오버플로우

 

from flask import Flask, _app_ctx_stack
from sqlalchemy.orm import scoped_session
from .database import SessionLocal, engine

app = Flask(__name__)
app.session = scoped_session(SessionLocal, scopefunc=_app_ctx_stack.__ident_func__)

 

 

그러나 최신 Flask에서는 먹히질 않는다. 최신 Flask에선 이런 속성들이 더이상 존재하지 않기 때문이다. 

 

AttributeError: '_FakeStack' object has no attribute '__ident_func__'

 

그러면 어떻게하면 좋을까? 오랜 시간 다른 방법을 고민하고 방법을 찾아봤지만 딱히 방법은 없었다. Flask API를 모두 뒤져볼 수도 없는 노릇이다. 그래서 데이터베이스를 초기화하는 부분만 직접 Flask SQLAlchemy를 사용했다. 그리고 세션을 호출하는 부분은 SQLAlchemy 에서 갖고 있는 session 값을 리턴하는 것으로 처리했다. 대신 Entity를 선언한 부분들은 SQLAlchemy 라이브러리를 사용해서 구현했다. Flask SQLAlchemy도 동일한 SQLAlchemy를 바라보는 것이기 때문에 호환성 이슈가 발생하지 않을 것이라고 생각했고 실제로도 동작할 때 문제가 없었다.

 

from flask import Flask, current_app, g
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.orm import scoped_session
from flask_sqlalchemy import SQLAlchemy

SQLALCHEMY_DATABASE_URL = "postgresql://common:common@localhost/common"

db = SQLAlchemy()

Base = db.Model

def get_session() -> Session:
    return db.session

def close_db(e=None):
    db.session.close()

def init_app(app: Flask):
    app.teardown_appcontext(close_db)

    app.config[
        "SQLALCHEMY_DATABASE_URI"
    ] = "postgresql://common:common@localhost/facegame_flask"

    db.init_app(app)

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .db import Base

class PlayerEntity(Base):
    __tablename__ = "player"

    id: int = Column(Integer, primary_key=True, autoincrement=True)
    user_id: int = Column(Integer, ForeignKey("user.id"), nullable=True)
    name: str = Column(String(20), unique=True, nullable=False)
    type: int = Column(Integer, nullable=False)
    profile: str = Column(String(50), nullable=True)

 

처음부터 Flask-SQLAlchemy를 이용하는게 좋긴하다. 그런데 이게.... Flask SQLAlchemy는 한 번더 계층(layer)가 추가된 단계여서 그런지 IDE 상에서 함수가 자동 완성이 되지 않는 문제가 있었다. 물론 실행하면 문제는 없지만 비쥬얼 코드상에서 함수 정보를 읽어오지 못하다 보니까 코딩 실수가 많아지고 내가 제대로 코드를 짰는지 안짰는지 알 수가 없다. 강타입 언어를 사용한 입장에서는 심히 불편한 요소다.

 

최대한 타입을 바라볼 수 있는 방향으로 코드를 짰다. get_session의 경우 리턴할 때 원래 SQLAlchemy 세션 클래스를 리턴했기 때문에 함수 자동완성에는 문제가 없었다.

 

def get_session() -> Session:
    return db.session