[Python] SqlAlchemy 1.4 -> 2.0 마이그레이션 단계별 가이드

들어가며

SqlAlchemy 버전 업그레이드 진행 과정에 대해 설명한 포스팅 입니다. 이 글을 통해 다음과 같은 내용을 확인할 수 있습니다

1. SqlAlchemy 1.4/2.0 버전의 차이
2. 마이그레이션을 위한 단계
3. 각 버전 별 쿼리 스타일의 차이

---

왜 마이그레이션을 해야할까?

SQLAlchemy 2.0 버전은 1.4 버전에서 지원하지 않는 기능들을 추가적으로 제공합니다.

• SQL 표현식 변경
• typing 지원
• Declarative(선언형) ORM 모델 개선
• bulk insert
• ORM insert
• update, upsert 등...
• 비동기 컨텍스트 매니저 지원

이외에도 버전 업그레이드에 따른 쿼리 성능의 개선이 있었기 때문에 마이그레이션을 통해 얻을 수 있는 기능상의 이점이 상당합니다.

자세한 내용은 다음 공식 문서에서 확인할 수 있습니다.

참고 : What's New in SQLAlchemy 2.0?

 

---

마이그레이션 Guide

SQLAlchemy 공식 문서에서 마이그레이션을 위한 단계적 작업이 정리되있습니다. 문서에서 언급되는 1.4 -> 2.0 으로의 마이그레이션은 1.4 버전에서 메이저 패치를 통해 상위 호환성을 보장한다고 합니다. 이는 2.0에서 변경되는 주요한 아키텍처 및 기능 수정사항이 이미 반영되어 있다는 것을 의미합니다.

 

이제부터 본적적으로 마이그레이션을 위한 단계적인 작업을 설명하겠습니다.

 

Step #1(Python 3.7)

SQLAlchemy 2.0 버전은 python 2의 EOL에서 영감을 받고 시작되었기 때문에 python 2.7 버전에 대한 지원 중단을 목표하고 있습니다. 따라서 2.0 버전의 최소 요구사항은 python 3.7이며 이에따른, python 버전 요구사항을 충족시켜야합니다.

 

마이그레이션 대상 프로젝트의 Python 버전을 3.7 로 올려주세요!

 

Step #2 (RemovedIn20Warnings)

RemovedIn20Warnings을 출력하는 플래그를 활성화 하여 레거시 패턴을 테스트코드에서 감지 할 수 있게합니다. 해당 경고를 출력하기 위해 SQLALCHEMY_WARN_20 환경변수를 true 혹은 1로 설정합니다.

 

SQLALCHEMY_WARN_20=1 pytest

 

출력 예시

