SQLAlchemy 튜토리얼

SQLAlchemy는 Python 생태계에서 가장 인기 있는 ORM(Object Relational Mapping)입니다. 우아한 디자인을 가지고 있으며, 기본 Core와 상위 레벨의 전통적인 ORM의 두 부분으로 나뉩니다. Python뿐만 아니라 다른 언어의 대부분의 ORM에서는 훌륭한 계층적 디자인이 구현되지 않았습니다. 예를 들어 Django의 ORM에서는 데이터베이스 연결과 ORM 자체가 완전히 섞여 있습니다.

Core가 필요한 이유?

Core 레이어는 주로 클라이언트 연결 풀을 구현합니다. 현대 웹 애플리케이션의 핵심으로서, 관계형 데이터베이스의 동시 연결 기능은 종종 강력하지 않습니다. 많은 수의 짧은 연결을 사용하는 것은 일반적으로 권장되지 않으며, 대부분의 경우 연결 풀이 필요합니다. 연결 풀에는 대략 두 가지 유형이 있습니다.

• 서버 측 연결 풀: 짧은 연결마다 재사용을 위해 긴 연결을 할당하는 특수 연결 풀 미들웨어입니다.
• 클라이언트 측 연결 풀: 일반적으로 코드에 타사 라이브러리로 도입됩니다.

SQLAlchemy의 연결 풀은 클라이언트 측 연결 풀에 속합니다. 이 연결 풀에서 SQLAlchemy는 특정 수의 긴 연결을 유지 관리합니다. connect가 호출되면 실제로 풀에서 연결을 검색하고, close가 호출되면 실제로 연결을 풀로 반환합니다.

연결 생성

SQLAlchemy에서는 create_engine을 사용하여 연결(풀)을 생성합니다. create_engine의 매개변수는 데이터베이스의 URL입니다.

from sqlalchemy import create_engine

# MySQL 연결 예제
engine = create_engine(
    "mysql://user:password@localhost:3306/dbname",
    echo=True,  # echo를 True로 설정하면 실제 실행된 SQL이 출력되어 디버깅에 더 편리합니다.
    future=True,  # SQLAlchemy 2.0 API를 사용하며, 이전 버전과 호환됩니다.
    pool_size=5,  # 연결 풀의 크기는 기본적으로 5입니다. 0으로 설정하면 연결에 제한이 없습니다.
    pool_recycle=3600  # 데이터베이스의 자동 연결 해제를 제한하는 시간을 설정합니다.
)

# 메모리 내 SQLite 데이터베이스를 만듭니다. check_same_thread=False를 추가해야 합니다. 그렇지 않으면 다중 스레드 환경에서 사용할 수 없습니다.
engine = create_engine("sqlite:///:memory:", echo=True, future=True,
                       connect_args={"check_same_thread": False})

# MySQL에 연결하는 또 다른 방법
# pip install mysqlclient
engine = create_engine('mysql+mysqldb://user:password@localhost/foo?charset=utf8mb4')

Core 레이어 -- SQL 직접 사용

CRUD

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("select * from users"))
    print(result.all())

# 결과를 반복할 수 있으며, 각 행 결과는 Row 객체입니다.
for row in result:
    # Row 객체는 세 가지 액세스 방법을 지원합니다.
    print(row.x, row.y)
    print(row[0], row[1])
    print(row["x"], row["y"])

# 매개변수를 전달하려면 `:var`를 사용하여 전달합니다.
result = conn.execute(
    text("SELECT x, y FROM some_table WHERE y > :y"),
    {"y": 2}
)

# 매개변수를 미리 컴파일할 수도 있습니다.
stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6)

# 삽입할 때 여러 행을 직접 삽입할 수 있습니다.
conn.execute(
    text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
    [{"x": 11, "y": 12}, {"x": 13, "y": 14}]
)

트랜잭션 및 커밋

SQLAlchemy는 커밋하는 두 가지 방법을 제공합니다. 하나는 수동 commit이고, 다른 하나는 반자동 commit입니다. 공식 문서에서는 engine.begin()을 사용하는 것을 권장합니다. 각 행마다 한 번씩 커밋하는 완전 자동 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()  # 여기에서 커밋에 유의하십시오.

# "begin once" 반자동 커밋
with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 6, "y": 8}, {"x": 9, "y": 10}]
    )

ORM

세션

Session은 스레드로부터 안전하지 않습니다. 그러나 일반적으로 웹 프레임워크는 각 요청 시작 시 session을 획득해야 하므로 문제가 되지 않습니다.

from sqlalchemy.orm import Session

with Session(engine) as session:
    session.add(foo)
    session.commit()

