SQLAlchemy 2.0 전격 해부: 왜 파이썬(Python) 최강 ORM으로 불릴까요?
SQLAlchemy 튜토리얼
파이썬(Python) 생태계에서 가장 인기 있는 객체 관계 매핑(ORM, Object Relational Mapping) 도구를 꼽으라면 단연 SQLAlchemy(에스큐엘알케미)일 텐데요.
SQLAlchemy(에스큐엘알케미)는 정말 우아한 설계를 자랑하는데, 크게 두 부분으로 나뉩니다.
하나는 기본적인 기능을 담당하는 코어(Core) 계층이고, 다른 하나는 그 위에 만들어진 전통적인 ORM(오알엠) 계층입니다.
파이썬(Python)의 다른 ORM(오알엠)이나 심지어 다른 언어의 ORM(오알엠)들을 봐도 이렇게 계층 설계가 잘 된 경우는 드문데요.
예를 들어, 장고(Django)의 ORM(오알엠) 같은 경우는 데이터베이스 연결 부분과 ORM(오알엠) 자체가 완전히 섞여 있기도 합니다.
마치 잘 지어진 건물처럼, SQLAlchemy(에스큐엘알케미)는 탄탄한 기초(코어) 위에 편리한 기능(ORM)들이 차곡차곡 쌓여있는 모습이랍니다.
왜 우리에게 코어(Core) 계층이 필요할까요?
코어(Core) 계층은 주로 클라이언트 연결 풀(connection pool)을 구현하는 역할을 담당합니다.
현대 웹 애플리케이션의 심장이라고 할 수 있는 관계형 데이터베이스는 동시에 많은 연결을 처리하는 능력이 생각보다 강하지 않은 경우가 많은데요.
그래서 보통 짧은 연결을 아주 많이 사용하는 것은 권장되지 않고, 대부분의 경우 연결 풀(connection pool)이 필요하게 됩니다.
연결 풀(connection pool)에는 대략 두 가지 종류가 있습니다.
- 서버 측 연결 풀(Server-side connection pool): 짧은 연결이 요청될 때마다 미리 만들어둔 긴 연결을 재사용하도록 할당해주는 전문적인 연결 풀 미들웨어(middleware)입니다.
- 클라이언트 측 연결 풀(Client-side connection pool): 보통 서드파티(third-party) 라이브러리 형태로 코드에 포함되어 사용됩니다.
SQLAlchemy(에스큐엘알케미)의 연결 풀은 바로 이 클라이언트 측 연결 풀(Client-side connection pool)에 해당합니다.
이 연결 풀 안에서 SQLAlchemy(에스큐엘알케미)는 일정 수의 긴 연결을 유지하고 있는데요.
우리가 connect를 호출하면 사실은 풀에서 연결을 하나 가져오는 것이고, close를 호출하면 풀에 연결을 다시 반납하는 방식입니다.
매번 식당에 갈 때마다 새로 식기를 만들 필요 없이, 미리 준비된 식기를 쓰고 반납하는 것과 비슷하다고 생각하면 이해하기 쉽습니다.
연결 만들기
SQLAlchemy(에스큐엘알케미)에서는 create_engine 함수를 사용해서 데이터베이스 연결(정확히는 연결 풀)을 만듭니다.create_engine 함수의 인자로는 데이터베이스의 URL(주소)을 전달합니다.
from sqlalchemy import create_engine
# MySQL(마이에스큐엘) 연결 예시
engine = create_engine(
"mysql://사용자이름:비밀번호@호스트주소:3306/데이터베이스이름", # 데이터베이스 접속 정보
echo=True, # 이 옵션을 True로 설정하면 실제 실행되는 SQL(에스큐엘) 문이 출력되어 디버깅에 편리합니다.
future=True, # SQLAlchemy 2.0 API(에이피아이)를 사용합니다. 하위 호환성을 제공합니다.
pool_size=5, # 연결 풀의 크기는 기본적으로 5입니다. 0으로 설정하면 연결 수에 제한이 없습니다.
pool_recycle=3600 # 데이터베이스에서 자동으로 연결을 끊는 시간을 제한합니다 (초 단위, 여기서는 1시간).
)
# 메모리 내 SQLite(에스큐엘라이트) 데이터베이스 생성 예시. 멀티스레드 환경에서 사용하려면 check_same_thread=False를 꼭 추가해야 합니다.
engine = create_engine("sqlite:///:memory:", echo=True, future=True,
connect_args={"check_same_thread": False})
# MySQL(마이에스큐엘)에 연결하는 또 다른 방법 (mysqlclient 드라이버 사용). pip install mysqlclient 명령어로 설치해야 합니다.
engine = create_engine('mysql+mysqldb://사용자이름:비밀번호@호스트주소/데이터베이스이름?charset=utf8mb4')코어(Core) 계층 -- SQL(에스큐엘) 직접 사용하기
CRUD (생성, 읽기, 수정, 삭제)
SQLAlchemy(에스큐엘알케미)의 코어(Core) 계층을 사용하면 마치 데이터베이스에 직접 SQL(에스큐엘) 명령을 내리듯이 작업할 수 있습니다.
from sqlalchemy import text # SQL(에스큐엘) 텍스트를 직접 사용하기 위해 임포트합니다.
# engine.connect()를 사용해 실제 데이터베이스 연결을 얻습니다.
with engine.connect() as conn:
# text() 함수로 SQL(에스큐엘) 문을 감싸서 실행합니다.
result = conn.execute(text("select * from users"))
# result.all()은 모든 결과를 리스트 형태로 가져옵니다.
print(result.all())
# 결과(result)는 반복문으로 하나씩 처리할 수 있고, 각 행(row)은 Row 객체입니다.
for row in result:
# Row 객체는 세 가지 방식으로 값에 접근할 수 있습니다.
print(row.x, row.y) # 속성 이름으로 접근
print(row[0], row[1]) # 인덱스로 접근
print(row["x"], row["y"]) # 컬럼 이름(문자열)으로 접근
# SQL(에스큐엘) 문에 파라미터를 전달할 때는 `:변수명` 형식을 사용합니다.
result = conn.execute(
text("SELECT x, y FROM some_table WHERE y > :y_param"), # 파라미터 이름은 SQL 키워드와 겹치지 않게 하는 것이 좋습니다.
{"y_param": 2} # 딕셔너리 형태로 파라미터 값을 전달합니다.
)
# 파라미터를 미리 바인딩(연결)해 둘 수도 있습니다.
stmt = text("SELECT x, y FROM some_table WHERE y > :y_param ORDER BY x, y").bindparams(y_param=6)
# result = conn.execute(stmt) # 이렇게 실행 가능
# 데이터를 삽입할 때, 여러 행을 한 번에 삽입할 수 있습니다.
conn.execute(
text("INSERT INTO some_table (x, y) VALUES (:x_val, :y_val)"),
[{"x_val": 11, "y_val": 12}, {"x_val": 13, "y_val": 14}] # 리스트 안에 딕셔너리 형태로 각 행의 데이터를 전달
)
# conn.commit() # 변경사항을 데이터베이스에 최종 반영하려면 commit이 필요합니다 (아래 트랜잭션 부분 참고).트랜잭션(Transaction)과 커밋(Commit)
SQLAlchemy(에스큐엘알케미)는 변경 사항을 데이터베이스에 최종적으로 반영하는 커밋(commit)을 두 가지 방식으로 제공하는데요.
하나는 수동으로 커밋(commit)하는 것이고, 다른 하나는 반자동으로 커밋(commit)하는 방식입니다.
공식 문서에서는 engine.begin() 사용을 권장합니다.
행마다 한 번씩 커밋(commit)하는 완전 자동 autocommit 방식도 있지만, 이 방식은 권장되지 않는답니다.
"commit as you go" 방식은 수동 커밋이 필요합니다.
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() # 여기서 직접 commit() 메서드를 호출합니다.
"begin once" 방식은 with 블록이 끝나면 자동으로 commit 또는 rollback 됩니다.
with engine.begin() as conn: # begin()을 사용합니다.
conn.execute(
text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
[{"x": 6, "y": 8}, {"x": 9, "y": 10}]
)
# with 블록이 정상적으로 완료되면 여기서 자동으로 commit이 됩니다.
# 만약 블록 실행 중 에러가 발생하면 rollback이 됩니다.
<br /><br />
### ORM(오알엠)<br /><br />
#### 세션(Session)<br /><br />
세션(Session)은 스레드에 안전하지 않은데요.<br /><br /> 하지만 보통 웹 프레임워크에서는 요청이 시작될 때마다 새로운 세션(Session)을 얻기 때문에 문제 되지 않습니다.<br /><br />
```python
from sqlalchemy.orm import Session # ORM을 사용하기 위해 Session을 임포트합니다.
# Session 객체를 생성하고 with 문으로 사용합니다.
with Session(engine) as session:
# ORM 객체(foo)를 세션에 추가합니다.
session.add(foo)
# 세션에 쌓인 변경 사항을 데이터베이스에 반영합니다.
session.commit()
# sessionmaker를 사용해서 세션 팩토리(공장 함수)를 만들 수도 있습니다.
# 이렇게 하면 매번 매개변수를 입력할 필요가 없어 편리합니다.
from sqlalchemy.orm import sessionmaker
new_session = sessionmaker(engine) # engine을 인자로 넘겨 세션 팩토리를 만듭니다.
with new_session() as session: # 팩토리 함수를 호출하여 세션을 얻습니다.
# ... 세션을 사용하여 ORM 작업 수행 ...
pass선언적 API (Declarative API)
선언적 API(Declarative API)는 파이썬(Python) 클래스를 이용해서 데이터베이스 테이블 구조와 ORM 모델을 편리하게 정의할 수 있도록 도와주는 방식인데요.
__tablename__속성을 사용해서 데이터베이스 테이블 이름을 지정합니다.Mapped와 파이썬 네이티브 타입(native types)을 사용해서 각 필드를 선언합니다.Integer,String등의 SQLAlchemy(에스큐엘알케미) 타입을 사용해서 필드 타입을 구체적으로 지정합니다.index매개변수를 사용해서 인덱스를 지정합니다.unique매개변수를 사용해서 유니크(unique) 인덱스를 지정합니다.__table_args__를 사용해서 복합 인덱스 같은 다른 속성들을 지정합니다.relationship을 사용해서 모델 간의 연관 관계를 정의합니다.
from datetime import datetime # 날짜/시간 처리를 위해 임포트합니다.
# 필요한 SQLAlchemy(에스큐엘알케미) 타입 및 함수들을 임포트합니다.
from sqlalchemy import Integer, String, func, UniqueConstraint, DateTime, Text, ForeignKey, Column
# ORM 관련 모듈들을 임포트합니다.
from sqlalchemy.orm import relationship, mapped_column, Mapped, DeclarativeBase
# Optional 타입 힌트를 위해 typing 모듈에서 Optional을 임포트할 수 있습니다.
from typing import Optional
# 모든 ORM 모델의 기반이 될 Base 클래스를 선언합니다. DeclarativeBase를 상속받습니다.
class Base(DeclarativeBase):
pass
# User 모델을 정의합니다. Base 클래스를 상속받습니다.
class User(Base):
__tablename__ = "users" # 데이터베이스 테이블 이름을 "users"로 지정합니다.
# 테이블 레벨 속성들을 지정합니다. 복합 유니크 제약조건 예시입니다.
# 이는 반드시 튜플(tuple) 형태여야 하며, 리스트(list)가 아닙니다.
__table_args__ = (UniqueConstraint("name", "time_created"),)
# id 필드 정의: Mapped[int]는 파이썬 타입 힌트이고, mapped_column은 데이터베이스 컬럼 매핑입니다.
# Integer 타입, primary_key=True로 기본 키임을 지정합니다.
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# name 필드 정의: String 타입(최대 길이 30), index=True로 인덱스를 생성합니다.
name: Mapped[str] = mapped_column(String(30), index=True)
# fullname 필드 정의: String 타입, unique=True로 유니크 제약조건을 추가합니다.
fullname: Mapped[str] = mapped_column(String, unique=True)
# Optional 타입 힌트 사용 예시 (nullable=True와 유사, 단 mapped_column에도 nullable=True 명시 권장)
# nullable=True가 필요하면 mapped_column에도 명시해야 합니다.
nickname: Mapped[Optional[str]] = mapped_column(String, nullable=True)
# 특히 큰 필드의 경우 deferred=True를 사용하면 기본적으로 로드되지 않도록 할 수 있습니다.
description: Mapped[str] = mapped_column(Text, deferred=True)
# 기본값 설정: default=datetime.now 와 같이 함수 자체를 전달해야 합니다.
# 함수를 호출한 결과(현재 시간 값)를 직접 넣으면 모델 정의 시점의 시간이 기본값이 됩니다.
time_created: Mapped[datetime] = mapped_column(DateTime(Timezone=True), default=datetime.now)
# 또는 데이터베이스 서버의 기본값을 사용할 수 있습니다.
# 하지만 이는 테이블 생성 시점에 설정되어야 하고, 테이블 스키마의 일부가 됩니다.
time_created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
# 데이터가 업데이트될 때마다 자동으로 현재 시간을 기록하는 필드입니다.
time_updated: Mapped[datetime] = mapped_column(DateTime(timezone=True), onupdate=func.now())
# Address 모델을 정의합니다.
class Address(Base):
__tablename__ = "address" # 데이터베이스 테이블 이름을 "address"로 지정합니다.
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# email_address 필드: String 타입, nullable=False로 값이 반드시 존재해야 함을 지정합니다.
email_address: Mapped[str] = mapped_column(String, nullable=False)
# User 모델과의 관계를 위한 외래 키 필드 추가 예시 (ORM 관계는 별도 설명)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# create_all 메서드를 호출해서 정의된 모든 모델에 해당하는 테이블을 데이터베이스에 생성합니다.
# 이미 테이블이 존재하면 생성되지 않습니다.
Base.metadata.create_all(engine)
# 만약 특정 모델 하나만 테이블로 만들고 싶다면 다음과 같이 할 수 있습니다.
# User.__table__.create(engine) 외래 키(Foreign Keys)
relationship을 사용해서 모델 간의 연관 관계를 지정하는데요.
일대다(One-to-Many)나 다대다(Many-to-Many) 같은 복잡한 관계를 모델 레벨에서 편리하게 다룰 수 있게 해줍니다.
일대다(One-to-Many) 관계 양방향 매핑
그룹(Group) 하나가 여러 사용자(User)를 가지고, 사용자(User) 한 명은 하나의 그룹(Group)에 속하는 관계를 생각해 볼까요?
from sqlalchemy import create_engine, Integer, String, ForeignKey, Column # 필요한 것들 임포트
from sqlalchemy.orm import DeclarativeBase, relationship, Session, Mapped, mapped_column
# 위에서 정의한 Base 클래스를 사용합니다.
# class Base(DeclarativeBase): pass
class Group(Base):
__tablename__ = 'groups' # 그룹 테이블
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String)
# 이 그룹에 속한 여러 사용자를 나타내는 필드입니다.
# 'User'는 연결될 모델 이름(클래스 이름)입니다. 이 필드를 통해 Group 객체에서 User 객체들을 리스트처럼 접근할 수 있습니다.
members = relationship('User', back_populates="group") # User 모델의 'group' 필드와 연결됩니다.
class User(Base):
__tablename__ = 'users' # 사용자 테이블
# id = Column(Integer, primary_key=True) # Mapped 방식을 사용하는 것이 권장됩니다.
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# name = Column(String) # Mapped 방식을 사용하는 것이 권장됩니다.
name: Mapped[str] = mapped_column(String)
# group_id는 데이터베이스의 실제 외래 키 컬럼 이름입니다.
# mapped_column에 ForeignKey('연결될테이블이름.컬럼이름')을 사용해서 외래 키임을 명시합니다.
group_id: Mapped[Optional[int]] = mapped_column(ForeignKey('groups.id'), nullable=True)
# 이 사용자가 속한 그룹을 나타내는 필드입니다.
# 'Group'는 연결될 모델 이름(클래스 이름)입니다. 이 필드를 통해 User 객체에서 Group 객체(또는 None)에 접근할 수 있습니다.
# back_populates="members"는 Group 모델의 'members' 필드와 연결됩니다.
group: Mapped[Optional['Group']] = relationship('Group', back_populates="members")
# 예시 사용법
# Base.metadata.create_all(engine) # 테이블 생성 (필요시)
# session = Session(engine) # 세션 생성 (필요시)
# group1 = Group(name="Developers")
# user1 = User(name="Alice", group=group1) # User 생성 시 Group 객체 직접 할당 가능
# user2 = User(name="Bob")
# group1.members.append(user2) # Group 객체에 User 객체 추가 가능
# session.add_all([group1, user1, user2])
# session.commit()
# print(user1.group.name) # Developers
# print(group1.members[1].name) # Bob
다대다(Many-to-Many) 매핑, 중간 테이블이 필요합니다.
사용자(User) 한 명이 여러 권한(Permission)을 가질 수 있고, 하나의 권한(Permission)은 여러 사용자(User)에게 부여될 수 있는 관계를 생각해 볼까요? 이런 다대다 관계는 중간에 연결(association) 테이블이 필요합니다.
from sqlalchemy import create_engine, Integer, String, ForeignKey, Column # 필요한 것들 임포트
from sqlalchemy.orm import DeclarativeBase, relationship, Session, Mapped, mapped_column
# 위에서 정의한 Base 클래스를 사용합니다.
# class Base(DeclarativeBase): pass
# 중간 연결 테이블 역할을 할 모델을 정의합니다.
class UserPermissions(Base):
__tablename__ = 'user_permissions' # 중간 테이블 이름
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# User 테이블의 id를 가리키는 외래 키
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True) # 복합 기본 키로 설정할 수도 있습니다.
# Permission 테이블의 id를 가리키는 외래 키
permission_id: Mapped[str] = mapped_column(String, ForeignKey('permissions.id'), primary_key=True) # 복합 기본 키로 설정할 수도 있습니다.
class User(Base):
__tablename__ = 'users' # 사용자 테이블
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# name: Mapped[str] = Column(String) # Mapped 방식 권장
name: Mapped[str] = mapped_column(String)
# 이 사용자가 가진 권한 목록을 나타내는 필드입니다.
# secondary 인자로 중간 테이블 모델 또는 테이블 이름을 지정합니다.
# back_populates는 Permission 모델의 'users' 필드와 연결됩니다.
permissions: Mapped[list['Permission']] = relationship('Permission', secondary="user_permissions", back_populates="users")
class Permission(Base):
__tablename__ = 'permissions' # 권한 테이블
# id: Mapped[int] = mapped_column(Integer, primary_key=True) # id 타입을 int가 아닌 String으로 가정 (원문 따라감)
id: Mapped[str] = mapped_column(String, primary_key=True) # id 타입을 문자열로 설정
# name: Mapped[str] = Column(String) # Mapped 방식 권장
name: Mapped[str] = mapped_column(String)
# 이 권한을 가진 사용자 목록을 나타내는 필드입니다.
# secondary 인자로 중간 테이블 모델 또는 테이블 이름을 지정합니다.
# back_populates는 User 모델의 'permissions' 필드와 연결됩니다.
users: Mapped[list['User']] = relationship('User', secondary="user_permissions", back_populates="permissions")
# 예시 사용법
# Base.metadata.create_all(engine) # 테이블 생성 (필요시)
# session = Session(engine) # 세션 생성 (필요시)
# user1 = User(name='user1', group_id=1) # User 모델에 group_id 필드가 있다고 가정
# user2 = User(name='user2') # group_id가 없는 사용자
# group1 = Group(name='group1') # Group 모델이 있다고 가정
# group2 = Group(name='group2', members=[user2]) # Group 생성 시 사용자 추가 가능
# permission1 = Permission(id="open", name="open_file") # Permission id가 문자열이라고 가정
# permission2 = Permission(id="save", name="save_file")
# user1에게 permission1을 추가합니다.
# user1.permissions.append(permission1)
# 여러 객체를 한 번에 추가합니다.
# session.add_all([user1, user2, group1, group2, permission1, permission2])
# 변경 사항을 데이터베이스에 저장합니다.
# session.commit()
# print(user1.permissions[0].id) # user1이 가진 첫 번째 권한의 id를 출력
다른 튜토리얼들에서는 backref를 사용해서 다른 모델의 속성을 자동으로 생성하는 경우가 많은데요.
여기서는 해당 모델에서 접근 가능한 속성을 명시적으로 선언하는(back_populates 사용) 방식이 더 좋다고 생각합니다.
코드를 읽을 때 관계를 더 명확하게 알 수 있거든요.
CRUD (ORM)
SQLAlchemy(에스큐엘알케미) 1.x API(에이피아이)와 다르게, 2.0 API(에이피아이)에서는 데이터를 조회할 때 더 이상 query를 사용하지 않고 select를 사용합니다.
from sqlalchemy import select, update, delete # 데이터 조회, 수정, 삭제를 위해 임포트합니다.
# where 조건 등을 구성하기 위해 필요한 연산자들을 임포트할 수 있습니다.
# from sqlalchemy import and_, or_, not_
# User 모델에서 name이 "john"인 사용자를 id 순서대로 조회하는 쿼리입니다.
stmt = select(User).where(User.name == "john").order_by(User.id)
# where의 인자는 `==`, `>`, `<` 등으로 구성된 표현식입니다.
# 장점은 코드를 작성할 때 오타가 나면 바로 감지된다는 점입니다.
# filter_by는 키워드 인자(**kwargs)를 사용합니다. 단순 동등 비교에 편리합니다.
stmt = select(User).filter_by(name="some_user")
# order_by는 User.id.desc()와 같이 사용하여 역순 정렬을 나타낼 수도 있습니다.
# 세션의 execute 메서드를 사용하여 쿼리를 실행합니다.
result = session.execute(stmt)
# 일반적으로 객체 전체를 선택할 때는 scalars() 메서드를 사용해야 합니다.
# scalars()를 사용하지 않으면 튜플 안에 객체가 하나 들어있는 형태로 결과가 반환됩니다.
for user in result.scalars(): # scalars()를 사용하여 User 객체 자체를 얻습니다.
print(user.name)
# 모델의 특정 속성 하나만 조회할 때는 scalars()를 사용할 필요가 없습니다.
result = session.execute(select(User.name))
for row in result: # 각 row는 단일 값(name)을 가진 튜플입니다.
print(row.name)
# id로 조회하는 지름길 메서드도 있습니다.
user = session.get(User, 1) # pk=1 대신 1만 전달해도 됩니다. id가 1인 User 객체를 가져옵니다.
# 데이터를 수정하려면 update 문을 사용해야 합니다.
# update(모델).where(...).values(...) 형식으로 사용합니다.
# synchronize_session 옵션은 데이터베이스 업데이트 후 세션 내 객체를 어떻게 동기화할지 제어합니다.
# 세 가지 옵션이 있습니다: false, "fetch", "evaluate" (기본값은 evaluate)
# false: 파이썬 객체는 전혀 업데이트하지 않습니다.
# fetch: 데이터베이스에서 객체를 다시 불러와 업데이트합니다.
# evaluate: 데이터베이스를 업데이트하면서 파이썬 객체에도 같은 연산을 최대한 적용하려고 시도합니다.
stmt = update(User).where(User.name == "john").values(name="John").execution_options(synchronize_session="fetch")
session.execute(stmt)
# 또는 ORM 객체의 속성에 직접 값을 할당하는 방식으로 수정할 수도 있습니다.
user.name = "John"
session.commit() # commit 시점에 변경 사항이 데이터베이스에 반영됩니다.
# 여기서 경쟁 조건(Race Condition)이 발생할 수 있는 중요한 부분이 있습니다.
# user 객체의 visit_count를 1 증가시키고 싶다고 가정합니다.
# 틀린 방법! 만약 두 개의 프로세스가 동시에 이 값을 업데이트하려고 하면, 한 명의 업데이트만 반영될 수 있습니다.
# 두 프로세스 모두 현재 값(예: 1)을 읽고, +1을 해서 2라는 값을 계산한 뒤, 그 2를 데이터베이스에 쓰려고 합니다.
# 결과적으로 최종 값은 2가 되지만, 실제로는 1+1+1=3이 되어야 합니다.
# 이 코드에 해당하는 SQL: UPDATE users SET visit_count = 2 WHERE user.id = 1
# user 객체를 가져와서 값에 1을 더하는 파이썬 코드:
# user = session.get(User, 1)
# user.visit_count += 1 # 파이썬 객체의 값을 변경
# 올바른 방법: 대문자 U, 즉 모델 자체의 속성을 사용합니다.
# 이렇게 하면 생성되는 SQL은 SQL 서버 측에서 1을 더하도록 합니다.
# 이렇게 하면 여러 프로세스가 동시에 실행되어도 visit_count 값을 안전하게 증가시킬 수 있습니다.
# 이 코드에 해당하는 SQL: UPDATE users SET visit_count = visit_count + 1 WHERE user.id = 1
# user 객체를 가져와서 모델의 속성을 사용하는 파이썬 코드:
# user = session.get(User, 1)
# user.visit_count = User.visit_count + 1 # 모델 속성을 사용한 표현식 할당
# 객체를 추가하려면 session.add 메서드를 직접 사용합니다.
session.add(user)
# 또는 여러 객체를 한 번에 추가할 때는 add_all을 사용합니다.
session.add_all([user1, user2, group1])
# 삽입된 객체의 ID를 얻고 싶다면, commit 후에 읽을 수도 있지만,
# commit 전에 flush()를 호출해서 ID를 미리 데이터베이스로부터 가져올 수 있습니다.
session.flush() # flush는 commit이 아닙니다. 트랜잭션은 아직 완료되지 않았습니다.
# 데이터베이스의 격리 수준에 따라 반복 가능한 읽기(repeatable read)가 가능할 수 있습니다.
print(user.id) # flush 후에는 데이터베이스에 저장되어 ID를 얻을 수 있습니다.
# 객체를 삭제하려면 session.delete를 사용합니다.
session.delete(user)
# session.commit() # 삭제를 최종 반영하려면 commit이 필요합니다.연관된 모델 로딩하기
N개의 레코드 목록을 읽은 뒤, 각 항목의 상세 값을 데이터베이스에서 하나씩 다시 조회한다면 N+1번의 쿼리가 발생하는데요.
이는 데이터베이스에서 가장 흔하게 저지르는 실수인 N+1 문제입니다.
기본적으로 쿼리할 때 외래 키로 연결된 연관 모델은 로드되지 않는답니다.selectinload 옵션을 사용하면 외래 키를 로드해서 N+1 문제를 피할 수 있습니다.
외래 키가 로드되지 않은 경우
session.execute(select(User)).scalars().all() # User 객체들은 가져오지만, User.groups에 접근하면 별도 쿼리 발생
외래 키가 로드된 경우 (N+1 문제 방지)
from sqlalchemy.orm import selectinload # selectinload를 임포트합니다.
session.execute(select(User).options(selectinload(User.groups))).scalars().all()
User 객체와 함께 연결된 Group 객체들도 한 번의 추가 쿼리로 효율적으로 가져옵니다.
<br /><br />
`Selectinload`의 원리는 서브쿼리(subquery)에서 `SELECT IN` 절을 사용하는 것입니다.<br /><br /> `selectinload` 외에 전통적인 `joinedload`도 사용할 수 있는데요.<br /><br /> 이 방식은 가장 일반적인 조인(join) 테이블을 사용하는 것이 원리입니다.<br /><br />
```python
from sqlalchemy.orm import joinedload # joinedload를 임포트합니다.
# joinedload를 사용하여 외래 키를 로드합니다. 2.0에서는 .unique() 메서드를 사용해야 하는 경우가 있습니다.
session.execute(select(User).options(joinedload(User.groups))).unique().scalars().all()
# User 테이블과 Group 테이블을 JOIN해서 데이터를 가져옵니다.
# JOIN 결과에 중복 행이 있을 수 있으므로 .unique()를 사용하여 중복을 제거하고 ORM 객체를 얻습니다.
2.0 버전에서는 joinedload보다는 selectinload 사용을 더 권장한답니다.
일반적으로 selectinload가 더 성능이 좋고, unique() 메서드를 사용할 필요가 없습니다.
외래 키 다루기
SQLAlchemy(에스큐엘알케미)에서는 외래 키로 연결된 연관 객체들을 마치 배열(list)을 다루듯이 직접 처리할 수 있습니다.
# user 객체에 open_permission 객체를 연결(추가)합니다. (다대다 관계 예시)
user.permissions.append(open_permission)
# user 객체에서 save_permission 객체 연결을 제거합니다.
user.permissions.remove(save_permission)
# user 객체와 연결된 모든 외래 키(permissions)를 제거합니다.
user.permissions.clear()
# 위와 동일한 효과를 냅니다.
user.permissions = [] JSON 필드 특별 처리
대부분의 데이터베이스는 이제 JSON 필드를 지원하는데요.
SQLAlchemy(에스큐엘알케미)에서는 필드에서 JSON 객체를 직접 읽어오거나 JSON 객체를 쓸 수 있습니다.
하지만 이 JSON 객체를 직접 업데이트하고 데이터베이스에 다시 쓰려고 기대해서는 절대 안 됩니다.
이건 신뢰할 수 없는 방법인데요.
반드시 JSON 객체를 복사(copy)해서 읽고 쓰고, 그 복사본을 다시 속성에 할당해 주어야 합니다.
import copy # 객체 복사를 위해 copy 모듈을 임포트합니다.
article = session.get(Article, 1) # id가 1인 Article 객체를 가져옵니다.
tags = copy.copy(article.tags) # Article 객체의 tags 필드(JSON)를 복사합니다.
tags.append("iOS") # 복사한 tags에 값을 추가합니다.
article.tags = tags # 수정된 복사본을 다시 Article 객체의 tags 필드에 할당합니다.
session.commit() # commit 시점에 변경 사항이 데이터베이스에 반영됩니다.배치 삽입(Batch Insertion)
많은 양의 데이터를 삽입해야 할 때, 하나씩 삽입하는 방식을 사용하면 데이터베이스와의 상호작용에 많은 시간이 소모되어 효율이 매우 낮습니다.
MySQL(마이에스큐엘)과 같은 대부분의 데이터베이스는 INSERT ... VALUES (...), (...) ... 형태의 배치 삽입 API(에이피아이)를 제공하는데요.
SQLAlchemy(에스큐엘알케미)에서도 이 기능을 잘 활용할 수 있습니다.
session.bulk_save_objects(...)를 사용하면 여러 ORM 객체를 한 번에 삽입할 수 있습니다.
from sqlalchemy.orm import Session
s = Session() # 세션 생성
objects = [
User(name="u1"),
User(name="u2"),
User(name="u3")
]
s.bulk_save_objects(objects) # 객체 리스트를 전달하여 대량 삽입
s.commit() # commit으로 최종 반영
bulk_insert_mappings를 사용하면 객체를 생성하는 오버헤드 없이 바로 딕셔너리 형태 데이터를 삽입할 수 있어 더 효율적입니다.
users = [
{"name": "u1"},
{"name": "u2"},
{"name": "u3"},
]
s.bulk_insert_mappings(User, users) # User 모델에 해당하는 딕셔너리 리스트와 모델 클래스 전달
s.commit()
bulk_update_mappings를 사용하면 객체들을 배치로 업데이트할 수 있습니다.
딕셔너리의 'id' 필드는 WHERE 조건으로 사용되고, 다른 모든 필드는 UPDATE 대상이 됩니다.
예를 들어 name을 업데이트하고 싶다면
users_to_update = [
{"id": 1, "name": "Updated User 1"},
{"id": 2, "name": "Updated User 2"},
]
session.bulk_update_mappings(User, users_to_update) # User 모델과 업데이트할 딕셔너리 리스트 전달
session.commit()
<br /><br />
### DeclarativeBase (선언적 기본 클래스)<br /><br />
`DeclarativeBase`는 파이썬(Python)의 네이티브 타입 시스템을 완전히 수용하여 모델 정의를 더 간결하고 명확하게 할 수 있게 해주는 SQLAlchemy(에스큐엘알케미) 2.0의 핵심 기능 중 하나인데요.<br /><br />
```python
from sqlalchemy.orm import DeclarativeBase # DeclarativeBase를 임포트합니다.
# 모든 ORM 모델의 기반이 될 Base 클래스를 DeclarativeBase를 상속받아 정의합니다.
class Base(DeclarativeBase):
pass
from sqlalchemy.orm import mapped_column, Mapped # mapped_column과 Mapped를 임포트합니다.
from typing import Optional # Optional 타입 힌트를 위해 임포트합니다.
from sqlalchemy import Integer, String # 필요한 컬럼 타입 임포트
# Mapped를 사용하여 파이썬 타입 힌트와 데이터베이스 컬럼 매핑을 결합합니다.
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# Optional 타입 힌트와 mapped_column에 nullable=True를 함께 사용하여 Null 허용 필드를 정의합니다.
fullname: Mapped[Optional[str]] = mapped_column(String, nullable=True)Asyncio (비동기 입출력)
Asyncio(에이싱크아이오)와 함께 SQLAlchemy(에스큐엘알케미)를 사용할 때는 작업(task)마다 하나의 AsyncSession 객체를 사용하는 것이 중요합니다.AsyncSession 객체는 진행 중인 하나의 상태를 가진 데이터베이스 트랜잭션을 나타내는 변경 가능한 상태 저장 객체이기 때문인데요.asyncio.gather() 같은 API(에이피아이)를 사용하여 여러 비동기 작업을 동시에 실행할 때, 각 개별 작업은 반드시 별도의 AsyncSession을 사용해야 합니다.
# asyncio 지원을 위한 모듈들을 임포트합니다.
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import select # 쿼리를 위해 임포트합니다.
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column # 모델 정의를 위해 임포트합니다.
from sqlalchemy import Integer # 필요한 컬럼 타입 임포트
# 비동기 모델의 기반이 될 Base 클래스를 정의합니다.
class Base(DeclarativeBase):
pass
# 예시 모델
class A(Base):
__tablename__ = "a"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# 비동기 엔진을 생성합니다.
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=True) # 비동기 SQLite 예시
# 비동기 세션 팩토리를 생성합니다.
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) # 커밋 시 객체 만료 방지 옵션
# 테이블 생성 (비동기 엔진에서는 run_sync를 사용해야 합니다).
async def create_tables():
async with engine.begin() as conn: # 비동기 연결 시작
await conn.run_sync(Base.metadata.create_all) # 동기 함수를 비동기 컨텍스트에서 실행
# 데이터 삽입 예시
async def insert_data():
async with AsyncSessionLocal() as db: # 비동기 세션 얻기
new_obj = A(id=1)
db.add(new_obj)
await db.commit() # 비동기 commit
# 데이터 조회 예시
async def query_data():
async with AsyncSessionLocal() as db: # 비동기 세션 얻기
stmt = select(A) # A 모델 전체 선택 쿼리
result = await db.execute(stmt) # 비동기 쿼리 실행
for obj in result.scalars(): # 객체 결과 가져오기
print(obj.id)
# 엔진 자원 해제 (프로그램 종료 시 호출 권장)
async def dispose_engine():
await engine.dispose() # 비동기 엔진 해제
# 위 함수들을 asyncio로 실행하는 예시 코드가 필요합니다 (asyncio.run 등).
# import asyncio
# async def main():
# await create_tables()
# await insert_data()
# await query_data()
# await dispose_engine()
#
# if __name__ == "__main__":
# asyncio.run(main())멀티프로세싱(Multiprocessing) 환경에서의 사용
파이썬(Python)의 전역 인터프리터 록(GIL, Global Interpreter Lock) 때문에 멀티코어 프로세서의 성능을 제대로 활용하려면 멀티프로세싱(multiprocessing)을 사용해야 합니다.
멀티프로세싱(multiprocessing) 환경에서는 자원을 공유할 수 없습니다.
SQLAlchemy(에스큐엘알케미)에 해당하는 것은 연결 풀(connection pool)을 공유할 수 없다는 것인데요.
이 문제는 우리가 수동으로 해결해야 한답니다.
일반적으로 여러 프로세스 간에 같은 세션(Session)을 공유하려고 시도하지 않는 것이 가장 좋습니다.
각 프로세스를 초기화할 때 세션(Session)을 새로 만드는 것이 제일 좋은 방법입니다.
값이 있을 때만 Where 조건 추가하기
웹 애플리케이션에서 URL(주소)을 통해 사용자가 어떤 옵션을 선택했는지에 따라 해당하는 결과만 반환해야 하는 경우가 자주 있는데요.
query = select(User) # User 모델 전체를 선택하는 기본 쿼리를 만듭니다.
# 만약 사용자 이름(username) 값이 있다면, WHERE 조건을 추가하여 쿼리를 수정합니다.
if username is not None:
query = query.where(User.username == username)
# 만약 비밀번호(password) 값이 있다면, WHERE 조건을 추가하여 쿼리를 다시 수정합니다.
if password is not None:
query = query.where(User.password == password)
# 마지막으로 완성된 쿼리를 실행합니다.
# result = session.execute(query)
# ... 결과 처리 ...이처럼 SQLAlchemy(에스큐엘알케미) 2.0은 강력하고 유연한 기능들을 제공하여 파이썬(Python) 데이터베이스 프로그래밍을 훨씬 효율적이고 안전하게 만들어 준답니다.
코어(Core)와 ORM(오알엠)의 명확한 분리, 개선된 타입 힌트 지원, Asyncio(에이싱크아이오)와의 연동, 그리고 다양한 최적화 기능들 덕분에 대규모 프로젝트에서도 빛을 발하는데요.
SQLAlchemy(에스큐엘알케미) 2.0을 마스터한다면 파이썬(Python)으로 데이터베이스 작업을 할 때 강력한 무기를 갖게 되는 셈입니다.
'Python' 카테고리의 다른 글
| 파이썬(Python) 최강의 기술, 데코레이터(Decorator) 완전 정복 (0) | 2025.05.17 |
|---|---|
| 파이썬 메타프로그래밍 마스터하기: 원하는 모든 것을 제어하는 방법 (0) | 2025.05.06 |
| 블룸 필터(Bloom Filter) 완벽 해부: 원리, 장단점, 파이썬(Python) 코드까지! (0) | 2025.05.06 |
| ASGI 깊이 알기: 파이썬 비동기 웹 앱 통신 규약 파헤치기! (FastAPI, Uvicorn 연관성 포함) (1) | 2025.05.06 |
| 파이썬 리스트 정렬의 숨겨진 비밀: 팀소트(Timsort)는 왜 빠를까요? (0) | 2025.05.05 |