Microsoft Python SQL Server용 드라이버 - mssql-python

mssql-python는 SQL Server, Azure SQL Database, Azure SQL Managed Instance 및 Microsoft Fabric의 SQL 데이터베이스용 Microsoft의 Python 드라이버입니다. 직접 데이터베이스 연결(DDBC)을 사용해 외부 드라이버 관리자를 설치하지 않고도 연결할 수 있습니다. 이 드라이버는 Python 3.10 이상을 지원하며, Python Database API Specification 2.0을 준수하는 동시에 일상 개발을 위한 Python 친화적인 개선 사항을 추가합니다.

시작점 선택

• SQL Server, Azure SQL Database, 또는 Microsoft Fabric의 SQL 데이터베이스에 연결해 첫 쿼리를 실행하려면 Quickstart부터 시작하세요: mssql-python 드라이버로 연결하세요.
• 로컬 SQL Server 컨테이너를 만들려면 Container와 로컬 개발부터 시작하세요.
• 비밀번호 없는 인증으로 Azure SQL에 연결하려면 Microsoft Entra 인증과 Connection 문자열부터 시작하세요.
• 데이터를 인터랙티브하게 탐색하려면 Jupyter Notebook에서 Connect를 사용하거나 Rapid 프로토타이핑을 시작하세요.
• 대량의 데이터를 효율적으로 이동하려면 대량 복사 작업 이나 대량 복사 퀵스타트를 선택하세요.
• 다른 드라이버에서 마이그레이션하려면 pyodbc에서 마이그레이션, pymssql에서 마이그레이션, SQLite에서 마이그레이션, 또는 PostgreSQL에서 마이그레이션을 선택하세요.

Azure SQL 대한 프로덕션 기준

이 샘플은 관리 신원 및 Tabular Data Stream(TDS) 8.0 암호화를 사용하는 Azure SQL에 연결됩니다. 환경 변수에서 연결 설정을 읽고 별도의 로그인 및 쿼리 타임아웃을 설정합니다. 각 재시도마다 실패한 연결 객체를 재사용하는 대신 connect()를 다시 호출합니다. 컨텍스트 매니저는 각 시도 후 연결과 커서를 해제합니다.

이 헬퍼는 반복해도 안전한 멱등성 읽기 쿼리에만 사용하세요. 개별 기록이나 부분적으로 완료된 거래를 다시 재생하는 데 사용하지 마세요.

연결 문자열의 ConnectRetryCount 및 ConnectRetryInterval 키워드는 SQL Server의 유휴 연결 복원 기능을 활성화합니다. 즉, 드라이버가 끊어진 유휴 연결을 자동으로 다시 연결합니다. 이는 이 샘플에서 애플리케이션 레벨 재시도와는 다릅니다. 재시도는 교착 상태나 쿼리 타임아웃 같은 일시적 오류가 발생해 실패한 쿼리 를 재시도합니다. 두 가지 모두 상호 보완적이니 둘 다 유지하세요.

Azure 오류 검사는 진단에 오류 번호가 빠져 있어도 표준 영어 오류 40501 메시지와 일치합니다. 또한 알려진 숫자가 있으면 ddbc_error를 확인합니다. 매칭은 가능한 범위에서 수행되며, 지역화되었거나 형식이 다른 메시지는 일치하지 않을 수 있습니다. 자세한 내용은 일시적 연결 오류를 참조하세요.

import logging
import os
import re
import time

import mssql_python

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("app")

TRANSIENT_DRIVER_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
    "Serialization failure",
    "Statement completion unknown",
})

AZURE_THROTTLING_ERRORS = frozenset({
    40197, 40501, 40540, 40613, 40680, 49918, 49919, 49920, 10928, 10929,
})

_ERROR_NUMBER_RE = re.compile(r"\b(?:Error|Msg)\s+(\d+)\b")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


