MSSQL-python과 SQLAlchemy를 함께 사용하세요

SQLAlchemy는 가장 널리 사용되는 Python ORM 및 데이터베이스 툴킷입니다. SQLAlchemy 2.1.0b2부터는 mssql-python 드라이버에 내장된 방언을 통해 Microsoft SQL 및 Azure SQL Database와 함께 SQLAlchemy ORM과 코어를 사용할 수 있습니다.

중요합니다

mssql-python 방언은 SQLAlchemy 2.1.0b2(2026년 4월 16일 출시)에서 추가되었습니다. SQLAlchemy 2.1은 현재 프리릴리즈 시리즈로 운영 환경 에서는 권장되지 않습니다. SQLAlchemy 2.0에서 업그레이드하기 전에 다음을 이해하세요:

• API는 최종 안정 버전(2.1 GA) 이전에 변경될 수 있습니다
• 배치 전에 업무량을 철저히 테스트하세요
• 2.1이 GA에 도달할 때까지 안정적인 SQLAlchemy 2.0.x를 프로덕션 시스템용으로 사용하세요
• 의존성을 특정 버전(예: sqlalchemy==2.1.0b2)에 고정하세요 버전 범위를 사용하는 대신

사전 출시 버전을 언제 사용해야 하는지에 대한 자세한 내용은 알려진 제한 사항을 참조하세요.

사전 요구 사항

• Python 3.10 이상. SQLAlchemy 2.1은 Python 3.9 및 그 이전 버전에 대한 지원을 중단했습니다.
• mssql-python 그리고 sqlalchemy 패키지들(2.1.0b2 또는 그 이후).

이 글의 예시들은 AdventureWorksLT 샘플 데이터베이스를 사용합니다. AdventureWorksLT가 설치되어 있지 않다면, AdventureWorks 샘플 데이터베이스를 참고하세요.

프리릴리즈 설치하세요

SQLAlchemy 2.1이 베타 버전이기 때문에 pip install sqlalchemy 기본적으로 최신 안정적인 2.0.x 버전을 설치합니다. 사전 출시 버전을 명시적으로 설치하세요:

pip install mssql-python "sqlalchemy>=2.1.0b2"

설치된 버전을 확인합니다.

import sqlalchemy
print(sqlalchemy.__version__)  # Should show 2.1.0b2 or later

연결 URL

mssql-python 방언은 URL 스킴으로 mssql+mssqlpython를 사용합니다. 일반적인 형식은 다음과 같습니다.

mssql+mssqlpython://<username>:<password>@<host>:<port>/<database>

SQL 인증

SQL 인증을 위해 연결 URL에 사용자 이름과 비밀번호를 포함하세요:

from sqlalchemy import create_engine

# Replace <password> with your actual password. Avoid using the sa account in production.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)

"Microsoft Entra" 인증

Microsoft Entra 인증의 경우, 빈 사용자 이름과 authentication 쿼리 매개변수를 사용하세요:

from sqlalchemy import create_engine

engine = create_engine(
    "mssql+mssqlpython://@<server>.database.windows.net/<database>"
    "?authentication=ActiveDirectoryDefault&encrypt=yes"
)

비고

ActiveDirectoryDefault는 여러 자격 증명 제공자를 순서대로 시도하는 DefaultAzureCredential를 사용합니다. 첫 번째 연결은 SDK가 작동하는 공급자를 찾을 때까지 체인을 따라 걸어 다니기 때문에 느릴 수 있습니다. 운영 환경에서는 어떤 자격 증명 유형을 사용하는지 알면, 체인 워크를 피하기 위해 직접 지정하세요(예: ActiveDirectoryMSI 관리 신원). 자세한 내용은 Microsoft Entra 인증을 참조하세요.

URL을 프로그래밍적으로 구축하세요

수동 URL 인코딩을 피하는 방법 sqlalchemy.engine.URL.create :

from sqlalchemy.engine import URL

url = URL.create(
    "mssql+mssqlpython",
    username="dbuser",
    password="<password>",
    host="localhost",
    port=1433,
    database="<database>",
)
engine = create_engine(url)