# sessionmaker를 사용하여 팩토리 함수를 만들 수도 있으므로 매번 매개변수를 입력할 필요가 없습니다.
from sqlalchemy.orm import sessionmaker
new_session = sessionmaker(engine)

with new_session() as session:
    ...

Declarative API

• 데이터베이스 테이블 이름을 지정하려면 __tablename__을 사용하십시오.
• 각 필드를 선언하려면 Mapped 및 기본 형식을 사용하십시오.
• 필드 형식을 지정하려면 Integer, String 등을 사용하십시오.
• 인덱스를 지정하려면 index 매개변수를 사용하십시오.
• 고유 인덱스를 지정하려면 unique 매개변수를 사용하십시오.
• 복합 인덱스와 같은 다른 속성을 지정하려면 __table_args__를 사용하십시오.

from datetime import datetime
from sqlalchemy import Integer, String, func, UniqueConstraint
from sqlalchemy.orm import relationship, mapped_column, Mapped
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    # 목록이 아닌 튜플이어야 합니다.
    __table_args__ = (UniqueConstraint("name", "time_created"),)
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), index=True)
    fullname: Mapped[str] = mapped_column(String, unique=True)
    # 특히 큰 필드의 경우 기본적으로 이 필드가 로드되지 않도록 deferred를 사용할 수도 있습니다.
    description: Mapped[str] = mapped_column(Text, deferred=True)
    # 기본값은 현재 시간이 아닌 함수가 전달됩니다.
    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())


class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)

# 모든 모델을 만들려면 create_all을 호출하십시오.
Base.metadata.create_all(engine)

# 하나의 모델만 만들어야 하는 경우
User.__table__.create(engine)

외래 키

모델 간의 연결 관계를 지정하려면 relationship을 사용하십시오.

일대다 관계의 양방향 매핑

from sqlalchemy import create_engine, Integer, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, relationship, Session, Mapped, mapped_column

class Group(Base):
    __tablename__ = 'groups'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String)
    # 해당하는 여러 사용자, 여기서는 모델 이름을 매개변수로 사용합니다.
    members = relationship('User')

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # group_id는 데이터베이스의 실제 외래 키 이름이며, 두 번째 필드 ForeignKey는 해당 ID를 지정하는 데 사용됩니다.
    group_id = Column(Integer, ForeignKey('groups.id'))
    # 모델의 해당 그룹 필드. 해당 모델에서 겹치는 필드를 선언해야 합니다.
    group = relationship('Group', overlaps="members")

다대다 매핑에는 연결 테이블이 필요합니다.

# 연결 테이블
class UserPermissions(Base):
    __tablename__ = 'user_permissions'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    # 외래 키를 사용하여 외래 키를 지정하십시오.
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey('users.id'))
    permission_id: Mapped[str] = mapped_column(String, ForeignKey('permissions.id'))

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    # 보조를 사용하여 연결 테이블을 지정하고 overlaps를 사용하여 모델의 해당 필드를 지정하십시오.
    permissions = relationship('Permission', secondary="user_permissions", overlaps="users")

class Permission(Base):
    __tablename__ = 'permissions'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    # 위와 같습니다.
    users = relationship('User', secondary="user_permissions", overlaps="permissions")


user1 = User(name='user1', group_id=1)
user2 = User(name='user2')
group1 = Group(name='group1')
group2 = Group(name='group2', members=[user2])
permission1 = Permission(name="open_file")
permission2 = Permission(name="save_file")
user1.permissions.append(permission1)

db.add_all([user1, user2, group1, group2, permission1, permission2])

db.commit()

print(user1.permissions[0].id)

대부분의 다른 튜토리얼에서는 backref를 사용하여 해당 모델의 속성을 생성합니다. 여기서 해당 모델에서 접근 가능한 속성을 명시적으로 선언하는 것이 좋습니다.

CRUD

1.x API와 달리 2.0 API에서는 더 이상 query가 사용되지 않고 select가 사용되어 데이터를 쿼리합니다.

from sqlalchemy import select

# where의 매개변수는 `==`로 구성된 표현식입니다. 장점은 코드를 작성할 때 맞춤법 오류가 감지된다는 것입니다.
stmt = select(User).where(User.name == "john").order_by(User.id)
# filter_by는 **kwargs를 매개변수로 사용합니다.
stmt = select(User).filter_by(name="some_user")
# order_by는 User.id.desc()를 사용하여 역순 정렬을 나타낼 수도 있습니다.

result = session.execute(stmt)

# 일반적으로 전체 객체를 선택할 때는 스칼라 메서드를 사용해야 합니다. 그렇지 않으면 하나의 객체를 포함하는 튜플이 반환됩니다.
for user in result.scalars():
    print(user.name)