def is_azure_throttling(error: BaseException) -> bool:
    """Best-effort detection of Azure SQL throttling in ProgrammingError text."""
    if not isinstance(error, mssql_python.ProgrammingError):
        return False
    ddbc_text = getattr(error, "ddbc_error", "") or ""
    return (
        _AZURE_40501_MESSAGE in ddbc_text
        or any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))
    )


def query_with_retry(
    conn_str: str,
    sql: str,
    *params,
    max_attempts: int = 3,
    login_timeout_s: int = 5,
    query_timeout_s: int = 10,
) -> list:
    """Fetch an idempotent read query, reconnecting before each retry."""
    for attempt in range(1, max_attempts + 1):
        try:
            with mssql_python.connect(conn_str, timeout=login_timeout_s) as conn:
                conn.timeout = query_timeout_s
                with conn.cursor() as cursor:
                    cursor.execute(sql, *params)
                    rows = cursor.fetchall()
            logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return rows
        except mssql_python.Error as exc:
            azure_throttling = is_azure_throttling(exc)
            transient = (
                isinstance(exc, mssql_python.OperationalError)
                and exc.driver_error in TRANSIENT_DRIVER_ERRORS
            )
            if not (transient or azure_throttling) or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = (10 if azure_throttling else 1) * (2 ** (attempt - 1))
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise ValueError("max_attempts must be at least 1")


def main() -> None:
    # Read configuration from the environment; never hard-code secrets.
    server = os.environ["SQL_SERVER"]      # for example, <server>.database.windows.net
    database = os.environ["SQL_DATABASE"]  # for example, AdventureWorks
    client_id = os.getenv("AZURE_CLIENT_ID")  # set for a user-assigned managed identity

    # Authenticate with the workload's managed identity over TDS 8.0 encryption.
    # ConnectRetryCount/ConnectRetryInterval transparently reconnect a dropped
    # idle connection; they don't replay a failed query.
    conn_str = (
        f"Server={server};"
        f"Database={database};"
        "Authentication=ActiveDirectoryMsi;"
        "Encrypt=strict;"
        "ConnectRetryCount=3;"
        "ConnectRetryInterval=10;"
        # Parallel dials to all resolved IPs; safe on single-IP targets.
        "MultiSubnetFailover=Yes;"
    )
    if client_id:
        conn_str += f"UID={client_id};"

    query = """
        SELECT TOP 10
            p.BusinessEntityID,
            p.FirstName,
            p.LastName
        FROM Person.Person AS p
        ORDER BY p.BusinessEntityID;
    """

    try:
        rows = query_with_retry(conn_str, query)
        for business_entity_id, first_name, last_name in rows:
            print(f"{business_entity_id}\t{first_name}\t{last_name}")
    except mssql_python.Error:
        logger.exception("query failed")
        raise


if __name__ == "__main__":
    main()

이 샘플의 각 문제에 대한 더 깊은 안내는 Microsoft Entra 인증, 연결 풀링, 암호화 및 인증서, 재시도 논리, 오류 처리를 참조하세요.

주요 기능

• PEP 249 준수: 표준 connect, cursor, execute, fetch* 인터페이스, 그리고 파이썬 확장 기능.
• 직접 데이터베이스 연결(DDBC): 외부 드라이버 관리자가 필요하지 않습니다. mssql-python을 설치하면 바로 연결할 준비가 됩니다.
• Microsoft Entra ID 인증: 관리 신원 및 서비스 주체를 포함한 인증 모드에 대한 내장 지원.
• SQL Server 및 Windows 인증: 지원 플랫폼에서의 SQL 로그인, Kerberos, Windows 싱글 사인온(SSO).
• 벌크 복사: 네이티브 TDS 프로토콜을 지원하는, 대량 데이터 로드를 위한 고성능 대량 삽입.
• 기본 데이터 형식 지원: JSON, XML, 공간, 희소 열, datetimeoffset, decimal/money를 정밀하게 처리합니다.
• Apache Arrow 통합: pandas, Polars, DuckDB와 빠른 데이터 교환을 위한 복사 없는 결과 집합.
• 비동기 패턴: ThreadPoolExecutor 우회 방법을 통해 asyncio 기반 애플리케이션 및 FastAPI에서 드라이버를 사용할 수 있습니다. 통합 패턴은 비동기 패턴 을 참조하세요.
• TLS가 기본값: 네이티브 ODBC 공급자 사용 시 TLS 암호화 및 인증서 검증이 기본적으로 켜져 있습니다. Encrypt=strict를 설정하면 TDS 8.0 암호화를 사용할 수 있습니다.