ORM 모델 정의

SQLAlchemy의 선언적 매핑을 사용해 Microsoft SQL 테이블에 매핑하는 모델을 정의하세요.

from datetime import datetime
from decimal import Decimal

from sqlalchemy import Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )

팁 (조언)

Microsoft SQL은 자동 증가 열에 IDENTITY를 사용합니다. SQLAlchemy 는 정수 기본 키 열에 대해 자동으로 매핑합니다. 위에 표시된 명시적 기능은 Identity() 시작 값과 증분 값을 제어해야 할 필요가 없다면 선택 사항입니다.

CRUD 작전

다음 예시들은 ORM 세션을 사용하여 행을 삽입, 쿼리, 업데이트, 삭제하는 방법을 보여줍니다. 각 예제에서는 행을 삽입할 때 반환되는 ProductID인 new_id를 재사용합니다. 네 가지 연산을 함께 실행하려면 전체 예제를 참고하세요.

세션 만들기

트랜잭션 내에서 작업을 실행할 세션을 생성하세요:

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Use session for queries and modifications
    pass

많은 세션을 생성하는 애플리케이션의 경우, 다음을 사용하세요 sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

행 추가

다음 예시에서 생성된 ProductID 생성 결과를 캡처하여 새 제품을 추가하고, 세션을 커밋합니다:

from datetime import datetime

with Session(engine) as session:
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()

    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

비고

SalesLT.Product에서 Name와 ProductNumber는 모두 고유한 제약 조건을 가진다. 이 삽입 파일을 여러 번 실행한다면, 먼저 값을 바꾸거나 이전 행을 삭제하세요. 완전한 예제는 생성된 행을 삭제하여 반복 실행이 가능합니다.

쿼리 행

기본 키로 단일 행을 가져오거나, 필터링된 쿼리에 사용 select() 하세요:

from sqlalchemy import select

with Session(engine) as session:
    # Single row by primary key (new_id is from the insert example)
    product = session.get(Product, new_id)
    if product:
        print(f"{product.name}: ${product.list_price}")

    # Filtered query
    stmt = select(Product).where(Product.list_price < 500).order_by(Product.name)
    products = session.scalars(stmt).all()
    for p in products:
        print(f"{p.name}: ${p.list_price}")

행 업데이트

기존 행의 필드를 수정하고 커밋합니다:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        product.list_price = Decimal("1349.99")
        session.commit()

여러 행 삭제하기

행을 제거하고 커밋하세요:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        session.delete(product)
        session.commit()

전체 예제

이전 섹션에서는 각 작품을 개별적으로 보여줬습니다. 이 섹션에서는 이 두 단어를 하나의 독립된 스크립트로 결합하여 복사하고, 실행하고, 다시 실행할 수 있습니다.

crud.py 파일을 만들고 다음 코드를 추가합니다. 연결 정보를 create_engine 본인의 것으로 교체하세요(연결 URL 참조):

from datetime import datetime
from decimal import Decimal

from sqlalchemy import create_engine, Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# Replace <password> and <database> with your connection details.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )


with Session(engine) as session:
    # Create
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()
    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

    # Read
    product = session.get(Product, new_id)
    print(f"Read: {product.name} costs ${product.list_price}")

    # Update
    product.list_price = Decimal("1349.99")
    session.commit()
    print(f"Updated price to ${product.list_price}")

    # Delete
    session.delete(product)
    session.commit()
    print(f"Deleted ProductID: {new_id}")

스크립트를 실행합니다.

python crud.py

다음과 같은 결과물을 볼 수 있습니다:

Inserted ProductID: 1019
Read: Classic Road Bike costs $1299.9900
Updated price to $1349.9900
Deleted ProductID: 1019

스크립트는 생성된 행을 삭제해서 다시 실행할 때 고유 제약 NameProductNumber 조건을 충족하지 않습니다. 실행할 때마다 새 행이 삽입되므로 ProductID는 매번 증가합니다.

핵심 쿼리