# 모델의 단일 속성을 쿼리할 때는 스칼라를 사용할 필요가 없습니다.
result = session.execute(select(User.name))
for row in result:
    print(row.name)

# ID로 쿼리하는 바로 가기도 있습니다.
user = session.get(User, pk=1)

# 데이터를 업데이트하려면 update 문을 사용해야 합니다.
from sqlalchemy import update
# synchronize_session에는 false, "fetch", "evaluate"의 세 가지 옵션이 있으며 기본값은 evaluate입니다.
# false는 Python에서 객체를 전혀 업데이트하지 않음을 의미합니다.
# fetch는 데이터베이스에서 객체를 다시 로드함을 의미합니다.
# evaluate는 데이터베이스를 업데이트하는 동안 Python에서 가능한 한 객체에 대해서도 동일한 작업을 시도하는 것을 의미합니다.
stmt = update(User).where(User.name == "john").values(name="John").execution_options(synchronize_session="fetch")
session.execute(stmt)

# 또는 속성에 직접 값을 할당하십시오.
user.name = "John"
session.commit()

# 여기에 경쟁 조건(競態條件)을 도입할 수 있는 장소가 있습니다.
# 잘못되었습니다! 두 프로세스가 동시에 이 값을 업데이트하면 하나의 값만 업데이트될 수 있습니다.
# 둘 다 올바르다고 생각하는 값인 2를 할당하지만 실제 올바른 값은 1 + 1 + 1 = 3입니다.
# 해당 SQL: Update users set visit_count = 2 where user.id = 1
user.visit_count += 1
# 올바른 접근 방식: 대문자 U, 즉 모델의 속성을 사용하고 생성된 SQL은 SQL 서버 측에서 1을 추가하는 것입니다.
# 해당 SQL: Update users set visit_count = visit_count + 1 where user.id = 1
user.visit_count = User.visit_count + 1

# 객체를 추가하려면 session.add 메서드를 직접 사용하십시오.
session.add(user)
# 또는 add_all
session.add_all([user1, user2, group1])

# 삽입된 ID를 가져오려면 커밋 후 읽을 수도 있습니다.
session.flush()  # 플러시는 커밋이 아니며 트랜잭션이 커밋되지 않았습니다. 데이터베이스의 격리 수준과 관련된 반복 가능한 읽기여야 합니다.
print(user.id)

# 삭제하려면 session.delete를 사용하십시오.
session.delete(user)

연결된 모델 로드

N개 레코드 목록을 읽은 후 데이터베이스로 이동하여 각 항목의 특정 값을 하나씩 읽으면 N+1개 쿼리가 생성됩니다. 이는 데이터베이스에서 가장 흔한 실수인 N+1 문제입니다.

기본적으로 쿼리에서 외래 키 연결 모델이 로드되지 않습니다. 외래 키를 로드하려면 selectinload 옵션을 사용하여 N+1 문제를 방지할 수 있습니다.

# 외래 키가 로드되지 않았습니다.
session.execute(select(User)).scalars().all()
# 외래 키가 로드되었습니다.
session.execute(select(User).options(selectinload(User.groups))).scalars().all()

Selectinload의 원리는 select in 하위 쿼리를 사용하는 것입니다. selectinload 외에도 기존 joinedload를 사용할 수도 있으며 그 원리는 가장 일반적인 join 테이블입니다.

# joinedload를 사용하여 외래 키를 로드합니다. unique 메서드를 사용해야 합니다. 2.0에서 지정됩니다.
session.execute(select(User).options(joinedload(User.groups))).unique().scalars().all()

2.0에서는 joinedload보다 selectinload를 사용하는 것이 좋습니다. 일반적으로 selectinload가 더 좋으며 unique를 사용할 필요가 없습니다.

외래 키 쓰기

SQLAlchemy에서는 배열을 처리하는 것처럼 외래 키를 직접 처리할 수 있습니다.

user.permissions.append(open_permission)  # 추가
user.permissions.remove(save_permission)  # 제거
# 모든 외래 키 지우기
user.permissions.clear()
user.permissions = []

JSON 필드의 특수 처리

대부분의 데이터베이스는 이제 JSON 필드를 지원합니다. SQLAlchemy에서는 필드에서 JSON 객체를 직접 읽거나 JSON 객체를 필드에 쓸 수 있습니다. 그러나 이 JSON 객체에 대해 직접 update를 수행하고 데이터베이스에 다시 쓸 것으로 기대하지 마십시오. 확실하지 않습니다. 복사, 읽기 및 쓰기를 수행한 다음 다시 할당하십시오.

import copy
article = session.get(Article, 1)
tags = copy.copy(article.tags)
tags.append("iOS")
article.tags = tags
session.commit()

