SQLAlchemy Unified Tutorial – 엔지니어 대한민국

외국문서 보면 말이 거창해서 겁부터 나는데요. 알고 보면 너무 당연하고 상당히 쉬운 컨셉의 내용들이 참 많습니다. 여러분께 조금이나마 도움이 되는 문서가 되길 바랍니다. 본 문서는 SQLAlchemy매뉴얼의 SQLAlchemy 1.4 / 2.0 튜토리얼의 일부를 발췌하여 번역한 내용입니다.

Establishing Connectivity – the Engine

기본적으로 가지고 계신 DB에 연결을 우선 먼저 할게요.

from sqlalchemy import create_engine
from sqlalchemy import text

uri = 'mysql://DB_USER:DB_PASS@DB_HOST/DB_NAME'

engine = create_engine(uri, echo=False)

# 커넥션 테스트
with engine.connect() as conn:
    result = conn.execute(text("select 'hello world'"))
    print(result.all())

Working with Transactions and the DBAPI

연결이 잘 되었으면 테스트로 테이블을 하나 만들어 볼게요. DDL은 수동으로 커밋하도록 합니다.

with engine.connect() as conn:
    conn.execute(text("CREATE TABLE some_table (x int, y int)"))
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 1, "y": 1}, {"x": 2, "y": 4}],
    )
    conn.commit()

아래는 자동커밋의 예제입니다. begin()함수를 사용하면 DML은 commit()함수의 호출없이도 자동으로 커밋이 되어 데이타가 삽입이됩니다. DDL은 여전히 수동으로 커밋을 해야한다네요.

with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 6, "y": 8}, {"x": 9, "y": 10}],
    )

아래는 데이타를 가져와서 속성이름으로 데이타에 접근하는 예제입니다.

with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table"))
    for row in result:
        print(f"x: {row.x}  y: {row.y}")

for문에서 변수명을 지정해서 바로 쓸수 있게 할수 있지만, 하지만 두개 이상일때는 곤란해지기 때문에 잘 사용하지 않습니다.

with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table"))
    for x, y in result:
        print(f"x: {x}  y: {y}")

아래는 배열방 번호로 접근하는 예제입니다. 하지만 우리가 코딩할때 배열방 번호를 일일이 기억하고 있을 수는 없지요.

with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table"))
    for row in result:
        print(f"x: {row[0]}  y: {row[1]}")

그래서 사용하는 것이 mappings()입니다. 매핑은 필드명으로 접근하기 때문에 정확도가 높고 편리합니다.

with engine.connect() as conn:
    result = conn.execute(text("select x, y from some_table"))

    for dict_row in result.mappings():
        x = dict_row["x"]
        y = dict_row["y"]
        print(f"x: {x}  y: {y}")

Executing with an ORM Session

SQLAlchemy의 가장 상층에 존재하는 ORM(Object Relational Mapper)는 Python에서 데이터베이스를 사용하는데 필요한 많은 기능을 가지고 있는 종합선물세트 같은 존재입니다. 일부만 발췌하여 사용할 수도 있고, 종합적으로 사용할 수도 있습니다. 아래 그림은 ORM의 종속성을 보여주는 구성 계층도입니다.

아래는 ORM의 Session을 사용하여 쿼리를 해오는 예제입니다.

from sqlalchemy.orm import Session

stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y")
with Session(engine) as session:
    result = session.execute(stmt, {"y": 6})
    for row in result:
        print(f"x: {row.x}  y: {row.y}")

ORM사용시에도 UPDATE를 할때는 commit()을 명시적으로 호출하여 변경된 내용을 데이타베이스에 적용합니다.

with Session(engine) as session:
    result = session.execute(
        text("UPDATE some_table SET y=:y WHERE x=:x"),
        [{"x": 9, "y": 11}, {"x": 13, "y": 15}],
    )
    session.commit()

Working with Database Metadata

아래는 MetaData 선언하여 user테이블을 매핑하는 코드입니다. Table()에 칼럼정보를 저장하고, 모델에는 순수하게 필요한 기능만 나열한 뒤, mapper()함수로 둘을 연결해주면 클래스가 간결해지니까 코드가 훨씬 보기 수월해 지겠죠.

from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey
from sqlalchemy.orm import mapper

metadata = MetaData()

user = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(50)),
            Column('fullname', String(50)),
            Column('nickname', String(12))
        )

class User(object):
    def __init__(self, name, fullname, nickname):
        self.name = name
        self.fullname = fullname
        self.nickname = nickname

# sqlalchemy.orm.mapper()는 현재 없어졌다네요
# 이 방법은 옛날 방법이니까 가급적 사용하지 마시고요
# 아래에 소개해 드리는 ORM으로 구현한 방법을 사용하시기 바랍니다.
mapper(User, user)

참고로 create_all()함수를 실행하면 위의 테이블을 생성 할수 있게 됩니다.

metadata.create_all(engin)

ORM의 DeclarativeBase

Base클래스를 만들어서 MetaData에 접근할수 있습니다.

from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass

이렇게 하면 MetaData를 따로 가져오지 않아도 필요한 객체에 접근할 수 있습니다.

>>> Base.metadata
MetaData()
>>> Base.registry
<sqlalchemy.orm.decl_api.registry object at 0x1069cf0e0>

아래는 Declarative를 이용하여 두개의 모델을 매핑하는 PEP484에 따른 가장 현대적인 유형의 코드입니다.

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

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

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

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))
    user: Mapped[User] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

주석때문에 너무 정신없죠. 주석빼고 이렇게 써도 됩니다.

from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy import ForeignKey
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

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

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"

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

    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, user_id={self.user_id!r}, email_address={self.email_address!r})"

마찬가지로 이렇게 선언한 후에 create_all()함수를 호출하면 테이블이 생성됩니다.

Base.metadata.create_all(engine)

그런데 이렇게 선언을 다 해놓고, 옛날 방법으로 데이타를 불러온다면 말짱 도로묵입니다. 아래와 같이 암만 User를 ORM Base로 선언하고, ORM의 mapped_column()를 이용하여 칼럼을 선언했어도. engine.connect()를 통해서 statement을 실행하면 그 결과는 ORM이 아니라 그냥 옛날에 사용하던 방법으로 나오던 쿼리 결과일 뿐입니다.

from sqlalchemy import select

stmt = select(User).where(User.name == "spongebob")
with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(row)
# engine.connect.execute()을 통해서 실행한 결과 그냥 데이타 Set임
(1, 'spongebob', 'Spongebob Squarepants')

ORM을 제대로 사용하시려면, 선언도 ORM으로 하시고 실행도 아래와 같이 ORM.Session으로 하셔야합니다.

from sqlalchemy.orm import Session

with Session(engine) as session:
    row = session.execute(select(User)).first()
    print(row)
# ORM으로 실행한 결과 모델클래스의 객체가 반환됨
(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)

다음 시간에 이어서 데이타를 입력하고 수정하고 검색하는 내용을 배워보도록 하겠습니다. 오늘은 여기까지 잘 따라오셨습니다. 쉬는 시간에는 꼭 쉬세요^^

Establishing Connectivity – the Engine

Working with Transactions and the DBAPI

Executing with an ORM Session

Working with Database Metadata

ORM의 DeclarativeBase

Resources