SqlAlchemy：基于货币的金额精度

Question

我的模型有两个字段 amount 和 currency。 amount 字段目前是一种自定义类型，它处理 Decimal 值，但将它们作为整数保存。 Decimal 类型看起来像这样：

class Decimal(sa.types.TypeDecorator):
    impl = sa.types.INTEGER

    # This was computed by `math.log(2 ** 64, 10) - 1`. This is the maximum power of 10 we can
    # store in a 64 bit integer.
    MAX_POW = 18

    def __init__(self, digits, decimal_places):
        if digits < 0 or decimal_places < 0:
            raise ValueError("Can't have negative amounts of digits or decimal places")
        if (digits + decimal_places) > self.MAX_POW:
            raise ValueError("We can't store such precision")
        super().__init__()
        self.digits = digits
        self.decimal_places = decimal_places

    def process_bind_param(self, value, dialect):
        if value is None:
            return value

        if isinstance(value, int):
            value = decimal.Decimal(value)
        if not isinstance(value, decimal.Decimal):
            raise TypeError("Values are expected to be python Decimals")

        if count_integer_part(value) > self.digits:
            raise ValueError("Can't store that many digits before the decimal point")
        if count_fractional_part(value) > self.decimal_places:
            raise ValueError("Can't store that many digits after the decimal point")

        integer = value.scaleb(self.decimal_places)
        assert integer % 1 == 0, "Integer should not have digits after the decimal point"
        return int(integer)

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        return decimal.Decimal(value).scaleb(-self.decimal_places)

    def __repr__(self):
        return f"Decimal({self.digits}, {self.decimal_places})"

目前所有持有货币值的字段都使用参数化的MoneyAmount

MoneyAmount = functools.partial(Decimal, digits=16, decimal_places=2)

我现在想支持具有两位以上小数位的货币。实际的小数位数应使用适当的查找表从currency 字段计算。我一直在尝试使用hybrid_property，但在定义hybrid_property.expression（或hybrid_property.comparator）时遇到了一些麻烦，我必须在SQL 表达式中定义相同的查找。

decimal_places: dict[str, int] = {
    "BHD": 3,
}

class Table(declarative_base()):
    __tablename__ = "test"

    def __init__(self, **kwargs):
        amount = kwargs.pop("amount", None)
        super().__init__(**kwargs)
        if amount is not None:
            self.amount = amount

    id = sa.Column(sa.Integer, primary_key=True)
    _amount = sa.Column(sa.Integer, nullable=False, default=0)
    currency = sa.Column(sa.String(3), nullable=False, default="")

    @hybrid_property
    def amount(self):
        places = decimal_places.get(self.currency, 2)
        return Decimal(self._amount).scaleb(-places)

    @amount.setter
    def amount(self, value):
        if not isinstance(value, Decimal):
            value = Decimal(value)
        places = decimal_places.get(self.currency, 2)
        integer = value.scaleb(places)
        self._amount = int(integer)

    @amount.expression
    def amount(cls):
        # TODO: What to do here?

我发现 this 旧博客帖子以某种方式处理相同的问题域，但使用数字/浮点类型来存储数据。 你们对如何解决这个问题有什么建议吗？

Answer 1

@amount.expression 可以按以下方式工作：

@amount.expression
def amount(cls):
    sub_queries = [
        select(literal(k).label("currency"), literal(v).label("decimal_places"))
        for k, v in decimal_places.items()
    ]
    cte = union_all(*sub_queries).cte("cte")

    return (
        cls._amount
        * sa.func.power(
            10,
            -sa.func.coalesce(
                select(cte.c.decimal_places)
                .filter(cte.c.currency == cls.currency)
                .scalar_subquery(),
                2,
            ),
        )
    ).label("amount")

请注意，要在sqlite 上工作（正如您在问题的标签中指出的那样），power 函数需要包含在构建中。或者，您可以使用 python 实现进行测试：

# sqlite:only: create power function
if engine.name == "sqlite":
    session.connection().connection.connection.create_function(
        "power", 2, lambda x, y: x ** y
    )

进一步说明：对于 postgresql，我会使用 values/cte 而不是 union_all，但 sqlite 不适用于使用它的子查询。

---

我认为这回答了您的问题，但这不是您要解决的更高级别任务的完整解决方案，因为返回的值不会具有相同的 quantum。