Alert

์ด ๊ธ€์€ Claude Code์˜ ๋„์›€์„ ๋ฐ›์•„ ์ž‘์„ฑ๋˜์—ˆ์Šต๋‹ˆ๋‹ค

TL;DR

  • SQLAlchemy๋Š” Python์˜ ์‚ฌ์‹ค์ƒ ํ‘œ์ค€ ORM
  • Core(SQL ํ‘œํ˜„์‹)์™€ ORM(๊ฐ์ฒด ๋งคํ•‘) ๋‘ ๋ ˆ์ด์–ด๋กœ ๊ตฌ์„ฑ
  • 2.0 ์Šคํƒ€์ผ์—์„œ select() ๊ธฐ๋ฐ˜ ์ฟผ๋ฆฌ๊ฐ€ ๊ธฐ๋ณธ
  • Alembic์œผ๋กœ ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜ ๊ด€๋ฆฌ

SQLAlchemy?

  • Python์—์„œ ๊ฐ€์žฅ ๋„๋ฆฌ ์“ฐ์ด๋Š” ๋ฐ์ดํ„ฐ๋ฒ ์ด์Šค ํˆดํ‚ท ๊ฒธ ORM
  • 2006๋…„ ์ฒซ ๋ฆด๋ฆฌ์Šค, ํ˜„์žฌ 2.x ๋ฒ„์ „
  • PostgreSQL, MySQL, SQLite ๋“ฑ ์ฃผ์š” RDBMS ์ง€์›
  • https://www.sqlalchemy.org/

1. ORM์ด๋ž€

ORM(Object-Relational Mapping)์€ ๋ฐ์ดํ„ฐ๋ฒ ์ด์Šค ํ…Œ์ด๋ธ”์„ Python ํด๋ž˜์Šค๋กœ ๋งคํ•‘ํ•˜๋Š” ๊ธฐ๋ฒ•์ด๋‹ค. SQL์„ ์ง์ ‘ ์“ฐ์ง€ ์•Š๊ณ  Python ์ฝ”๋“œ๋กœ ๋ฐ์ดํ„ฐ๋ฒ ์ด์Šค๋ฅผ ๋‹ค๋ฃฐ ์ˆ˜ ์žˆ๋‹ค.

# SQL๋กœ ์ง์ ‘ ์ž‘์„ฑ
cursor.execute("SELECT * FROM users WHERE age > 20")
 
# ORM์œผ๋กœ ์ž‘์„ฑ
session.query(User).filter(User.age > 20).all()

ORM์˜ ์žฅ๋‹จ์ 

  • ์žฅ์ : ๋ฐ์ดํ„ฐ๋ฒ ์ด์Šค ์ข…๋ฅ˜์— ๋…๋ฆฝ์ , ์ฝ”๋“œ ๊ฐ€๋…์„ฑ ํ–ฅ์ƒ, SQL ์ธ์ ์…˜ ๋ฐฉ์ง€
  • ๋‹จ์ : ๋ณต์žกํ•œ ์ฟผ๋ฆฌ์—์„œ ์„ฑ๋Šฅ ์˜ค๋ฒ„ํ—ค๋“œ ๊ฐ€๋Šฅ, SQL ์ž์ฒด๋ฅผ ๋ชจ๋ฅด๋ฉด ๋””๋ฒ„๊น… ์–ด๋ ค์›€

2. Python ORM ์„ ํƒ์ง€

ORMํŠน์ง•
SQLAlchemy๋…๋ฆฝํ˜• ํ‘œ์ค€. Flask, FastAPI ๋“ฑ ํ”„๋ ˆ์ž„์›Œํฌ ๊ฐ€๋ฆฌ์ง€ ์•Š๊ณ  ์‚ฌ์šฉ
Django ORMDjango ํ”„๋ ˆ์ž„์›Œํฌ ๋‚ด์žฅ. Django ๋ฐ–์—์„œ๋Š” ์‚ฌ์šฉํ•˜๊ธฐ ๋ถˆํŽธํ•จ
SQLModelFastAPI ์ œ์ž‘์ž(Sebastian Ramirez)๊ฐ€ ๋งŒ๋“  ๋ผ์ด๋ธŒ๋Ÿฌ๋ฆฌ. ๋‚ด๋ถ€์ ์œผ๋กœ SQLAlchemy + Pydantic ๋ž˜ํผ
Peewee๊ฒฝ๋Ÿ‰ ORM. ์†Œ๊ทœ๋ชจ ํ”„๋กœ์ ํŠธ์— ์ ํ•ฉ
Tortoise ORMasync ๋„ค์ดํ‹ฐ๋ธŒ ORM. asyncio ๊ธฐ๋ฐ˜ ํ”„๋กœ์ ํŠธ์—์„œ ์‚ฌ์šฉ