일괄 삽입

많은 양의 데이터를 삽입해야 하는 경우 하나씩 삽입하는 방법을 사용하면 데이터베이스와의 상호 작용에 많은 시간이 낭비되고 효율성이 매우 낮습니다. MySQL과 같은 대부분의 데이터베이스는 insert ... values ​​(...), (...) ... 일괄 삽입 API를 제공하며 SQLAlchemy에서도 이를 잘 활용할 수 있습니다.

# session.bulk_save_objects(...)를 사용하여 여러 객체를 직접 삽입하십시오.
from sqlalchemy.orm import Session

s = Session()
objects = [
    User(name="u1"),
    User(name="u2"),
    User(name="u3")
]
s.bulk_save_objects(objects)
s.commit()

# bulk_insert_mappings를 사용하면 객체 생성에 대한 오버헤드를 줄이고 사전을 직접 삽입할 수 있습니다.
users = [
    {"name": "u1"},
    {"name": "u2"},
    {"name": "u3"},
]
s.bulk_insert_mappings(User, users)
s.commit()

# bulk_update_mappings를 사용하면 객체를 일괄적으로 업데이트할 수 있습니다. 사전의 id는 where 조건으로 사용됩니다.
# 다른 모든 필드는 업데이트에 사용됩니다.
session.bulk_update_mappings(User, users)

DeclarativeBase

Python 기본 형식 시스템을 완전히 수용하십시오.

from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass
from sqlalchemy.orm import mapped_column, MappedColumn
id: Mapped[int] = mapped_column(Integer, primary_key=True)
fullname: Mapped[Optional[str]]

Asyncio

태스크당 하나의 AsyncSession. AsyncSession 객체는 진행 중인 단일 상태 저장 데이터베이스 트랜잭션을 나타내는 변경 가능한 상태 저장 객체입니다. asyncio.gather()와 같은 API를 사용하여 동시 태스크에 asyncio를 사용하는 경우 각 개별 태스크는 별도의 AsyncSession을 사용해야 합니다.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(url, echo=True)
session = async_sessionmaker(engine)

# 객체 생성
async with engine.begin() as conn:
       await conn.run_sync(Base.metadata.create_all)

# 데이터 삽입
async with session() as db:
    db.add(...)
    await db.commit()

# 데이터 쿼리
async with session() as db:
    stmt = select(A)
    row = await db.execute(stmt)
    for obj in row.scalars():
        print(obj.id)

await engine.dispose()

다중 처리 환경에서 사용

Python의 GIL(Global Interpreter Lock)로 인해 다중 코어 프로세서를 활용하려면 다중 처리를 사용해야 합니다. 다중 처리 환경에서는 리소스를 공유할 수 없습니다. SQLAlchemy에 해당하는 즉, 연결 풀을 공유할 수 없습니다. 이 문제를 수동으로 해결해야 합니다.

일반적으로 여러 프로세스 간에 동일한 Session을 공유하려고 시도하지 않는 것이 가장 좋습니다. 각 프로세스를 초기화할 때 Session을 만드는 것이 가장 좋습니다.

값이 설정된 경우에만 Where 조건 추가

URL에서는 사용자가 지정한 옵션에 따라 해당 결과를 반환해야 하는 경우가 많습니다.

query = select(User)
if username is not None:
    query = query.where(User.username == username)
if password is not None:
    query = query.where(User.password == password)

Leapcell: 웹 호스팅, 비동기 태스크 및 Redis를 위한 차세대 서버리스 플랫폼

마지막으로 Python 서비스를 배포하는 데 가장 적합한 플랫폼인 **Leapcell**을 추천합니다.

1. 다국어 지원

• JavaScript, Python, Go 또는 Rust로 개발하십시오.

2. 무제한 프로젝트를 무료로 배포하십시오.

• 사용량에 대해서만 지불하십시오. 요청도 없고 요금도 없습니다.

3. 타의 추종을 불허하는 비용 효율성

• 유휴 요금 없이 사용한 만큼 지불하십시오.
• 예: $25는 평균 응답 시간이 60ms인 6.94백만 개의 요청을 지원합니다.

4. 능률적인 개발자 경험

• 간편한 설정을 위한 직관적인 UI.
• 완전 자동화된 CI/CD 파이프라인 및 GitOps 통합.
• 실행 가능한 통찰력을 위한 실시간 메트릭 및 로깅.

5. 간편한 확장성 및 고성능

• 높은 동시성을 쉽게 처리하기 위한 자동 확장.
• 제로 운영 오버헤드. 구축에만 집중하십시오.

설명서에서 자세히 알아보십시오!

Leapcell 트위터: https://x.com/LeapcellHQ