SQLAlchemy ORM & async 활용 (2.0.X)

SQLAlchemy에서 2.0 버전이 나와서 문법이 꽤 많이 변경되었습니다.

그래서 오늘은 변경된 ORM 문법과 함께 async로 접근하는 방법을 알아보았습니다.

 

변경된 ORM 문법을 바로 보도록 하겠습니다.

from datetime import datetime

from sqlalchemy import String, BIGINT
from sqlalchemy.dialects.mysql import DATETIME
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    type_annotation_map = {
        int: BIGINT,
        datetime: DATETIME
    }


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

이전의 문법과 달리 Mapped를 활용하여 파이썬의 자료구조에 매핑하는 방식을 활용하고 있습니다.

mapped_column에서 실제 db에 해당하는 자료구조로 매핑을 하여 사용하는데, type_annotation_map으로 선언해서 활용할수도 있습니다.

 

async와 orm을 함께 활용하는 방법도 알아보겠습니다.

sqlalchemy 2.0 버전에서는 async에 대한 개선도 많이 이루어졌다고 합니다.

create_async_engine을 활용하여 engine을 만들고, async_sessionmaker를 활용하여 session으로 활용할 수 있습니다.

예시 코드는 아래와 같습니다.

import asyncio
from datetime import datetime

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy import String, BIGINT, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.dialects.mysql import DATETIME


class Base(DeclarativeBase):
    type_annotation_map = {
        int: BIGINT,
        datetime: DATETIME
    }


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())


async def insert_object(async_session, user: User):
    async with async_session() as session:
        async with session.begin():
            session.add_all(
                [user]
            )
            session.commit()


async def select_all_objects(async_session):
    async with async_session() as session:
        query = select(User)
        results = await session.execute(query)
        print([user.name for user in results.scalars().all()])


async def main():
    engine = create_async_engine("mysql+aiomysql://root:123123@localhost/study", echo=True)

    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    await insert_object(async_session, User(name="test"))

    await select_all_objects(async_session)

    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

engine을 생성할 때, 주의할 점은 db client를 async가 지원되는 라이브러리를 꼭 써야합니다.

mysql을 기준으로는 pymysql을 많이 활용하는데, async는 지원되지 않기 때문에 aiomysql을 사용하였습니다.

 

테이블을 생성할 때는 run_sync() 함수를 활용하여 생성할 수 있습니다.

다른 작업들은 session만 열어서 활용하면 되는데, async_sessionmaker()로 async_session을 생성하여 필요할 때마다 context 방식으로 사용하도록 구현하였습니다.

이렇게 되면, context 내에서 세션을 활용할 때 마다 connection pool에서 connection을 하나씩 할당받아서 사용하고 자동으로 context가 끝이나면 connection을 반환하게 사용합니다.

sqlalchemy에서도 이렇게 사용하는 것을 권장한다고 합니다.

 

sqlalchemy 2.0에서 async에 대한 부분이 많이 개선이 되어, 기존 버전보다 좀 더 깔끔하고 편하게 async를 활용할 수 있게 되었습니다.

fastapi를 활용하면, async로 서버가 동작하기 때문에 db도 async로 접근하는 것이 필수인데요.

python에서는 orm을 활용하려면 sqlalchemy가 제일 잘 되어 있기 때문에, sqlalchemy 2.0이 많이 선택될 것 같습니다.