Django๋ฅผ ์“ฐ๋ฉด Django ORM, ๊ทธ ์™ธ์—๋Š” SQLAlchemy๊ฐ€ ๊ธฐ๋ณธ ์„ ํƒ์ด๋‹ค. SQLModel๋„ ๊ฒฐ๊ตญ SQLAlchemy๋ฅผ ๊ธฐ๋ฐ˜์œผ๋กœ ๋™์ž‘ํ•˜๋ฏ€๋กœ, SQLAlchemy๋ฅผ ์ดํ•ดํ•˜๋ฉด SQLModel๋„ ์ž์—ฐ์Šค๋Ÿฝ๊ฒŒ ์“ธ ์ˆ˜ ์žˆ๋‹ค.


3. ์„ค์น˜

# SQLAlchemy๋งŒ ์„ค์น˜
pip install sqlalchemy
 
# uv ์‚ฌ์šฉ ์‹œ
uv add sqlalchemy
 
# ๋ฐ์ดํ„ฐ๋ฒ ์ด์Šค ๋“œ๋ผ์ด๋ฒ„๋„ ํ•จ๊ป˜ ์„ค์น˜ (PostgreSQL ์˜ˆ์‹œ)
pip install sqlalchemy psycopg2-binary
 
# ๋น„๋™๊ธฐ ๋“œ๋ผ์ด๋ฒ„ (asyncio ์‚ฌ์šฉ ์‹œ)
pip install sqlalchemy asyncpg

4. ๋‘ ๊ฐœ์˜ ๋ ˆ์ด์–ด

SQLAlchemy๋Š” Core์™€ ORM ๋‘ ๊ณ„์ธต์œผ๋กœ ๋‚˜๋‰œ๋‹ค.

โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”
โ”‚       ORM           โ”‚  โ† ํด๋ž˜์Šค๋กœ ํ…Œ์ด๋ธ” ๋งคํ•‘, ์„ธ์…˜ ๊ด€๋ฆฌ
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚       Core          โ”‚  โ† SQL ํ‘œํ˜„์‹, ์—”์ง„, ์ปค๋„ฅ์…˜ ํ’€
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚      DBAPI          โ”‚  โ† psycopg2, sqlite3 ๋“ฑ ๋“œ๋ผ์ด๋ฒ„
โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜
  • Core: SQL์„ Python ํ‘œํ˜„์‹์œผ๋กœ ์ž‘์„ฑํ•˜๋Š” ๋ ˆ์ด์–ด. select(), insert() ๊ฐ™์€ ํ•จ์ˆ˜๋ฅผ ์‚ฌ์šฉํ•œ๋‹ค.
  • ORM: Core ์œ„์—์„œ Python ํด๋ž˜์Šค์™€ ํ…Œ์ด๋ธ”์„ ๋งคํ•‘ํ•˜๋Š” ๋ ˆ์ด์–ด. ๋Œ€๋ถ€๋ถ„ ์ด ORM ๋ ˆ์ด์–ด๋ฅผ ์‚ฌ์šฉํ•œ๋‹ค.

๋‘ ๋ ˆ์ด์–ด๋ฅผ ์„ž์–ด ์“ธ ์ˆ˜ ์žˆ๋‹ค๋Š” ์ ์ด SQLAlchemy์˜ ๊ฐ•์ ์ด๋‹ค. ORM์ด ๋ถˆํŽธํ•œ ๋ณต์žกํ•œ ์ฟผ๋ฆฌ๋Š” Core๋กœ ์ง์ ‘ ์ž‘์„ฑํ•  ์ˆ˜ ์žˆ๋‹ค.


5. ๊ธฐ๋ณธ ์‚ฌ์šฉ๋ฒ• (2.0 ์Šคํƒ€์ผ)

SQLAlchemy 2.0๋ถ€ํ„ฐ select() ๊ธฐ๋ฐ˜ ์ฟผ๋ฆฌ๊ฐ€ ํ‘œ์ค€์ด๋‹ค. ์ด์ „ 1.x์˜ session.query() ๋ฐฉ์‹๋„ ๋™์ž‘ํ•˜์ง€๋งŒ ์ƒˆ ํ”„๋กœ์ ํŠธ์—์„œ๋Š” 2.0 ์Šคํƒ€์ผ์„ ๊ถŒ์žฅํ•œ๋‹ค.

์—”์ง„ ์ƒ์„ฑ
from sqlalchemy import create_engine
 
# SQLite
engine = create_engine("sqlite:///app.db")
 
# PostgreSQL
engine = create_engine("postgresql://user:password@localhost:5432/mydb")
๋ชจ๋ธ ์ •์˜
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String
 
 
class Base(DeclarativeBase):
    pass
 
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)
    age: Mapped[int | None]  # nullable ์ปฌ๋Ÿผ

