엔지니어 대한민국

시작하기에 앞서 데이타베이스 연결을 우선적으로 합니다.

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)

코드를 실행하다보면 connection이 끊어질때가 있는데 그때마다 engine을 다시 연결해주시면 됩니다.

그리고 이전시간에 사용했던 테이블을 계속해서 사용할 건데요. 테이블이 데이타베이스에 없으신 분들은 아래 코드를 실행하여 생성해주시기 바랍니다.

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})"

# 테이블 생성
Base.metadata.create_all(engine)

# 생성한 테이블 삭제
Base.metadata.drop_all(engine)

select() SQL 표현식 구조

select문도 insert와 마찬가지로 statement을 먼저 만든후에 실행을 하게 되는데요. select()함수의 statement를 보시면 아래와 같습니다. 아래는 이름이 “spongebob”인 사용자의 레코드를 가져오는 코드입니다.

from sqlalchemy import select
stmt = select(User).where(User.name == "spongebob")

>>> print(stmt)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = :name_1

위의 User는 ORM으로 정의 되었지만 ORM없이 mapper()를 통하여 mapping을 하는 경우 Table()로 정의된 변수로 대체하여 실행해도 같은 결과가 나옵니다. 다만 Table객체는 칼럼이 .c안에 들어가 있으므로 Table.c.칼럼명 이렇게 해 주셔야 에러없이 실행이 됩니다. 결과는 ORM통해서 실행했을때와 완전 동일하게 나옵니다.

from sqlalchemy import select
stmt = select(user_table).where(user_table.c.name == "spongebob")

>>> print(stmt)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = :name_1

위와 같이 statement을 만들고나서 engine에 connect해서 해당 statement을 execute()함수로 실행을 시켜야 결과를 반환합니다.

with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(row)

위와 같은 코드를 실행하면 아래와 같이 spongebob의 레코드를 볼수 있습니다. Model은 mapped_column이나 relationship또는 DeclarativeBase등을 사용하여 베이스를 만들었지만 사실상 ORM의 Session으로 실행하지 않으면 온전히 ORM을 사용한다고 볼수가 없습니다. 그런 의미에서 아래 나오는 결과는 쿼리의 결과이지 ORM이 넘겨주는 객체의 결과물은 아니에요.

(1, 'spongebob', 'Spongebob Squarepants')

ORM을 사용할때는 engine에서 connect를 하지 않고, Session을 통해서 쿼리를 하는데요.

from sqlalchemy.orm import Session

stmt = select(User).where(User.name == "spongebob")
with Session(engine) as session:
    for obj_user in session.execute(stmt):
        print(obj_user)

위와 같이 Session을 사용하는 경우 반환되는 레코드가 User모델객체이기 때문에 print를 했을때 위에서 선언한 User.__repr__()함수에서 정의한 대로 결과를 출력하니까 여기서 보여주지 않아도 되는 필드는 빼거나 보여주고 싶은 항목은 추가하시면 됩니다. 위의 ORM 코드를 실행하면 다음과 같습니다.

(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)

검색필드 지정 및 가져올 테이블 설정

select()함수는 해당 모델의 데이타를 불러오는 기능을 합니다. 기본쿼리를 보면 다음과 같습니다.

>>> print(select(User))
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account

select에 인자로 필드명을 나열하면 명시한 필드만 불러오도록 쿼리를 생성합니다.

>>> print(select(User.name, User.fullname))
SELECT user_account.name, user_account.fullname
FROM user_account

실제로 쿼리를 실행해보겠습니다

from sqlalchemy.orm import Session

with Session(engine) as session:
    row = session.execute(select(User)).first()
    print(row)

위의 코드를 실행하면 다음과 같습니다.

(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)

검색결과가 배열로 반환되기 때문에 User객체 하나만 보고 싶다면 배열의 첫번째 항목을 읽어오면 됩니다.

>>> row[0]
User(id=1, name='spongebob', fullname='Spongebob Squarepants')

To be continued…사장님이 일시켜서 가봐야대요….

Resources

• Using SELECT Statements

시작하기에 앞서 데이타베이스 연결을 우선적으로 합니다.

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)

코드를 실행하다보면 connection이 끊어질때가 있는데 그때마다 engine을 다시 연결해주시면 됩니다.

그리고 이전시간에 사용했던 테이블을 계속해서 사용할 건데요. 테이블이 데이타베이스에 없으신 분들은 아래 코드를 실행하여 생성해주시기 바랍니다.

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})"

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})"

# 테이블 생성
Base.metadata.create_all(engine)

# 생성한 테이블 삭제
Base.metadata.drop_all(engine)

INSERT 문 사용하기

insert를 구현하는 가장 간단한 방법은 insert()함수를 사용하는 것입니다.

