SQL Server용 Django 백엔드 - mssql-django

mssql-django는 SQL Server, Azure SQL Database, Azure SQL Managed Instance 및 Microsoft Fabric의 SQL 데이터베이스를 위한 Microsoft의 Django용 데이터베이스 백엔드입니다. 연결하려면 Django DATABASES 구성에서 ENGINE을(를) "mssql"(으)로 설정하세요. 백 엔드는 pyodbc 및 Microsoft ODBC Driver for SQL Server 기반으로 하며 Django 3.2~6.0, Python 3.8~ 3.14, 2016~2025년 SQL Server 지원합니다.

시작점 선택

• Django 프로젝트가 SQL Server 빠르게 통신하도록 하려면 빠른 시작: Django를 SQL Server 연결합니다.
• 암호 없는 인증을 사용하여 Django를 Azure SQL 연결하려면 Microsoft Entra 인증 및 구성 참조로 시작합니다.
• 기존 SQL Server 데이터베이스를 Django로 가져오려면 inspectdb를 사용하여 리버스 엔지니어링 모델로 이동합니다.
• Django 사이트를 Azure에 배포하려면 Azure App Service에 배포 및 컨테이너 및 로컬 개발로 이동합니다.
• 다른 Django 백 엔드 또는 데이터베이스에서 마이그레이션하려면 django-mssql-backend에서 마이그레이션, 다른 데이터베이스에서 마이그레이션 또는 PostgreSQL에서 마이그레이션으로 이동합니다.

Azure SQL 대한 프로덕션 기준

이 코드 조각을 프로덕션 지향 Azure SQL 구성의 시작점으로 사용합니다. 4개의 파일 settings.py (Django 데이터베이스 구성, 미들웨어 등록 및 로깅), myproject/retry.py (일시적인 오류 카탈로그 및 retry_on_transient 데코레이터), myproject/middleware.py (요청 수준 재시도 미들웨어) 및 myapp/views.py (예: 트랜잭션 뷰)를 결합합니다.

# settings.py

import logging.config
import os

DATABASES = {
    "default": {
        "ENGINE": "mssql",
        "NAME": os.environ["SQL_DATABASE"],          # for example, appdb
        "HOST": os.environ["SQL_SERVER"],            # for example, contoso.database.windows.net
        "PORT": "1433",
        "CONN_MAX_AGE": 300,                         # reuse pooled connections for 5 minutes
        "CONN_HEALTH_CHECKS": True,                  # validate connections before reuse (Django 4.1 and later)
        "ATOMIC_REQUESTS": False,                    # wrap mutating views in transactions explicitly (see the following view example)
        "OPTIONS": {
            "driver": "ODBC Driver 18 for SQL Server",
            "extra_params": (
                "Authentication=ActiveDirectoryMsi;"
                "Encrypt=yes;"
                "TrustServerCertificate=no;"
                # ODBC driver reconnects connections dropped while idle.
                "ConnectRetryCount=3;"
                "ConnectRetryInterval=10;"
            ),
            # Backend-level retry for the initial connect call. Complements
            # ConnectRetryCount, which only covers idle drops on an
            # already-established connection.
            # See Retry logic and connection resilience for the recognized error list.
            "connection_retries": 3,
            "connection_retry_backoff_time": 5,
        },
    },
}

MIDDLEWARE = [
    # Defined in myproject/middleware.py. Catches transient OperationalErrors
    # and retries the request. Add "1205" (deadlock victim) and "1222"
    # (lock-request timeout) to TRANSIENT_ERROR_CODES to also retry
    # statement-level failures.
    "myproject.middleware.DatabaseRetryMiddleware",
    "django.middleware.security.SecurityMiddleware",
    # ... your other middleware
]

LOGGING_CONFIG = None
logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "format": (
                '{"time":"%(asctime)s","level":"%(levelname)s",'
                '"logger":"%(name)s","message":"%(message)s"}'
            ),
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "json",
        },
    },
    "loggers": {
        "django.db.backends": {
            "handlers": ["console"],
            "level": "WARNING",      # raise to INFO or DEBUG to capture SQL
            "propagate": False,
        },
        "django.request": {
            "handlers": ["console"],
            "level": "WARNING",
            "propagate": False,
        },
        "mssql": {
            "handlers": ["console"],
            "level": "INFO",
            "propagate": False,
        },
    },
})

myproject/retry.py에서 공유 임시 오류 카탈로그와 retry_on_transient 데코레이터를 정의합니다:

# myproject/retry.py
import functools
import logging
import random
import re
import time

from django.db import OperationalError, connection

logger = logging.getLogger(__name__)

TRANSIENT_ERROR_CODES = {
    "64", "233", "4221",
    "10053", "10054", "10928", "10929",
    "40197", "40501", "40613",
    "49918", "49919", "49920",
    # Add "4060" only if targeting Azure SQL with geo-replication failover.
    # Add "1205" (deadlock victim) and "1222" (lock-request timeout) to
    # also retry statement-level failures.
}