SQLAlchemy Core는 하위 수준 SQL 표현식 API를 제공합니다. 동일한 엔진과 테이블 정의, ORM 매핑 클래스를 포함해 Core를 사용할 수 있습니다.

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT @@VERSION"))
    print(result.scalar())

타입 안전한 SQL 생성을 위해 테이블 수준 구조를 사용하세요:

from sqlalchemy import insert, select, update, delete

with engine.connect() as conn:
    # Insert
    conn.execute(
        insert(Product).values(
            name="Touring Bike",
            product_number="BK-T002",
            list_price=Decimal("999.99"),
            standard_cost=Decimal("575.00"),
            sell_start_date=datetime(2026, 1, 1)
        )
    )
    conn.commit()

    # Select
    stmt = select(
        Product.name.label("name"),
        Product.list_price.label("list_price"),
    ).where(Product.list_price > 100)
    for row in conn.execute(stmt):
        print(row.name, row.list_price)

    # Delete the inserted row so this example can run again
    conn.execute(delete(Product).where(Product.product_number == "BK-T002"))
    conn.commit()

비고

데이터베이스 이름이 속성 이름과 다른 개별 매핑된 열을 선택할 때(예 Product.name : 해당 열에 Name 매핑됨), 코어 행은 데이터베이스 열 이름으로 키가 지정됩니다. row.Name 대신 row.name로 값에 액세스하려면 .label("name")를 추가하세요.

연결 풀링 (Connection Pooling)

SQLAlchemy 는 기본적으로 연결 풀을 관리합니다. 작업 부하에 맞는 풀 설정 조정:

engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost/<database>",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
)

매개 변수	설명
pool_size	열어둘 연결 수(기본값: 5개).
max_overflow	pool_size를 초과하는 허용 연결 수(기본값: 10).
pool_timeout	오류를 발생시키기 전에 연결을 기다리는 시간(초)(기본값: 30).
pool_recycle	몇 초 후 연결이 재순환됩니다(기본값: -1, 비활성화). 데이터베이스가 유휴 연결을 종료할 때 이 값을 설정하세요.

웹 프레임워크와의 사용

SQLAlchemy 는 Flask와 FastAPI의 데이터베이스 계층으로 일반적으로 사용됩니다. mssql-python 방언은 SQLAlchemy를 지원하는 모든 프레임워크에서 작동합니다.

다음 스니펫은 각 프레임워크에 권장되는 요청당 세션 패턴을 보여줍니다. 이것들은 앞선 섹션에서 설명한 engine 및 Product 모델을 전제로 한 예시용 조각일 뿐, 완전한 애플리케이션은 아닙니다. 완전하고 실행 가능한 애플리케이션은 FastAPI 통합 및 Flask 통합 문서를 참조하세요.

FastAPI 예시

생성기 의존성을 사용하여 요청당 세션을 제공합니다:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/products/{product_id}")
def read_product(product_id: int, db: Session = Depends(get_db)):
    product = db.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return {"name": product.name, "price": float(product.list_price)}

Flask 예제

컨텍스트 매니저를 사용해 세션을 요청에 맞게 범위를 조정하세요:

from flask import Flask, jsonify
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = Flask(__name__)


@app.route("/products/<int:product_id>")
def read_product(product_id):
    with SessionLocal() as session:
        product = session.get(Product, product_id)
        if not product:
            return jsonify({"error": "Not found"}), 404
        return jsonify({"name": product.name, "price": float(product.list_price)})

알렘빅 이동

Alembic 은 SQLAlchemy 프로젝트의 스키마 마이그레이션을 처리하며, mssql-python 방언과 함께 작동합니다. Alembic의 자동 생성 기능은 모델을 실시간 데이터베이스와 비교해주기 때문에, 몇 가지 추가 단계를 통해 관리하지 않는 테이블에 변경 제안을 방지할 수 있습니다.

알렘빅 설립

Alembic을 설치하고 마이그레이션 디렉터리를 초기화하세요:

pip install alembic
alembic init migrations

에서 alembic.ini연결 URL을 설정합니다:

sqlalchemy.url = mssql+mssqlpython://dbuser:<password>@localhost/<database>

