[번역] Python의 ORM - SQLAlchemy 튜토리얼

WEB BE Repository/Python + FastAPI

[번역] Python의 ORM - SQLAlchemy 튜토리얼

조금씩 차근차근 2026. 1. 5. 19:00

ORM 빠른 시작

기본 ORM 사용 형태를 빠르게 보고 싶은 신규 사용자를 위해, SQLAlchemy Unified Tutorial에서 사용하는 매핑과 예제를 축약해 보여준다.

이 섹션의 설명은 의도적으로 매우 짧게 작성되어 있으므로, 여기서 다루는 각 개념에 대한 더 자세한 설명은 SQLAlchemy Unified Tutorial을 참고하는 것이 권장된다.

모델 선언(Declare Models)

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

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

매핑은 Base(여기서는 DeclarativeBase를 상속한 클래스)에서 시작한다.
개별 매핑 클래스는 Base를 상속해 만들며, 일반적으로 하나의 DB 테이블을 나타낸다. 테이블명은 __tablename__ 클래스 속성으로 지정한다.
테이블 컬럼은 Mapped라는 특수 타입 애너테이션을 포함하는 속성으로 선언한다.
- 이때, 각 속성 이름이 곧 컬럼 이름이 된다.
- 컬럼의 SQL 타입은 Mapped에 연결된 파이썬 타입(int→INTEGER, str→VARCHAR 등)에서 우선적으로 유도되며, Optional[...] 사용 여부로 NULL 허용 여부가 결정된다.
- 더 구체적인 타입 정보는 mapped_column() 오른쪽에 SQLAlchemy 타입(예: String)을 지정해 표현할 수 있다.
- 파이썬 타입과 SQL 타입의 매핑은 “type annotation map”으로 커스터마이즈할 수 있다.
mapped_column()은 서버 기본값, 제약조건(기본키/외래키 포함) 등 컬럼 세부 옵션을 표현하는 데 사용한다. 또한 Core의 Column이 받는 인자들의 상위 집합을 받을 수 있다.
모든 ORM 매핑 클래스는 최소 1개 이상의 기본키 컬럼이 필요하다.
- 예: User.id, Address.id
테이블 이름과 컬럼 선언의 조합은 SQLAlchemy에서 테이블 메타데이터(table metadata) 라고 부른다.
- 위 매핑은 “Annotated Declarative Table” 구성의 예이다.
relationship()은 컬럼 기반 속성과 달리, ORM 클래스 간 연관을 나타낸다.
- 예제에서는 User.addresses가 User→Address, Address.user가 Address→User 연결을 의미하고 있다.
__repr__()는 필수는 아니지만 디버깅에 유용하다.
- dataclasses를 사용해 __repr__() 같은 메서드를 자동 생성하는 방식도 가능하다.

엔진 생성(Create an Engine)

Engine은 새 DB 커넥션을 만들 수 있는 팩토리이며, 빠른 재사용을 위해 커넥션 풀(Connection Pool) 에 커넥션을 보관한다.
학습 목적에서는 편의를 위해 메모리 기반 SQLite를 흔히 사용하니, SQLite를 사용해보자.

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

SQLite는 따로 직접 설치하자.

팁: echo=True는 커넥션이 발생시키는 SQL을 표준 출력으로 로깅하도록 지정한다.

CREATE TABLE DDL 생성(Emit CREATE TABLE DDL)

테이블 메타데이터와 엔진을 사용해, 대상 SQLite DB에 스키마를 한 번에 생성할 수 있다.
이는 MetaData.create_all()로 수행할 수 있다.

from startsqlalchemy.model import Base

Base.metadata.create_all(engine)

객체 생성 및 영속화(Create Objects and Persist)

이제 DB에 데이터를 삽입할 준비가 되었다!

선언적 매핑 과정에서 User, Address에는 이미 __init__()가 설정되어 있으므로, 객체 인스턴스를 만든 뒤 Session을 통해 DB로 전달하면 된다.
Session은 Engine을 사용해 DB와 상호작용한다.

