适用于 SQL Server 的 Django 后端 - mssql-django

mssql-django 是 Microsoft 面向 SQL Server、Azure SQL 数据库、Azure SQL 托管实例 以及 Microsoft Fabric 中 SQL 数据库的 Django 数据库后端。 在 Django DATABASES 配置中将 ENGINE 设置为 "mssql" 以进行连接。 后端基于 pyodbc 和 Microsoft ODBC Driver for SQL Server，支持 Django 3.2 到 6.0、Python 3.8 到 3.14，SQL Server 2016 到 2025 年。

选择起点

• 若要让 Django 项目快速与SQL Server通信，请从快速入门开始：将 Django 连接到 SQL Server。
• 若要使用无密码身份验证将 Django 连接到Azure SQL，请从Microsoft Entra身份验证和配置参考开始。
• 若要将现有SQL Server数据库引入 Django，请使用 inspectdb 转到反向工程模型。
• 若要将 Django 站点部署到Azure，请转到“部署到Azure 应用服务”和“容器”和“本地开发”。
• 若要从另一个 Django 后端或数据库迁移，请转到 从 django-mssql-backend 迁移、 从其他数据库迁移或 从 PostgreSQL 迁移。

Azure SQL的生产基线

将此代码片段用作面向生产Azure SQL配置的起点。 它结合了四个文件： 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，会修改数据的视图必须开启各自的事务。 用 @retry_on_transient 包裹 atomic() 代码块，以便每次重试都会在一个新的事务中运行：

# 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 驱动程序 18 构建：默认情况下，TLS 加密的连接以及 Windows、Linux 和 macOS 上的广泛平台支持。
• 宽版本矩阵：Django 3.2 到 6.0，Python 3.8 到 3.14，SQL Server 2016 到 2025。
• Microsoft Entra ID 身份验证：通过 extra_params 使用托管标识、服务主体、交互式和集成流程实现无密码连接。
• Django 迁移：面向 SQL Server 的架构迁移，包括 SQL Server 特有的列类型。
• JSONField 支持：由 nvarchar(max) 存储和 Django 查找功能提供支持的原生 JSONField。
• Always Encrypted：敏感列的客户端加密。
• 批量操作：bulk_create 和 bulk_update，针对 SQL Server，并采用合理的批处理大小。
• 暂时性重试：在连接和查询执行期间针对常见Azure SQL暂时性错误的内置处理。
• inspectdb：从现有SQL Server架构生成 Django 模型。

开始

配置和连接

模型、迁移和数据类型

查询和使用数据

部署、测试和优化

迁移到 mssql-django

相关任务

相关内容

• GitHub上的 mssql-django
• PyPI 上的 mssql-django
• mssql-django wiki
• 发行说明
• Django 文档
• Azure Python 开发人员中心