알렘빅을 모델에 조준하세요

자동 생성은 모델의 메타데이터가 필요합니다. Alembic이 관리하는 모델을 임포트 가능한 모듈(예: models.py.)에 넣으세요. 자동 생성은 모델이 빠뜨린 열을 삭제할 것을 제안하므로, 이 글 앞서 설명한 단순화된 Product 모델을 재사용하지 않고 테이블을 완전히 소유하는 모델을 정의하세요:

# models.py
from datetime import datetime

from sqlalchemy import Identity, String, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class ProductReview(Base):
    __tablename__ = "ProductReview"
    __table_args__ = {"schema": "SalesLT"}

    review_id: Mapped[int] = mapped_column("ReviewID", Integer, Identity(), primary_key=True)
    product_id: Mapped[int] = mapped_column("ProductID", Integer)
    reviewer_name: Mapped[str] = mapped_column("ReviewerName", String(50))
    rating: Mapped[int] = mapped_column("Rating", Integer)
    comments: Mapped[str | None] = mapped_column("Comments", String(500))
    modified_date: Mapped[datetime] = mapped_column("ModifiedDate", DateTime, server_default=func.getdate())

주의

기본적으로 autogenerate는 target_metadata에 없는 데이터베이스의 모든 테이블을 제거된 것으로 간주하고, 각 테이블에 대해 drop_table을 생성합니다. AdventureWorksLT 같은 기존 데이터베이스를 상대로는 그 행동이 수십 개의 테이블을 버릴 수 있습니다. Alembic이 모델이 정의한 테이블만 관리하도록 필터를 include_name 추가하고, 생성된 스크립트를 적용하기 전에 항상 검토하세요.

migrations/env.py에서 target_metadata = None을 다음 코드로 바꾸세요. 모델을 가져오고, 자동 생성을 그들이 정의한 스키마와 테이블로 제한합니다:

from models import Base

target_metadata = Base.metadata

# Limit autogenerate to the tables your models define.
managed_schemas = {table.schema for table in target_metadata.tables.values()}
managed_tables = {table.name for table in target_metadata.tables.values()}


def include_name(name, type_, parent_names):
    if type_ == "schema":
        return name in managed_schemas
    if type_ == "table":
        return name in managed_tables
    return True

run_migrations_offline와 run_migrations_online 모두에서 include_name 및 include_schemas=True을 context.configure에 전달하세요. 이 설정은 include_schemas=True Alembic이 다음과 같은 비기본 스키마 SalesLT의 테이블을 볼 수 있게 합니다:

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_name=include_name,
    include_schemas=True,
)

마이그레이션을 생성하고 적용하세요

모델에서 마이그레이션을 생성하세요:

alembic revision --autogenerate -m "add product review table"

Alembic은 새 테이블을 감지하고 마이그레이션 스크립트를 작성합니다:

INFO  [alembic.autogenerate.compare.tables] Detected added table 'SalesLT.ProductReview'
Generating .../versions/xxxx_add_product_review_table.py ... done

생성된 upgrade() 이 테이블을 생성하고 downgrade() 이를 드랍합니다:

def upgrade() -> None:
    op.create_table(
        "ProductReview",
        sa.Column("ReviewID", sa.Integer(), sa.Identity(always=False), nullable=False),
        sa.Column("ProductID", sa.Integer(), nullable=False),
        sa.Column("ReviewerName", sa.String(length=50), nullable=False),
        sa.Column("Rating", sa.Integer(), nullable=False),
        sa.Column("Comments", sa.String(length=500), nullable=True),
        sa.Column("ModifiedDate", sa.DateTime(), server_default=sa.text("getdate()"), nullable=False),
        sa.PrimaryKeyConstraint("ReviewID"),
        schema="SalesLT",
    )


def downgrade() -> None:
    op.drop_table("ProductReview", schema="SalesLT")

스크립트를 검토한 후 모든 대기 중인 마이그레이션을 적용하세요:

alembic upgrade head

pyodbc 방언과의 차이점