여기서는 여러 객체를 한 번에 추가하기 위해 Session.add_all()을 사용하며, Session.commit()은 대기 중인 변경을 DB로 flush한 뒤 현재 트랜잭션을 commit해보자.

세션 사용 시 트랜잭션은 기본적으로 진행 중인 상태이다.

from sqlalchemy import create_engine

from startsqlalchemy.model import Base, Address, User

engine = create_engine('sqlite:///:memory:', echo=True)

Base.metadata.create_all(engine)

from sqlalchemy.orm import Session

with Session(engine) as session:
    spongbob = User(
        name = 'Sponge Bob',
        fullname = 'Sponge Bob Square Pants',
        addresses = [Address(email_address="[email protected]")],
    )

    sandy = User(
        name = 'Sandy',
        fullname = 'Sandy Cheeks',
        addresses = [
            Address(email_address="[email protected]"),
            Address(email_address="[email protected]")
        ],
    )
    patrick = User(name='Patrick', fullname='Patrick Star')

    session.add_all([spongbob, sandy, patrick])
    session.commit()

팁: 위처럼 with 문을 사용한 컨텍스트 매니저 스타일로 Session을 사용하는 것이 권장된다.
Session은 활성 DB 리소스를 나타내므로, 작업 묶음이 끝나면 닫히도록 하는 것이 바람직하다.

단순 SELECT(Simple SELECT)

데이터가 들어간 뒤, ORM 객체를 로드하는 가장 단순한 SELECT 형태는 다음과 같다.
SELECT 문은 select() 함수로 Select 객체를 만들고, 이를 Session으로 실행한다.
ORM 객체를 조회할 때는 Session.scalars()가 유용하며, 이는 선택된 ORM 객체를 순회하는 ScalarResult를 반환한다.

from sqlalchemy import select
from sqlalchemy.orm import Session

import startsqlalchemy.createandpersist as engine_module
from startsqlalchemy.model import User, Address

session = Session(engine_module.engine)

stmt = select(User).where(User.name.in_(["Sponge Bob", "Sandy"]))

for user in session.scalars(stmt):
    print(user)

위 쿼리는 Select.where()로 WHERE 조건을 추가했고, ColumnOperators.in_()를 사용해 SQL의 IN 연산자를 구성했다.

JOIN을 포함한 SELECT(SELECT with JOIN)

여러 테이블을 함께 조회하는 것은 일반적이며, SQL에서는 JOIN이 핵심 수단이다.
SQLAlchemy에서는 Select.join()으로 조인을 구성할 수 있다.

from sqlalchemy import select
from sqlalchemy.orm import Session

import startsqlalchemy.createandpersist as engine_module
from startsqlalchemy.model import User, Address

session = Session(engine_module.engine)

stmt = select(User).where(User.name.in_(["Sponge Bob", "Sandy"]))

for user in session.scalars(stmt):
    print(user)


stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "Sandy")
    .where(Address.email_address == "[email protected]")
)

sandy_address = session.scalars(stmt).first()
print(sandy_address)

이 예시는 여러 WHERE 조건이 자동으로 AND로 연결되는 점, 그리고 컬럼 유사 객체에 대해 “동등 비교”를 만들 때 ColumnOperators.__eq__()가 오버라이드되어 SQL 조건 객체를 생성한다는 점을 보여준다.

변경 사항 반영(Make Changes)

Session은 ORM 매핑 클래스(User, Address)와 함께 객체 변경을 자동으로 추적하며, 다음 flush 시점에 반영될 SQL을 준비한다.
아래 예시는 “sandy”의 이메일 주소 하나를 변경하고, “patrick”에 새 이메일 주소를 추가한다(먼저 SELECT로 “patrick”을 조회).

from sqlalchemy import select
from sqlalchemy.orm import Session

import startsqlalchemy.createandpersist as engine_module
from startsqlalchemy.model import User, Address


session = Session(engine_module.engine)

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "Sandy")
    .where(Address.email_address == "[email protected]")
)

sandy_address = session.scalars(stmt).first()