시작하기

조항	설명
설치	Python 환경을 설치하고 검증하세요mssql-python.
퀵스타트: mssql-python과 연결하세요	로컬 또는 테스트 SQL Server 인스턴스에 연결해서 첫 쿼리를 실행하세요.
퀵스타트: Jupyter Notebook에서 연결	노트북 내에서 mssql-python을 사용해 인터랙티브 데이터 탐색을 하세요.
퀵스타트: 대량 복사	대용량 데이터 세트를 대량 복사 API를 사용해 SQL Server로 옮기세요.
빠른 시작: 신속 프로토타이핑	작은 스크립트와 개념 증명을 빠르게 만들어 보세요.
빠른 시작: 반복 배치	SQL과 연동하는 Python 애플리케이션을 패키징, 구성, 배포하세요.
아파치 애로우 퀵스타트	분석 워크플로우를 위해 쿼리 결과를 Apache Arrow 테이블로 가져오세요.

구성 및 인증

조항	설명
mssql-odbc provider (Preview)	프리뷰 네이티브 제공업체를 선택하고 지원되는 기능과 알려진 문제를 검토하세요.
연결 문자열	연결 문자열 문법, 일반적인 키워드, 그리고 예시들.
프로그래밍적으로 연결 문자열을 구축하기	구성 및 비밀 설정에서 연결 문자열을 안전하게 작성하세요.
연결 관리	연결을 깨끗하게 열고 재사용하며 닫으세요.
연결 풀링	수영장 튜닝, 수명, 재사용 패턴.
암호화 및 인증서	TLS 암호화 모드, 인증서 검증, 그리고 TDS 8.0.
Microsoft Entra 인증	Azure SQL에서 관리된 신원, 서비스 주체, 인터랙티브, 디바이스 코드 플로우를 포함한 비밀번호 없는 인증.
보안 모범 사례	매개 변수화, 비밀 관리, 최소 권한 및 암호화.
가용성 그룹	Always On 가용성 그룹과 읽기 전용 복제본에 연결하세요.

데이터 작업

조항	설명
쿼리 실행	execute, executemany, 다중 명령 배치, 결과 집합.
데이터 검색	fetchone, , fetchmany, fetchall그리고 스트리밍 패턴.
매개 변수가 있는 쿼리	SQL 인젝션을 방지하기 위해 매개변수를 안전하게 바인딩하세요.
저장 프로시저	프로시저 호출, 출력 매개변수 읽기, 결과 집합 처리 등.
커서 관리	커서 수명, 스크롤, 배열 크기 조정.
행 객체	인덱스, 이름, 또는 매핑으로 행에 접근할 수 있습니다.
트랜잭션 관리	커밋, 롤백, 세이브포인트 및 격리 수준.
페이지 나누기	큰 결과 세트에 대한 키셋 및 오프셋 페이지네이션 패턴.
오류 처리	mssql_python.Error, DatabaseError, 그리고 SQL Server 오류 구조를 포함합니다.
재시도 논리	일시적 오류를 감지하고 지수 백오프로 재시도하세요.

SQL Server 데이터 타입 및 기능