from sqlalchemy import insert

stmt = insert(User).values(name="spongebob", fullname="Spongebob Squarepants")

위에 코드는 insert를 실행한것은 아니구요. Insert를 하는 명령어를 stmt에 저장한 거에요. 출력해보면 다음과 같습니다.

>>> print(stmt)
INSERT INTO user_account (name, fullname) VALUES (:name, :fullname)

위의 statement을 컴파일하면 다음과 같이 넘겨주는 매개변수들을 parsing할 수 있습니다.

>>> compiled = stmt.compile()
>>> compiled.params
{'name': 'spongebob', 'fullname': 'Spongebob Squarepants'}

정의된 statement을 실행하여 데이타를 입력하려면 execute()함수를 실행합니다.

with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

실행 후 아래 명령을 실행하면 방금 실행한 레코드의 PK를 가져올수 있습니다.

>>> result.inserted_primary_key
(1,)

해당 테이블의 Insert문을 생성하는 방법은 아래와 같습니다.

>>> print(insert(User))
INSERT INTO user_account (id, name, fullname) VALUES (:id, :name, :fullname)

한번에 여러개의 행을 삽입하고 싶다면 아래의 코드와 같이 배열안에 값을 넣으시면 됩니다.

with engine.connect() as conn:
    result = conn.execute(
        insert(User),
        [
            {"name": "sandy", "fullname": "Sandy Cheeks"},
            {"name": "patrick", "fullname": "Patrick Star"},
        ],
    )
    conn.commit()

이번에는 User의 name을 username에 바인드를 해서 Address에 데이타를 넣는데, 이름 데이타만 넣어도 User.id를 자동으로 찾아와서 데이타를 넣어주는 고급기술을 써보도록 하겠습니다.

from sqlalchemy import select, bindparam

scalar_subq = (
    select(User.id)
    .where(User.name == bindparam("username"))
    .scalar_subquery()
)

with engine.connect() as conn:
    result = conn.execute(
        insert(Address).values(user_id=scalar_subq),
        [
            {
                "username": "spongebob",
                "email_address": "spongebob@sqlalchemy.org",
            }, {
                "username": "sandy",
                "email_address": "sandy@sqlalchemy.org",
            }, {
                "username": "sandy",
                "email_address": "sandy@squirrelpower.org",
            },
        ],
    )
    conn.commit()

이렇게 하면 scalar_subq가 서브쿼리로 들어가서 User.id를 가져오는 아래와 같은 쿼리를 생성하여 실행합니다.

BEGIN (implicit)
INSERT INTO address (user_id, email_address) VALUES ((SELECT user_account.id
FROM user_account
WHERE user_account.name = ?), ?)
[...] [('spongebob', 'spongebob@sqlalchemy.org'), ('sandy', 'sandy@sqlalchemy.org'),
('sandy', 'sandy@squirrelpower.org')]
COMMIT

이건 별로 중요한건 아닌데 참고로, insert()에 values를 아무것도 넣지않으면 필드와 데이타가 모두 비어있는 Insert문을 반환합니다. 필요한 경우 유용할수도 있겠죠.

>>> print(insert(User).values().compile(engine))
INSERT INTO user_account () VALUES ()

Insert후에 반환되는 값 사용하기

insert문을 만들때 아래와 같이 returning함수를 붙여서 그 안에 반환받을 필드를 나열하면

insert_stmt = insert(Address).returning(
    Address.id, Address.email_address
)

아래와 같이 쿼리를 만들어 주고 쿼리 실행 후에 반환되는 값을 사용할 수 있습니다.

>>> print(insert_stmt)
INSERT INTO address (id, user_id, email_address) VALUES (:id, :user_id, :email_address) RETURNING address.id, address.email_address

Insert.from_select()를 이용하면 Select해온 결과를 Insert할 수도 있습니다.

select_stmt = select(User.id, User.name + "@aol.com")
insert_stmt = insert(Address).from_select(
    ["user_id", "email_address"], select_stmt
)
>>> print(insert_stmt.returning(Address.id, Address.email_address))
INSERT INTO address (user_id, email_address) SELECT user_account.id, user_account.name || :name_1 AS anon_1
FROM user_account RETURNING address.id, address.email_address

Resources

• Using INSERT Statements

외국문서 보면 말이 거창해서 겁부터 나는데요. 알고 보면 너무 당연하고 상당히 쉬운 컨셉의 내용들이 참 많습니다. 여러분께 조금이나마 도움이 되는 문서가 되길 바랍니다. 본 문서는 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'),)

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

Resources

• SQLAlchemy Unified Tutorial
• Working with Transactions and the DBAPI
• Working with Database Metadata
• 엔:SQLAlchemy Classical Mappings