stmt = select(User).where(User.name == "Patrick")
patrick = session.scalars(stmt).one()

patrick.addresses.append(Address(email_address="[email protected]"))

sandy_address.email_address = "[email protected]"

session.commit()

patrick.addresses에 접근할 때 SELECT가 수행되는데, 이는 지연 로딩(lazy load) 이다.
연관 컬렉션을 더 적거나 더 많은 SQL로 로딩하는 다양한 방식(로더 전략)은 좀 더 긴 튜토리얼에서 다룬다.

삭제(Some Deletes)

여기서는 사용 사례에 따라 중요한 두 가지 삭제 형태를 간단히 보여준다.

먼저 “sandy” 사용자의 Address 하나를 컬렉션에서 제거한다. 다음 flush에서 해당 행이 삭제됩니다. 이는 매핑에서 설정한 delete cascade 동작의 결과다. Session.get()으로 기본키로 “sandy”를 로드한 뒤 작업해보자.

from sqlalchemy import select
from sqlalchemy.orm import Session

import startsqlalchemy.createandpersist as engine_module
from startsqlalchemy.model import User, Address


session = Session(engine_module.engine)

sandy = session.get(User, 2)

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "Sandy")
    .where(Address.email_address == "[email protected]")
)

sandy_address = session.scalars(stmt).first()
sandy.addresses.remove(sandy_address)

이 과정에서 sandy.addresses 컬렉션을 로딩하기 위한 lazy load SELECT가 수행된다.
더 적은 SQL로 동일 목적을 달성하는 다른 방법들도 존재한다. (ex: fetch)

트랜잭션을 commit하지 않고, 현재까지 설정된 변경에 대한 DELETE SQL만 먼저 내보내려면 Session.flush()를 사용할 수 있다.

session.flush()

다음으로 “patrick” 사용자를 객체 단위로 삭제한다. 최상위 객체 삭제는 Session.delete()로 수행하며, 이 호출은 즉시 DELETE를 실행하는 대신 다음 flush에 삭제되도록 설정한다. 또한 매핑에 지정한 cascade 옵션에 따라 관련 Address 객체에도 cascade가 적용되게 된다.

from sqlalchemy import select
from sqlalchemy.orm import Session

import startsqlalchemy.createandpersist as engine_module
from startsqlalchemy.model import User, Address


session = Session(engine_module.engine)

sandy = session.get(User, 2)

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "Sandy")
    .where(Address.email_address == "[email protected]")
)

sandy_address = session.scalars(stmt).first()
sandy.addresses.remove(sandy_address)


stmt = select(User).where(User.name == "Patrick")
patrick = session.scalars(stmt).one()

session.delete(patrick)
session.commit()

위 개념을 더 깊게 학습하기

위 내용은 빠른 개요이며, 각 단계에는 다루지 않은 중요한 개념이 많다.
전체적인 형태를 확인한 뒤에는 SQLAlchemy Unified Tutorial을 따라가며 상단에서 사용된 동작이 무엇을 의미하는지에 대한 실질적 이해를 확보하는 것이 권장된다.

'WEB BE Repository > Python + FastAPI' 카테고리의 다른 글

[FastAPI] FastAPI 기본 컨트롤러 사용 방법 (0)	2026.01.03

현재글[번역] Python의 ORM - SQLAlchemy 튜토리얼

Go Gradually - Repository

목표: 핀잇 실사용자 1명 이상 확보하기 글은 비정기적/저녁 7시에 올라옵니다.

rabbitmq, PS, DQN, 리팩터링, 코루틴, LeetCode, math, 쿠버네티스, k8s, 도메인 이벤트, Kubernetes, Greedy, 오블완, DP, 누적 합, prefix sum, 마틴 파울러, DRL, CI/CD, 티스토리챌린지,

Today :
Yesterday :

일	월	화	수	목	금	토
				1	2	3
4	5	6	7	8	9	10
11	12	13	14	15	16	17
18	19	20	21	22	23	24
25	26	27	28	29	30	31

Go Gradually - Repository