조항	설명
데이터 형식 매핑	SQL Server-to-Python 타입 테이블 및 변환 규칙.
날짜 처리	datetime, , datetime2, datetimeoffset, 그리고 시간대 고려사항.
십진과 화폐 유형	정확한 숫자 유형과 decimal.Decimal 정밀도.
문자열 및 유니코드 데이터	varchar, nvarchar, 콜레이션, 코드 페이지 등이 포함됩니다.
NULL 처리	삼중값 논리, 센티넬, 판다 상호운용.
이진 데이터	varbinary, image, 그리고 대형 물체를 스트리밍하는 것들입니다.
맞춤형 타입 변환기	맞춤형 타입용 레지스터 입력 및 출력 변환기.
대량 복사 작업	대량 복사 API를 이용한 고처리량 삽입.
JSON 데이터	FOR JSON 및 OPENJSON를 사용하여 JSON을 저장, 쿼리 및 분해합니다.
XML 데이터	xml 데이터 타입, XPath 및 XQuery를 다룹니다.
공간 데이터	geometry그리고 geography Python의 타입을 사용합니다.
스파스 열	와이드 테이블용 희소 열 및 열 집합.
스키마 검색	데이터베이스, 표, 열, 인덱스를 점검하세요.

Python 도구 및 프레임워크와 통합하세요

조항	설명
Apache Arrow 통합	결과를 Arrow 테이블로 가져오면 제로 복사 분석을 할 수 있습니다.
판다스 통합	쿼리 결과를 DataFrame에 불러와서 다시 쓰세요.
극지 적분	컬럼 작업 작업에는 mssql-python과 함께 Polars를 사용하세요.
DuckDB 통합	로컬 DuckDB 테이블과 함께 SQL Server 데이터를 쿼리하세요.
FastAPI 통합	mssql-python을 FastAPI 서비스에 연결하세요.
FastAPI 애플리케이션 테스트 및 배포	FastAPI 서비스에 대해 풀링, 오류, 인증, 테스트 및 배포 설정을 설정하세요.
플라스크 통합	Flask 애플리케이션에서 mssql-python을 사용하세요.
비동기 패턴	mssql-python을 asyncio 및 스레드 풀과 결합하세요.
데이터 접근 및 분석 패턴	SQL 데이터 분석을 위해 커서 액세스, Arrow 추출, pandas, Polars 및 DuckDB 분석에 적합한 올바른 읽기 경로를 선택하세요.
데이터 로딩 및 이동 패턴	행 삽입, 대량 복사, MERGE 업서트, DataFrame 로딩, CSV 인젯에 적합한 쓰기 경로를 선택하세요.

배포 및 운영

조항	설명
컨테이너 및 로컬 개발	Python 애플리케이션이 SQL에 연결될 때 Docker 컨테이너, 개발 컨테이너, CI 파이프라인을 설정하세요.
성능 튜닝	풀 튜닝, 준비된 명세서, 배치 크기, 그리고 대량 사본.
Troubleshooting	증상이나 오류 메시지로 안내를 찾으세요.
설치 및 연결 문제 해결	설치, 연결, 컨테이너, CI 문제를 해결하세요.
쿼리, 데이터 및 운영 문제 해결	쿼리, 데이터 유형, 성능, 트랜잭션, 대량 복사 문제를 해결하세요.
모듈 구성	모듈 수준 설정, 로깅 훅, 기능 플래그.

mssql-python으로 마이그레이션하세요

조항	설명
pyodbc에서 마이그레이션하세요	pyodbc API와 연결 문자열을 mssql-python에 매핑하세요.
pymssql에서 마이그레이션하기	pymssql을 mssql-python으로 대체하면서 동작을 유지하세요.
SQLite에서 마이그레이션하기	로컬 SQLite 워크로드를 SQL Server나 Azure SQL로 옮기세요.
PostgreSQL에서 마이그레이션	MSSQL-python을 이용해 PostgreSQL에서 SQL Server로 전환하는 Python 개발자를 위한 원스톱 가이드.

기능 요청

기능을 요청하려면 mssql-python GitHub 저장소에서 이슈를 열어보세요.

관련 콘텐츠

• mssql-python 수명주기 지원
• mssql-python의 새로운 소식
• GitHub의 mssql-python
• PyPI에서 mssql-python
• 릴리스 노트
• 로드맵
• Azure Python 개발자 센터
• Python 데이터베이스 API 명세 2.0