mssql+pyodbc에서 마이그레이션하는 경우, 두 드라이버가 동일한 ODBC 프레임워크를 기반으로 하므로 mssql-python 방언은 유사합니다. 주요 차이점:

주제	mssql+pyodbc	mssql+mssqlpython
ODBC 드라이버 설치	별도의 ODBC 드라이버(예: Microsoft SQL용 ODBC 드라이버 18)가 필요합니다.	드라이버는 번들로 제공됩니다. 별도의 ODBC 드라이버가 필요하지 않습니다.
연결 URL	mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server	mssql+mssqlpython://user:pass@host/db
fast_executemany	create_engine(..., fast_executemany=True)를 통해 지원됩니다.	적용할 수 없습니다. 드라이버는 배치 처리 성능을 내부적으로 담당합니다.
가용성	안정적이며 1.x부터 SQLAlchemy 에 포함되어 있습니다.	프리릴리즈 (SQLAlchemy 2.1.0b2+).

알려진 제한 사항

SQLAlchemy의 mssql-python 방언은 프리릴리즈 상태입니다. 생산 환경에서 사용하기 전에 다음 함의들을 이해하세요:

• API 변경: 메서드 서명, 예외 유형, 동작이 최종 안정 버전 이전에 변경될 수 있습니다. 항상 SQLAlchemy 버전을 특정 사전 릴리스 빌드(예: sqlalchemy==2.1.0b2)에 고정하고 업그레이드를 철저히 테스트하세요.

• 제한된 테스트: 이 방언은 안정 mssql+pyodbc 방언보다 커뮤니티 테스트가 적습니다. 예외적인 경우나 누락된 기능을 만날 수도 있습니다.

• 기능 격차: 일부 고급 ORM 또는 Core 기능은 작동하지 않을 수 있습니다. 프로젝트에 참여하기 전에 SQLAlchemy MSSQL 방언 문서를 참고하고 사용 사례를 테스트하세요.

• 지원 보장 없음: Microsoft와 SQLAlchemy는 최선의 노력 지원을 제공하지만, 안정 버전 출시 전까지 문제가 해결되지 않을 수 있습니다.

사전 출시 사용 시기:

• 개발 및 테스팅 환경
• 개념 증명 프로젝트
• 외부 ODBC 드라이버 의존성을 피하려면 mssql+pyodbc에서 마이그레이션하기
• API 변경에 대응하고 회귀 테스트를 수행할 수 있는 프로젝트들

사전 출시 시 사용하지 말아야 할 때:

• 엄격한 안정성 요건을 가진 생산 시스템
• 의존성 업데이트가 드문 다년간의 레거시 애플리케이션
• SQLAlchemy 2.1이 안정적인 GA에 도달하기 전까지의 중요한 비즈니스 워크로드

최신 사전 릴리스 방언 상태와 알려진 문제는 mssql-python GitHub 저장소를 확인해 보세요.

Troubleshooting

"'sqlalchemy.dialects.mssql.mssqlpython'이라는 이름의 모듈은 없습니다."

이 오류는 설치된 SQLAlchemy 버전에 mssql-python 방언이 포함되어 있지 않다는 뜻입니다. 2.1.0b2 이상을 사용 중인지 확인하세요:

pip install "sqlalchemy>=2.1.0b2"

연결 오류

create_engine는 성공하지만 쿼리가 실패하는 경우, mssql-python에 직접 연결해 연결 매개변수가 제대로 작동하는지 확인하세요:

import mssql_python

conn = mssql_python.connect(
    "Server=localhost;Database=<database>;UID=dbuser;PWD=<password>;Encrypt=yes"
)
cursor = conn.cursor()
cursor.execute("SELECT 1")
print(cursor.fetchone())
conn.close()

직접 연결은 되지만 SQLAlchemy 가 작동하지 않는다면, 비밀번호나 서버 이름 내 특수 문자로 URL 인코딩 문제를 확인해 보세요.

관련 콘텐츠

• MSSQL-Python 연결 문자열
• MSSQL-파이썬 연결 관리
• FastAPI 통합
• 플라스크 통합
• SQLAlchemy MSSQL 방언 문서화