# Microsoft ODBC driver formats native error codes as "(<number>)" in the
# message. Parenthesized matches avoid false positives for short codes like "64".
_CODE_RE = re.compile(r"\((\d+)\)")


def is_transient(error):
    codes_in_message = set(_CODE_RE.findall(str(error)))
    return bool(codes_in_message & TRANSIENT_ERROR_CODES)


def retry_on_transient(max_retries=3, base_delay=1, max_delay=30):
    """Retry on transient database errors with exponential backoff and full jitter."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except OperationalError as e:
                    if attempt < max_retries and is_transient(e):
                        capped = min(max_delay, base_delay * (2 ** attempt))
                        delay = random.uniform(0, capped)
                        logger.warning(
                            "Transient error in %s (attempt %d/%d), retrying in %.2fs: %s",
                            func.__name__, attempt + 1, max_retries, delay, e
                        )
                        connection.close()
                        time.sleep(delay)
                        continue
                    raise
        return wrapper
    return decorator

에서 요청 수준 미들웨어를 정의합니다 myproject/middleware.py. is_transient를 재사용하므로 두 계층 모두 동일한 오류 코드 집합을 인식합니다:

# myproject/middleware.py
import logging
import random
import time

from django.db import OperationalError, connection

from myproject.retry import is_transient

logger = logging.getLogger(__name__)


class DatabaseRetryMiddleware:
    """Retry the entire request on transient database errors."""

    def __init__(self, get_response):
        self.get_response = get_response
        self.max_retries = 3
        self.base_delay = 1   # seconds; doubled each attempt
        self.max_delay = 30   # cap on a single sleep, regardless of attempt

    def __call__(self, request):
        for attempt in range(self.max_retries + 1):
            try:
                return self.get_response(request)
            except OperationalError as e:
                if attempt < self.max_retries and is_transient(e):
                    capped = min(self.max_delay, self.base_delay * (2 ** attempt))
                    delay = random.uniform(0, capped)
                    logger.warning(
                        "Transient DB error (attempt %d/%d), retrying in %.2fs: %s",
                        attempt + 1, self.max_retries, delay, e
                    )
                    connection.close()
                    time.sleep(delay)
                    continue
                raise

ATOMIC_REQUESTS이(가) False이므로, 변경하는 뷰는 자체 트랜잭션을 열어야 합니다. 각 재시도에서 atomic()@retry_on_transient 새 트랜잭션을 실행되도록 블록을 래핑합니다.

# myapp/views.py
from django.db import transaction
from django.http import JsonResponse

from myproject.retry import retry_on_transient
from .models import Order


# Exponential backoff with full jitter: sleeps random within [0,2], [0,4], [0,8] seconds.
@retry_on_transient(max_retries=3, base_delay=2)
def submit_order(request, order_id):
    with transaction.atomic():
        order = Order.objects.select_for_update().get(id=order_id)
        order.status = "submitted"
        order.save()
    return JsonResponse({"id": order.id, "status": order.status})

이 구성의 각 부분에 대한 자세한 내용은 구성 참조, 연결 옵션, 연결 풀링, 재시도 논리 및 연결 복원력 및Microsoft Entra 인증을 참조하세요.

주요 기능

• 드롭인 Django 백엔드: ENGINE을 "mssql"(으)로 설정하면 Django의 ORM, 마이그레이션, 관리자 사이트 및 관리 명령이 SQL Server에서 작동합니다.
• pyodbc 및 ODBC Driver 18을 기반으로 빌드됨: 기본적으로 TLS로 암호화된 연결이며 Windows, Linux 및 macOS에서 광범위한 플랫폼 지원을 제공합니다.
• 와이드 버전 매트릭스: Django 3.2~ 6.0, Python 3.8~ 3.14, SQL Server 2016~2025.
• Microsoft Entra ID 인증: extra_params를 통한 관리되는 ID, 서비스 주체, 대화형 및 통합 흐름의 암호 없는 연결.
• Django 마이그레이션: SQL Server 전용 열 형식을 포함하여 SQL Server에 대한 스키마 마이그레이션.
• JSONField 지원: JSONFieldnvarchar(max) 저장소와 Django 조회 기능을 기반으로 하는 기본 지원.
• Always Encrypted: 중요한 열에 대한 클라이언트 쪽 암호화입니다.
• 대량 작업: SQL Server를 대상으로 bulk_create 및 bulk_update를 적절한 배치 크기로 수행합니다.
• 일시적 재시도: 연결 및 쿼리 실행 중 일반적인 Azure SQL 일시적인 오류에 대한 기본 제공 처리입니다.
• inspectdb: 기존 SQL Server 스키마에서 Django 모델을 생성합니다.

Get started

구성 및 연결

모델, 마이그레이션 및 데이터 형식

데이터 쿼리 및 작업

배포, 테스트 및 튜닝

mssql-django로 마이그레이션

관련 작업

관련 콘텐츠

• mssql-django on GitHub
• PyPI의 mssql-django
• mssql-django wiki
• 릴리스 노트
• Django 설명서
• Azure Python 개발자 센터