2.0์˜ ํƒ€์ž… ํžŒํŠธ ์Šคํƒ€์ผ

Mapped[int]์ฒ˜๋Ÿผ Python ํƒ€์ž… ํžŒํŠธ๋กœ ์ปฌ๋Ÿผ ํƒ€์ž…์„ ์„ ์–ธํ•œ๋‹ค. 1.x์˜ Column(Integer) ๋ฐฉ์‹๋ณด๋‹ค IDE ์ง€์›์ด ์ข‹๊ณ  ์ฝ”๋“œ๊ฐ€ ๊ฐ„๊ฒฐํ•˜๋‹ค.

ํ…Œ์ด๋ธ” ์ƒ์„ฑ
Base.metadata.create_all(engine)
CRUD ๊ธฐ๋ณธ
from sqlalchemy.orm import Session
from sqlalchemy import select
 
# Create
with Session(engine) as session:
    user = User(name="ํ™๊ธธ๋™", email="hong@example.com", age=30)
    session.add(user)
    session.commit()
 
# Read
with Session(engine) as session:
    stmt = select(User).where(User.age > 20)
    users = session.scalars(stmt).all()
 
# Update
with Session(engine) as session:
    stmt = select(User).where(User.name == "ํ™๊ธธ๋™")
    user = session.scalars(stmt).first()
    user.age = 31
    session.commit()
 
# Delete
with Session(engine) as session:
    stmt = select(User).where(User.name == "ํ™๊ธธ๋™")
    user = session.scalars(stmt).first()
    session.delete(user)
    session.commit()

6. ๊ด€๊ณ„(Relationship) ์„ค์ •

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
 
 
class Post(Base):
    __tablename__ = "posts"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
 
    author: Mapped["User"] = relationship(back_populates="posts")
 
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
 
    posts: Mapped[list["Post"]] = relationship(back_populates="author")
# ์‚ฌ์šฉ ์˜ˆ์‹œ
with Session(engine) as session:
    user = session.scalars(select(User).where(User.name == "ํ™๊ธธ๋™")).first()
    for post in user.posts:
        print(post.title)

7. Alembic์œผ๋กœ ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜

Alembic์€ SQLAlchemy์˜ ๊ณต์‹ ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜ ๋„๊ตฌ๋‹ค. ๋ชจ๋ธ์ด ๋ณ€๊ฒฝ๋˜๋ฉด Alembic์ด DDL(ALTER TABLE ๋“ฑ)์„ ์ž๋™ ์ƒ์„ฑํ•œ๋‹ค.

# ์„ค์น˜
pip install alembic
 
# ์ดˆ๊ธฐํ™”
alembic init alembic
 
# ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜ ํŒŒ์ผ ์ž๋™ ์ƒ์„ฑ
alembic revision --autogenerate -m "add users table"
 
# ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜ ์ ์šฉ
alembic upgrade head
 
# ๋กค๋ฐฑ
alembic downgrade -1

autogenerate ์ฃผ์˜์‚ฌํ•ญ

  • --autogenerate๋Š” ๋ชจ๋ธ๊ณผ ์‹ค์ œ DB ์Šคํ‚ค๋งˆ๋ฅผ ๋น„๊ตํ•ด์„œ ์ฐจ์ด๋ฅผ ๊ฐ์ง€ํ•œ๋‹ค
  • ์ปฌ๋Ÿผ ์ด๋ฆ„ ๋ณ€๊ฒฝ์€ ์ž๋™ ๊ฐ์ง€๊ฐ€ ์•ˆ ๋จ (์‚ญ์ œ + ์ƒ์„ฑ์œผ๋กœ ์ธ์‹)
  • ์ƒ์„ฑ๋œ ๋งˆ์ด๊ทธ๋ ˆ์ด์…˜ ํŒŒ์ผ์€ ๋ฐ˜๋“œ์‹œ ๊ฒ€ํ†  ํ›„ ์ ์šฉํ•  ๊ฒƒ

8. ์ž์ฃผ ์“ฐ๋Š” ์ฟผ๋ฆฌ ํŒจํ„ด

from sqlalchemy import select, func, or_, desc
 
# ์—ฌ๋Ÿฌ ์กฐ๊ฑด (AND)
stmt = select(User).where(User.age > 20, User.name.like("%ํ™%"))
 
# OR ์กฐ๊ฑด
stmt = select(User).where(or_(User.age > 30, User.name == "ํ™๊ธธ๋™"))
 
# ์ •๋ ฌ
stmt = select(User).order_by(desc(User.age))
 
# ํŽ˜์ด์ง€๋„ค์ด์…˜
stmt = select(User).offset(10).limit(20)
 
# ์ง‘๊ณ„
stmt = select(func.count()).select_from(User)
count = session.scalar(stmt)
 
# JOIN
stmt = select(User, Post).join(Post, User.id == Post.user_id)