~/service.py:305: RemovedIn20Warning: "User" object is being merged into a Session along the backref cascade path for relationship "Securitygroup.nics"; in SQLAlchemy 2.0, this reverse cascade will not take place.  Set cascade_backrefs to False in either the relationship() or backref() function for the 2.0 behavior; or to set globally for the whole Session, set the future=True flag (Background on this error at: https://sqlalche.me/e/14/s9r1) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
    user = User(

...
~/test_get_securitygroup_list.py::test_user_200
  ~/conftest.py:2073: RemovedIn20Warning: "User" object is being merged into a Session along the backref cascade path for relationship "Nic.securitygroups"; in SQLAlchemy 2.0, this reverse cascade will not take place.  Set cascade_backrefs to False in either the relationship() or backref() function for the 2.0 behavior; or to set globally for the whole Session, set the future=True flag (Background on this error at: https://sqlalche.me/e/14/s9r1) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
    saved_users.append(user)

...

 

Step #3 (RemovedIn20Warning 해결)

 

RemovedIn20Waring을 해결합니다. 공식문서에서는 각각의 Warning에 따른 해결 방법을 가이드 해주고 있습니다. 모든 경우를 이 포스팅에서 설명할 수는 없지만 한가지 케이스를 해결하는 방법을 소개합니다.

 

cascade_backrefs behavior deprecated for removal in 2.0

참고 : https://docs.sqlalchemy.org/en/14/changelog/migration_14.html#cascade-backrefs-behavior-deprecated-for-removal-in-2-0

RemovedIn20Warning: "User" object is being merged into a Session along the backref cascade path for relationship "Address.users"; in SQLAlchemy 2.0, this reverse cascade will not take place.
Set cascade_backrefs to False in either the relationship() or backref() function for the 2.0 behavior; or to set globally for the whole Session, set the future=True flag (Background on this error at: https://sqlalche.me/e/14/s9r1) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)

 

SQLAlchemy는 계단식 역참조(cascade_backref)를 지원해 왔지만, 다음과 같은 의도하지 않은 문제를 불러일으킬 소지가 있기 때문에 2.0 버전에서는 더 이상 지원하지 않습니다.

u1 = User()
session.add(u1)

a1 = Address()
a1.user = u1  # <--- adds "a1" to the Session

 

위 코드에서 이미 u1이 세션에 존재하는 상황에 a1.user = u1 을 하게 되면 u1.addresses.append(a1)을 암시하게 되고 a1이 계단식으로 세션에 등록되는 의도하지 않은 세션 등록이 발생하게 됩니다. (객체가 세션에 너무 일찍 배치되어 조기에 플러시 될 수 있습니다)

2.0에서는 relationship.cascade_backrefs 와 backref.cascade_backrefs 가 Fasle로 세팅되어 계단식 역참조현상이 발생하지 않도록 강제합니다.

 

마이그레이션 과정에서는 relationship의 cascade_backrefs를 False로 설정해 역참조 기능을 비활성화 시킬 수 있습니다.

class User(DecreativeBase):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String)
    addresses = relationship("Address", back_populates="user")
    
class Address(DecreativeBase):
    __tablename__ = 'addresses'
    
    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="addresses", cascade_backrefs=False)

 

Step #4, #5 (future flag의 사용)

1.4 버전에서 engine과 session의 future 플래그를 활성화 해 2.0버전에서 사용할 수 있는 새로운 API를 활성화 하고 검증할 수 있습니다.

 

create_engine.future의 활성화에 따른 변경 사항

• Connection.execute()의 새로운 인수 추가
• 암시적 자동커밋(autocommit)의 제거
• 문자열 SQL 구문에 text() 구문이 필요하게 된다. (단, Connection.exec_driver_sql() 메서드를 사용하지 않는 상황에서 만)
• 엔진 연결(engine.connection())이 없는 실행이 제거

Session.future의 활성화에 따른 변경 사항

• bound metadata 지원 종료
• session.start.subtransaction 플래그 지원 종료
  • session.commit()은 subtransaction을 조정하지 않고 항상 DB에 COMMIT을 시도
  • session.rollback()은 subtransaction을 유지하려 하지 않고 항상 전체 트랜잭션을 한번에 롤백
• session을 context manager로 사용가능

 

future flag 활성화 방법

from sqlalchemy import create_engine  
from sqlalchemy.orm import sessionmaker

engine = create_engine(settings.database.volume_url, pool_pre_ping=True, echo=True) 
session_maker = sessionmakerautocommit=False, autoflush=False, bind=engine, future=True)

 

Step #6 allow_unmapped 의 적용

SQLAlchemu 2.0버전은 typing을 지원하는 ORM model을 사용할 수 있습니다. typing을 사용하기 위한 기본 요구사항은 Mapped라는 Generic container를 사용하는 것 입니다. 2.0버전에서 Mapped를 사용하지 않은 relationship() 함수는 에러를 발생시킵니다.

 

__allow_unmapped__ 플래그를 사용해 1.4 버전에서 사용하던 모델을 변경없이 사용할 수 있습니다.

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]


class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }


# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

 

Step #7 SQLAlchemy 2.0 버전 업그레이드

마지막 단계입니다. 1.4버전에서 위 단계를 거쳐오며 모든 Warnning, Error를 해결했다면, 2.0버전으로 업그레이드하고 테스트합니다.

 

주의 : SQLAlchemy 2.0에서는 이전 버전과 호환되도록 의도된 API 동작 및 변경사항이 있지만, 그럼에도 일부 호환되지 않는 기능이 있을 수 있습니다. 충분한 테스트 이후에 버전을 업그레이드 하세요

---

SQLAlchemy 1.4 vs 2.0 Query 스타일 비교

2.0에서도 이전 버전인 1.4 버전에서 사용하던 Query를 지원하지만, 새롭게 변경된 2.0에서의 Query 스타일을 장려합니다.

눈에 띄는 변화는 Session.query()를 사용하는 대신 select()와 Session.execute()를 사용해 쿼리를 실행한다는 점입니다.

 

