python / / 2024. 8. 20. 07:02

[sqlalchemy] ORM Quick Start

이 내용은 공식문서의 일부를 번역한 내용이다. (https://docs.sqlalchemy.org/en/20/orm/quickstart.html)

아래 내용은 기본 ORM 사용 방식을 빠르게 확인하고 싶은 사용자들을 위해, SQLAlchemy Unified Tutorial에서 사용된 매핑과 예제를 간략하게 소개한다. 이 코드는 명령줄에서 완전히 실행할 수 있다.

이 섹션의 설명은 의도적으로 매우 간단하게 작성되었으므로, 여기서 설명하는 개념들에 대해 더 깊이 있는 설명을 원한다면 전체 SQLAlchemy Unified Tutorial을 참조하시기 바란다. (https://www.sqlalchemy.org/)

모델 선언

여기에서는 데이터베이스에서 쿼리할 구조를 형성할 모듈 수준의 구성 요소를 정의한다. 이 구조는 선언적 매핑(Declarative Mapping)이라고 하며, Python 객체 모델과 함께 특정 데이터베이스에 존재하거나 존재할 SQL 테이블을 설명하는 데이터베이스 메타데이터를 동시에 정의한다.

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


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", cascade="all, delete-orphan"
    )

    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[int] = 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})"
  • _tablename_ : 테이블 이름을 나타낸다.
  • Mapped[int]
    • Mapped 타입을 포함하면 테이블의 컬럼으로 생성된다.
    • Python 타입에서 데이터베이스 타입이 결정된다. (int --> INTEGER, str --> VARCHAR)
    • null 허용 여부는 Optional[] 타입을 사용하면 된다.
    • 좀 더 구체적인 타입을 지정하려면 User.name의 mapped_column() 에서 사용된 타입 객체를 사용할 수있다. (String
  • mapped_column()
    • 구체적인 정의가 필요한 경우 사용된다.
    • 기본키, 외래키등 세부정보를 나타낼 때 사용한다.
  • 기본키
    • 최소 하나의 열이 필요하며 mapped_column(primary_key=True)을 사용하여 나타낸다.
    • 위에서 User.id, Address.id가 기본키다.
  • _repr_
    • 필수항목은 아니지만 디버깅할 때 유용하다.

Engine 생성

Engine은 새로운 데이터베이스 연결을 생성할 수 있는 팩토리이며, 빠른 재사용을 위해 커넥션풀 안에 연결을 유지한다.

from sqlalchemy import create_engine
engine = create_engine("sqlite://", echo=True)

echo=True 는 SQL이 표준 출력으로 표시되게 하는 것이다.

CREATE TABLE DDL 생성

테이블 메타데이터와 엔진을 사용하여 MetaData.create_all()이라는 메서드를 통해 대상 SQLite 데이터베이스에 스키마를 한 번에 생성할 수 있다.

Base.metadata.create_all(engine)
BEGIN (implicit)
PRAGMA main.table_...info("user_account")
...
PRAGMA main.table_...info("address")
...
CREATE TABLE user_account (
    id INTEGER NOT NULL,
    name VARCHAR(30) NOT NULL,
    fullname VARCHAR,
    PRIMARY KEY (id)
)
...
CREATE TABLE address (
    id INTEGER NOT NULL,
    email_address VARCHAR NOT NULL,
    user_id INTEGER NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(user_id) REFERENCES user_account (id)
)
...
COMMIT

객체 생성 및 저장

User와 Address 클래스의 인스턴스를 생성하는데 자동으로 설정된 __init__() 메소드를 이미 가지고 있다. 이 인스턴스를 Session 객체를 사용하여 데이터베이스에 전달한다. Session.add_all() 메소드를 사용하여 객체를 한 번에 추가할 수도 있으며 Session.commit()은 변경 사항을 플러시하고 데이터베이스에 커밋하는 데 사용된다.

from sqlalchemy.orm import Session

with Session(engine) as session:
    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="spongebob@sqlalchemy.org")],
    )
    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="sandy@sqlalchemy.org"),
            Address(email_address="sandy@squirrelpower.org"),
        ],
    )
    patrick = User(name="patrick", fullname="Patrick Star")

    session.add_all([spongebob, sandy, patrick])

    session.commit()

실행결과

BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('spongebob', 'Spongebob Squarepants')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('sandy', 'Sandy Cheeks')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('patrick', 'Patrick Star')
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('spongebob@sqlalchemy.org', 1)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('sandy@sqlalchemy.org', 2)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('sandy@squirrelpower.org', 2)
COMMIT

세션(Session)은 위와 같이 컨텍스트 매니저 스타일로 사용하는 것이 권장된다. 즉, Python의 with: 문을 사용하는 것이다. Session 객체는 활성화된 데이터베이스 자원을 나타내므로 일련의 작업이 완료되면 세션을 닫는 것이 좋다.

단순 조회

아래는 데이터베이스에서 몇 개의 행을 선택해 객체를 로드하는 SELECT 문을 생성하는 가장 간단한 방법이다. SELECT 문을 생성하기 위해 select() 함수를 사용하여 새로운 Select 객체를 만들고, 이를 Session을 사용하여 호출한다. ORM 객체를 쿼리할 때 유용한 방법 중 하나는 Session.scalars() 메서드인데, 이 메서드는 우리가 선택한 ORM 객체들을 순회할 수 있는 ScalarResult 객체를 반환한다.

