ORM이란
ORM(Object-Relational Mapping)은 DB 테이블을 Python 클래스로, 행(row)을 객체로 다루게 해주는 기술이다.
ORM 없이 PostgreSQL을 쓰면 이렇게 된다.
cursor.execute("""
INSERT INTO users (email, username) VALUES (%s, %s)
""", ("hello@example.com", "goran"))
SQLAlchemy ORM을 쓰면 이렇게 된다.
user = User(email="hello@example.com", username="goran")
session.add(user)
session.commit()
SQL을 직접 쓰지 않아도 되고, Python 객체처럼 다룰 수 있어서 코드가 일관성 있어진다. 타입 힌트도 붙일 수 있어서 IDE 자동완성도 된다.
설치
pip install sqlalchemy psycopg2-binary
# Poetry 쓴다면
poetry add sqlalchemy psycopg2-binary
psycopg2-binary는 PostgreSQL 드라이버다. SQLAlchemy는 DB에 직접 연결하지 않고, 이런 드라이버를 통해 연결한다.
전체 구조 파악
SQLAlchemy ORM을 쓰려면 세 가지가 필요하다.
- Base — 모든 모델이 상속할 베이스 클래스
- Model — 테이블을 정의하는 Python 클래스
- Session — DB와 실제로 통신하는 객체
하나씩 만들어보자.
Base 클래스 정의
# app/database/base.py
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
Base는 모든 ORM 모델의 부모 클래스다. 직접 쓸 일은 없고, 모든 모델이 이걸 상속한다. SQLAlchemy가 Base를 통해 어떤 클래스가 테이블인지 파악한다.
모델(테이블) 정의
Base를 상속한 클래스가 DB 테이블이 된다. User 테이블을 예시로 보자.
# app/entity/user.py
import uuid
from datetime import datetime
from sqlalchemy import String, Boolean, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
class User(Base):
__tablename__ = "users" # 실제 DB 테이블 이름
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
email: Mapped[str] = mapped_column(String(255), unique=True)
username: Mapped[str] = mapped_column(String(50), unique=True)
bio: Mapped[str | None] = mapped_column(String(500))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow
)
Mapped[T]와 mapped_column()
컬럼 하나를 정의하는 방식이다.
email: Mapped[str] = mapped_column(String(255), unique=True)
- Mapped[str] — Python 타입 힌트. 이 컬럼이 str 타입이고 NOT NULL이라는 뜻
- mapped_column(...) — DB 수준 설정. 길이 제한, unique, default 등
nullable 여부는 Mapped[T]로 결정된다.
bio: Mapped[str | None] = mapped_column(String(500)) # NULL 허용
email: Mapped[str] = mapped_column(String(255)) # NOT NULL
str | None이면 NULL 허용, str이면 NOT NULL이다. DB 제약과 Python 타입 힌트가 한 줄에서 동시에 선언된다.
Enum 컬럼 처리
상태값처럼 정해진 값만 들어가는 컬럼은 Enum으로 정의하면 좋다.
# app/entity/enums.py
import enum
class PostStatus(str, enum.Enum):
DRAFT = "DRAFT"
PUBLISHED = "PUBLISHED"
ARCHIVED = "ARCHIVED"
str, enum.Enum을 같이 상속하면 .value를 따로 안 꺼내도 문자열처럼 동작한다.
# app/entity/post.py
import uuid
from datetime import datetime
from sqlalchemy import String, Text, ForeignKey, DateTime
from sqlalchemy import Enum as SAEnum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.entity.enums import PostStatus
class Post(Base):
__tablename__ = "posts"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
author_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
title: Mapped[str] = mapped_column(String(300))
content: Mapped[str | None] = mapped_column(Text)
status: Mapped[PostStatus] = mapped_column(SAEnum(PostStatus), default=PostStatus.DRAFT)
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow
)
author: Mapped["User"] = relationship(back_populates="posts")
SAEnum(PostStatus)으로 컬럼을 선언한다. PostgreSQL에서는 실제 ENUM 타입으로 생성되기 때문에 DB 수준에서도 유효값이 강제된다.
테이블 간 관계 — relationship()
User와 Post는 1:N 관계다. 유저 한 명이 게시글을 여러 개 쓸 수 있다.
# User 모델에 추가
posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")
# Post 모델에는 이미 있음
author: Mapped["User"] = relationship(back_populates="posts")
- Mapped[list["Post"]] → User 입장에서 여러 Post를 가짐 (1:N)
- Mapped["User"] → Post 입장에서 하나의 User를 가짐 (N:1)
- back_populates → 양방향 관계를 연결. 한쪽을 수정하면 반대쪽도 자동으로 반영됨
- cascade="all, delete-orphan" → User 삭제 시 연결된 Post도 같이 삭제
relationship()은 실제 FK 컬럼(author_id)과 별개다. FK는 ForeignKey("users.id")로 DB 연결을 담당하고, relationship()은 Python 객체 수준에서 편하게 접근하게 해준다.
# 관계를 통한 접근 예시
user = session.get(User, user_id)
print(user.posts) # Post 리스트
print(user.posts[0].author) # 다시 User 객체
Engine과 Session
모델을 정의했으면 실제 DB와 연결이 필요하다. 이때 등장하는 개념이 Engine과 Session이다.
Session이란
Session은 Python 코드와 DB 사이의 중간 관리자다. 두 가지 역할을 동시에 한다.
역할 1 — 객체 추적 (Identity Map)
Session은 자신이 불러온 객체들을 내부적으로 기억하고 있다. 같은 Session 안에서 같은 유저를 두 번 조회하면 DB를 두 번 치는 게 아니라 첫 번째 결과를 그대로 돌려준다. 또 객체의 속성을 바꾸면 Session이 "이 객체가 변경됐다"는 걸 감지해서 commit() 시 자동으로 UPDATE 쿼리를 날린다.
with SessionLocal() as session:
user1 = session.get(User, user_id) # DB 조회
user2 = session.get(User, user_id) # 캐시에서 반환 (DB 안 침)
print(user1 is user2) # True — 같은 객체
역할 2 — 트랜잭션 관리
session.commit()을 호출하기 전까지 모든 변경사항은 트랜잭션 안에 묶인다. 중간에 예외가 나면 전부 롤백되고, commit()이 성공해야 DB에 최종 반영된다.
with SessionLocal() as session:
session.add(User(email="a@a.com", username="a"))
session.add(User(email="b@b.com", username="b"))
# 여기서 예외 발생 시 위 두 INSERT 모두 롤백
session.commit() # 둘 다 성공해야 반영
Engine — 왜 하나만 만드나
Engine은 내부적으로 커넥션 풀(Connection Pool) 을 들고 있다. DB에 연결하는 건 TCP 소켓을 새로 열고 인증하는 과정이라 비용이 크다. 그래서 매 요청마다 연결을 새로 만들지 않고, 미리 만들어둔 연결을 재사용한다.
앱 시작 시 Engine 생성
└── 커넥션 풀 (연결 5개 미리 생성해둠)
├── Connection 1
├── Connection 2
└── Connection 3 ...
요청 올 때마다 → 풀에서 연결 하나 빌려서 사용 → 반납
engine을 여러 개 만들면 커넥션 풀도 여러 개가 생겨서 연결이 낭비된다. 그래서 앱 전체에서 하나만 만들고 공유한다.
Session vs Connection
Connection Session
| 역할 | 실제 DB 소켓 연결 | 트랜잭션 + 객체 추적 |
| 관리 주체 | Engine (커넥션 풀) | 개발자가 직접 생성/종료 |
| 수명 | 풀에서 빌렸다 반납 | 요청 단위 또는 작업 단위 |
Session은 Connection을 직접 들고 있지 않는다. 쿼리를 날릴 때 Engine의 풀에서 Connection을 잠깐 빌리고, 쿼리가 끝나면 반납한다. 그래서 Session이 열려 있다고 Connection이 계속 점유되는 게 아니다.
코드로 보기
# app/database/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import os
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+psycopg2://user:pass@localhost/mydb")
# Engine — 앱 전체에서 하나만 생성
engine = create_engine(DATABASE_URL, echo=False)
# SessionLocal — Session 객체를 찍어내는 팩토리
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
autocommit=False, autoflush=False 옵션
- autocommit=False (기본값, 권장)
False면 명시적으로 session.commit()을 호출해야 DB에 반영된다. 중간에 예외가 나면 자동으로 롤백된다.
True로 하면 쿼리 하나하나가 즉시 커밋된다. 실수로 잘못된 데이터를 넣어도 롤백할 수 없어서 실무에선 거의 안 쓴다.
- autoflush=False (권장)
flush는 Session이 들고 있는 변경사항을 DB에 SQL로 전송하는 것이다. 아직 커밋은 아니고, DB에 보내기만 한 상태다.
True면 session.execute() 같은 조회 쿼리를 날리기 직전에 자동으로 flush가 일어난다. 의도치 않은 시점에 SQL이 나가서 디버깅이 어려워질 수 있어서 False로 두고 직접 제어하는 게 낫다.
Session 생명주기 — add → flush → commit
Session 안에서 데이터가 어떤 단계를 거치는지 이해하면 동작 방식이 명확해진다.
with SessionLocal() as session:
user = User(email="hello@example.com", username="goran")
# 1. add — Session이 이 객체를 추적 시작. DB엔 아무것도 안 감
session.add(user)
print(user.id) # None (아직 DB 미반영)
# 2. flush — SQL을 DB에 전송. 트랜잭션은 아직 열려있음
# commit 전이라 다른 세션에선 이 데이터가 안 보임
session.flush()
print(user.id) # UUID 생성됨 (DB가 채워줬으니까)
# 3. commit — 트랜잭션 확정. 이제 다른 세션에서도 보임
session.commit()
# 4. refresh — commit 후 DB에서 최신 상태 다시 조회
session.refresh(user)
보통은 flush()를 직접 쓸 일이 많지 않다. commit() 직전에 자동으로 flush가 일어나기 때문이다. flush가 필요한 케이스는 같은 트랜잭션 안에서 INSERT 후 바로 그 ID를 참조해야 할 때 정도다.
Session 사용 패턴
from app.database.session import SessionLocal
# with 블록으로 사용 — 블록 종료 시 자동으로 close()
with SessionLocal() as session:
user = User(email="hello@example.com", username="goran")
session.add(user)
session.commit()
session.refresh(user) # commit 후 DB에서 최신값 재조회
print(user.id) # UUID가 채워진 상태
with 블록을 벗어나면 session.close()가 자동으로 호출된다. close는 Session이 빌렸던 Connection을 풀에 반납하는 동작이다. 예외가 나도 close는 보장된다.
session.commit()을 호출해야 실제 DB에 반영된다. 호출하지 않고 블록을 벗어나면 롤백된다.
데이터 조회 — select()
from sqlalchemy import select
with SessionLocal() as session:
# 전체 조회
stmt = select(Post)
posts = list(session.scalars(stmt))
# 조건 조회
stmt = (
select(Post)
.where(Post.status == PostStatus.PUBLISHED)
.order_by(Post.created_at.desc())
.limit(10)
)
recent_posts = list(session.scalars(stmt))
session.scalars(stmt) — ORM 모델 인스턴스를 바로 반환한다. list()로 감싸면 리스트가 된다.
단건 조회는 session.get()이 편하다.
with SessionLocal() as session:
user = session.get(User, user_id) # 없으면 None 반환
데이터 수정과 삭제
with SessionLocal() as session:
user = session.get(User, user_id)
# 수정 — 속성을 바꾸고 commit
user.bio = "새로운 소개"
session.commit()
# 삭제
session.delete(user)
session.commit()
session.add()를 다시 호출할 필요 없다. Session이 추적 중인 객체의 변경을 감지해서 commit() 시 자동으로 UPDATE 쿼리를 날린다.
PostgreSQL Upsert
같은 데이터를 반복적으로 넣어야 하는 상황(수집, 동기화 등)에서 유용하다. 이미 있으면 UPDATE, 없으면 INSERT하는 패턴이다.
from sqlalchemy.dialects.postgresql import insert as pg_insert
with SessionLocal() as session:
stmt = (
pg_insert(User)
.values(email="hello@example.com", username="goran")
.on_conflict_do_update(
index_elements=["email"], # 충돌 기준 컬럼 (unique constraint)
set_={"username": "goran_updated"}, # 충돌 시 업데이트할 값
)
.returning(User.id)
)
result = session.execute(stmt)
session.commit()
user_id = result.scalar_one()
index_elements에는 unique constraint가 걸린 컬럼을 적는다. 여기선 email이 겹칠 때 username을 업데이트한다.
테이블 생성
모델을 다 정의했으면 실제 DB에 테이블을 만들어야 한다.
from app.database.base import Base
from app.database.session import engine
from app.entity import user, post # 모든 모델 import 필수
Base.metadata.create_all(engine)
이 코드를 실행하면 Base에 등록된 모든 모델을 기준으로 테이블이 생성된다.
단, 실무에서는 이 방식보다 Alembic을 써서 마이그레이션으로 관리하는 게 일반적이다. 테이블을 통째로 다시 만드는 게 아니라, 변경사항만 추적해서 적용할 수 있기 때문이다.
정리
SQLAlchemy ORM의 흐름을 정리하면:
- Base 클래스를 만들고
- Base를 상속한 클래스로 테이블을 정의 (Mapped[T] + mapped_column())
- Engine으로 DB에 연결하고
- Session으로 데이터를 읽고 쓴다
Mapped[T]의 핵심은 타입 힌트와 nullable 여부가 한 곳에서 선언된다는 것이다. Mapped[str]이면 NOT NULL, Mapped[str | None]이면 NULL 허용.
'Python' 카테고리의 다른 글
| Prefect - 오케스트레이션 (0) | 2026.04.07 |
|---|---|
| Alembic - 스키마 버전 관리 (0) | 2026.04.07 |