2.0 버전으로 마이그레이션을 하면서 1.4 스타일과, 2.0 스타일을 혼용해서 사용하기 보다는 쿼리 스타일도 2.0으로 통일하는 것이 좋습니다.

 

1.x 스타일	2.0 스타일	레퍼런스
session.query(User).get(42)	session.get(User, 42)	ORM Query - get() method moves to Session
session.query(User).all()	session.execute( select(User) ).scalars().all() # or session.scalars(select(User)).all()	ORM Query Unified with Core Select Session.scalars() Result.scalars()
session.query(User).\ filter_by(name='some user').one()	session.execute( select(User). filter_by(name="some user") ).scalar_one()	ORM Query Unified with Core Select Result.scalar_one()
session.query(User).\ filter_by(name='some user').first()	session.scalars( select(User). filter_by(name="some user"). limit(1) ).first()	ORM Query Unified with Core Select Result.first()
session.query(User).options( joinedload(User.addresses) ).all()	session.scalars( select(User). options( joinedload(User.addresses) ) ).unique().all()	ORM Rows not uniquified by default
session.query(User).\ join(Address).\ filter(Address.email == 'e@sa.us').\ all()	session.execute( select(User). join(Address). where(Address.email == 'e@sa.us') ).scalars().all()	ORM Query Unified with Core Select Joins
session.query(User).from_statement( text("select * from users") ).all()	session.scalars( select(User). from_statement( text("select * from users") ) ).all()	Getting ORM Results from Textual and Core Statements
session.query(User).\ join(User.addresses).\ options( contains_eager(User.addresses) ).\ populate_existing().all()	session.execute( select(User). join(User.addresses). options(contains_eager(User.addresses)). execution_options(populate_existing=True) ).scalars().all()	ORM Execution Options Populate Existing
session.query(User).\ filter(User.name == 'foo').\ update( {"fullname": "Foo Bar"}, synchronize_session="evaluate" )	session.execute( update(User). where(User.name == 'foo'). values(fullname="Foo Bar"). execution_options(synchronize_session="evaluate") )	UPDATE and DELETE with arbitrary WHERE clause
session.query(User).count()	session.scalar(select(func.count()).select_from(User)) session.scalar(select(func.count(User.id)))	Session.scalar()

 

---

마이그레이션 이후에 해볼만한 작업

Migrating an existing mapping

Step#6 에서 allow_unmapped를 적용해 Mapped 컨테이너를 사용하지 않고 ORM 모델을 작성할 수 있었지만, Mapped를 사용하면 보다 간결한 ORM모델을 작성할 수 있고, ORM 객체 컬럼에 type을 명시해 IDE 레벨에서 사용할 수 있습니다. 마이그레이션 이후에는 현재 선언된 SQLAlchemy 모델을 Mapped를 적용해 보는건 어떨까요?

 

참고 : https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#migrating-an-existing-mapping

 

Before

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

 

After

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

 

---

마치며

지금까지 SQAlchemy 버전 업그레이드를 위한 단계적인 작업을 알아봤습니다. ORM 버전을 올리는 것은 쉬운 선택이 아닙니다. 특히 DB와 관련된 기술은 이러한 작업이 굉장히 보수적으로 진행되는게 당연하게 받아들여지고 있습니다. 인프라, 비지니스 로직과 밀접한 연관성이 있으니까요. 만일 진행하는 프로젝트가 충분한 e2e, Integration 테스트가 작성되지 않았다면, 마이그레이션 작업을 권장하지 않습니다. 마이그레이션을 통해 얻을 수 있는 이점보다, 변경 후 프로젝트가 내포하는 불안정성이 더 크기 때문입니다.

 

하지만 1.4 버전은 이미 레거시 버전으로 지정되어 사후지원이 불투명한 상황이고, 2.0 버전으로 마이그레이션을 통해 얻을 수 있는 기술적인 이점이 분명한 상황입니다. 2.0 버전에서 1.4 버전 하위호환성도 어느정도 보장되어 있습니다. 만일 현재 작업하고 있는 프로젝트의 테스트코드가 충분히 작업되있는 상황이라면, 공식 가이드 문서를 참고해 2.0 버전 업그레이드를 수행하는 것을 추천합니다 😄

 

참고

https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html
https://docs.sqlalchemy.org/en/20/changelog/migration_20.html
https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#migrating-an-existing-mapping