from sqlalchemy import select

session = Session(engine)

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):
    print(user)
BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name IN (?, ?)
[...] ('spongebob', 'sandy')

위 쿼리는 여러 WHERE 조건이 자동으로 AND로 연결되는 방식과 SQLAlchemy의 열과 유사한 객체를 사용하여 "동등성" 비교를 생성하는 방법을 보여준다. 이는 오버라이드된 Python 메서드인 ColumnOperators.__eq__()을 사용하여 SQL 조건 객체를 생성한다.

JOIN으로 조회

여러 테이블을 동시에 쿼리하는 것은 매우 일반적이며, SQL에서는 JOIN 키워드가 이를 수행하는 주요 방법이다. Select 구조체는 Select.join() 메서드를 사용하여 조인을 생성한다.

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "sandy@sqlalchemy.org")
 )
sandy_address = session.scalars(stmt).one()
SELECT address.id, address.email_address, address.user_id
FROM address JOIN user_account ON user_account.id = address.user_id
WHERE user_account.name = ? AND address.email_address = ?
[...] ('sandy', 'sandy@sqlalchemy.org')
sandy_address
Address(id=2, email_address='sandy@sqlalchemy.org')

위 쿼리는 여러 WHERE 조건이 자동으로 AND로 연결되는 방식을 보여주며, SQLAlchemy의 열과 유사한 객체를 사용하여 "동등성" 비교를 생성하는 방법을 설명한다. 이 비교는 오버라이드된 Python 메서드인 ColumnOperators.__eq__()를 사용하여 SQL 조건 객체를 생성한다.

변경

Session 객체는 ORM으로 매핑된 User 및 Address 클래스와 함께 객체에 대한 변경 사항을 자동으로 추적하여, Session이 다음에 플러시될 때 SQL 문을 생성한다. 아래 예시에서는 "sandy"와 연결된 이메일 주소 중 하나를 변경하고, SELECT 문을 통해 "patrick"의 행을 가져온 후 "patrick"에게 새로운 이메일 주소를 추가한다.

stmt = select(User).where(User.name == "patrick")
patrick = session.scalars(stmt).one()
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('patrick',)
patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (3,)
sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"

session.commit()
UPDATE address SET email_address=? WHERE address.id = ?
[...] ('sandy_cheeks@sqlalchemy.org', 2)
INSERT INTO address (email_address, user_id) VALUES (?, ?)
[...] ('patrickstar@sqlalchemy.org', 3)
COMMIT

patrick.addresses에 접근할 때 SELECT 문이 실행된 것을 확인할 수 있다. 이것을 지연 로드(lazy load)라고 한다.

삭제

"sandy" 사용자에게 연결된 Address 객체 중 하나를 제거하자. Session이 다음에 플러시될 때, 이 행은 삭제된다. 이 동작은 우리가 매핑에서 설정한 연쇄 삭제(delete cascade)로 인해 발생한다. Session.get()을 사용하여 기본 키로 sandy 객체를 가져온 후, 이 객체와 작업할 수 있다.

sandy = session.get(User, 2)
BEGIN (implicit)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name, user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (2,)
sandy.addresses.remove(sandy_address)
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (2,)

위의 마지막 SELECT는 지연 로드 작업이 진행되어 sandy.addresses 컬렉션이 로드되고, 이를 통해 sandy_address 멤버를 제거할 수 있도록 한 것다. 이 일련의 작업을 수행하는 다른 방법들도 있으며, 이러한 방법들은 SQL을 덜 생성한다.

Session.flush() 메서드를 사용하면, 트랜잭션을 커밋하지 않고도 변경될 예정인 내용에 대해 DELETE SQL을 실행할 수 있다.

session.flush()
DELETE FROM address WHERE address.id = ?
[...] (2,)

다음으로, "patrick" 사용자를 완전히 삭제해보자. 개별 객체를 최상위에서 삭제할 때는 Session.delete() 메서드를 사용한다. 이 메서드는 실제로 삭제를 수행하지는 않지만, 다음 플러시 시 객체가 삭제되도록 설정한다. 이 작업은 우리가 설정한 연쇄(cascade) 옵션에 따라 관련된 객체들, 이 경우에는 연결된 Address 객체들에도 연쇄적으로 영향을 미친다.

session.delete(patrick)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name, user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (3,)
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (3,)

이 경우 Session.delete() 메서드는 DELETE 문을 생성하지 않았음에도 불구하고 두 개의 SELECT 문을 실행했다. 이유는 메서드가 객체를 검사할 때, patrick 객체가 만료(expired)된 것으로 나타났기 때문이다. 이는 마지막으로 Session.commit()을 호출했을 때 발생한 것으로, SQL 문은 새로운 트랜잭션에서 행을 다시 로드하기 위한 것이다. 이 만료는 선택 사항이며, 일반적으로 적용되지 않는 상황에서는 이를 비활성화하는 경우가 많다.

삭제된 행을 보여주기 위해, 커밋을 예로 들겠다:

session.commit()
DELETE FROM address WHERE address.id = ?
[...] (4,)
DELETE FROM user_account WHERE user_account.id = ?
[...] (3,)
COMMIT

출처

반응형
저작자표시
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유
python 관련 글
글 더보기

티스토리툴바