Django自定义复杂字段报错：amount_amount列重复创建

问题背景 #

需要实现支持复杂金额结构的Django自定义字段，满足Ledger.objects.filter(amount_isDebit=True).aggregate(Sum('amount_amount'))这类查询需求，但当前实现会触发django.db.utils.OperationalError: duplicate column name: amount_amount错误，原因是自定义字段硬编码生成固定名称的数据库列，导致重复注册。

示例金额结构：

{
  "name": "asdfg",
  "amount": {
    "amount": 5000,
    "forexAmount": 10,
    "rateOfExchange": 500,
    "currency": "$",
    "isDebit": false
  }
}

错误原因分析 #

原AmountField的contribute_to_class方法硬编码生成amount_amount字段，每次字段被添加到模型时都会执行该逻辑。如果模型多次加载、继承或存在多个AmountField实例，就会重复创建同名字段，触发数据库列重复错误。

修正方案 #

1. 动态生成数据库字段名 #

基于自定义字段的属性名动态生成关联数据库列名，避免硬编码冲突。

2. 完善描述符逻辑 #

实现Amount对象与数据库字段的双向映射，保持模型层的易用性。

3. 拆分金额结构到独立列 #

将复杂金额的每个属性映射为独立数据库字段，确保ORM查询和聚合的高效性。

修正后的完整代码 #

金额数据类 #

class Amount:
    amount: float
    forexAmount: float | None
    rateOfExchange: float | None
    currency: str | None
    isDebit: bool

    def __init__(self, amount, isDebit, currency=None, forexAmount=None, rateOfExchange=None) -> None:
        self.amount = amount
        self.forexAmount = forexAmount
        self.rateOfExchange = rateOfExchange
        self.isDebit = isDebit
        self.currency = currency

自定义描述符 #

from django.db.models.query_utils import DeferredAttribute

class AmountDescriptor(DeferredAttribute):
    def __init__(self, field, base_name):
        super().__init__(field)
        self.base_name = base_name

    def __get__(self, instance, cls=None):
        if instance is None:
            return self
        # 从数据库字段组装Amount对象
        amount_val = getattr(instance, f"{self.base_name}_amount")
        if amount_val is None:
            return None
        
        return Amount(
            amount=amount_val,
            isDebit=getattr(instance, f"{self.base_name}_isDebit"),
            currency=getattr(instance, f"{self.base_name}_currency"),
            forexAmount=getattr(instance, f"{self.base_name}_forexAmount"),
            rateOfExchange=getattr(instance, f"{self.base_name}_rateOfExchange")
        )
    
    def __set__(self, instance, value):
        if isinstance(value, Amount):
            # 将Amount对象拆解到对应数据库字段
            setattr(instance, f"{self.base_name}_amount", value.amount)
            setattr(instance, f"{self.base_name}_forexAmount", value.forexAmount)
            setattr(instance, f"{self.base_name}_rateOfExchange", value.rateOfExchange)
            setattr(instance, f"{self.base_name}_currency", value.currency)
            setattr(instance, f"{self.base_name}_isDebit", value.isDebit)
            instance.__dict__[self.field.attname] = value
        elif isinstance(value, dict):
            # 支持直接传入字典赋值
            self.__set__(instance, Amount(**value))
        else:
            super().__set__(instance, value)

自定义字段类 #

from typing import Any, Type
from django.db import models
from django.db.models import Model
from django.utils.translation import gettext_lazy as _

class AmountField(models.Field):
    descriptor_class = AmountDescriptor
    description = _("Tally Amount")

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
    
    def deconstruct(self):
        name, path, args, kwargs = super().deconstruct()
        return name, path, args, kwargs
    
    def from_db_value(self, value, expression, connection):
        return None
    
    def to_python(self, value):
        if isinstance(value, Amount):
            return value
        if isinstance(value, dict):
            return Amount(**value)
        return super().to_python(value)
    
    def get_prep_value(self, value):
        return None
    
    def contribute_to_class(self, cls: type[Model], name: str, private_only: bool = False) -> None:
        base_name = name
        # 动态创建关联数据库字段，避免重复
        field_map = {
            f"{base_name}_amount": models.DecimalField(decimal_places=5, max_digits=15, null=True, blank=True),
            f"{base_name}_forexAmount": models.DecimalField(decimal_places=5, max_digits=15, null=True, blank=True),
            f"{base_name}_rateOfExchange": models.DecimalField(decimal_places=5, max_digits=15, null=True, blank=True),
            f"{base_name}_currency": models.CharField(max_length=10, null=True, blank=True),
            f"{base_name}_isDebit": models.BooleanField(default=False)
        }

        for field_name, field_instance in field_map.items():
            if not hasattr(cls, field_name):
                field_instance.contribute_to_class(cls, field_name)
        
        super().contribute_to_class(cls, name)
        setattr(cls, name, self.descriptor_class(self, base_name))

模型使用 #

from django.db import models
class Ledger(models.Model):
    name = models.CharField(max_length=100)
    amount = AmountField()

使用说明 #

1. 查询与聚合：现在可以直接用Django ORM语法执行需求中的查询：

   # 过滤借方记录并汇总金额
   Ledger.objects.filter(amount_isDebit=True).aggregate(total=models.Sum('amount_amount'))
   

2. 迁移处理：删除之前错误生成的迁移文件，执行makemigrations和migrate生成新的数据库表结构。

内容的提问来源于stack exchange，提问作者sai vineeth