Django中如何自动CAST特定SQL_VARIANT字段为VARCHAR？

我之前在对接MSSQL的Django项目里遇到过完全一样的SQL_VARIANT字段JOIN问题，折腾了一阵后找到了自动处理CAST的方案，完美支持annotate()、select_related()这类需要JOIN的操作，分享给你：

核心思路 #

MSSQL中SQL_VARIANT与VARCHAR直接JOIN会因为类型不匹配返回空结果，我们需要让Django ORM在生成SQL时自动对SQL_VARIANT字段执行CAST转换。最优雅的方式是自定义一个Django字段类型，接管该字段的SQL生成逻辑。

步骤1：实现自定义字段 #

创建一个继承自CharField的自定义字段，重写字段在查询和JOIN时的SQL生成逻辑，自动添加CAST操作：

from django.db import models
from django.db.models import Func

# 定义CAST转换的SQL函数
class CastToVarchar(Func):
    function = 'CAST'
    template = "%(function)s(%(expressions)s AS VARCHAR(%(max_length)s))"
    
    def __init__(self, expression, max_length, **extra):
        super().__init__(expression, max_length=max_length, **extra)

# 自定义SQL_VARIANT转VARCHAR的字段
class SQLVariantCharField(models.CharField):
    def get_col(self, alias=None, output_field=None, lookup=None):
        # 当字段被查询或用于JOIN时，自动包裹CAST
        col = super().get_col(alias, output_field, lookup)
        return CastToVarchar(col, max_length=self.max_length)

    def get_db_prep_lookup(self, lookup_type, value, connection, prepared=False):
        # 处理过滤查询时的参数转换，确保类型匹配
        preped_value = super().get_db_prep_lookup(lookup_type, value, connection, prepared)
        return [CastToVarchar(val, max_length=self.max_length) if isinstance(val, str) else val for val in preped_value]

步骤2：修改模型字段 #

把Vendor模型中对应SQL_VARIANT类型的id字段替换为我们的自定义字段：

class Vendor(models.Model):
    # 替换原CharField为自定义的SQLVariantCharField
    id = SQLVariantCharField(db_column='Id', primary_key=True, max_length=20)
    name = models.CharField(db_column='Name', max_length=30)
    address = models.CharField(db_column='Address', max_length=30)
    city = models.CharField(db_column='City', max_length=30)
    contact = models.CharField(db_column='Contact', max_length=30)
    type = models.IntegerField(db_column='Type')
    category = models.IntegerField(db_column='Category')

    class Meta:
        managed = False
        db_table = 'Vendor'

    def __str__(self):
        return f'{self.id} | {self.name}'

步骤3：验证效果 #

现在你可以正常使用所有需要JOIN的Django ORM操作了：

比如执行select_related关联查询：

# 自动生成带CAST的JOIN条件，正确关联产品和供应商
products_with_vendor = Product.objects.select_related('vendor_id').all()

生成的SQL会自动包含：

ON CAST([Vendor].[Id] AS VARCHAR(20)) = [Product].[Vendor Id]

再比如用annotate统计每个供应商的产品数量：

from django.db.models import Count

vendor_product_counts = Vendor.objects.annotate(product_count=Count('products'))

这时候生成的GROUP BY和JOIN条件都会自动带上CAST，不会再返回空结果。

补充说明 #

• 因为你的表是managed=False，不需要生成迁移文件，直接修改模型即可生效。
• 自定义字段中的max_length要和你模型中定义的一致，确保CAST后的长度匹配VARCHAR字段的长度。
• 如果SQL_VARIANT字段存储的数字转字符串后有格式需求（比如保留前导零），可以把CAST换成CONVERT函数并指定格式参数调整。

内容的提问来源于stack exchange，提问作者Juan Carlos Fdez Hdez