SQLAlchemy 自定义布尔值

Question

我有一个以这种方式实现布尔逻辑的 Oracle 数据库：

MYCOL NUMBER(1,0)

0 = False
-1 or 1 = True
NULL = None

使用默认的 SQLAlchemy 布尔列，-1 表值与 False 或 True 不匹配。 是否可以创建一个 SQLAlchemy 自定义类型，将此类列映射到 bool python 属性？

困难在于有多个与 true 关联的值。 例如：

session.query(User).filter_by(active=True)

应该生成以下查询之一：

select * from user where active != 0;
select * from user where active in (-1, 1);
select * from user where 1 = (CASE WHEN active != 0 THEN 1 ELSE active end);

Answer 1

此实现似乎适用于基本查询。
Case when 表达式似乎是最简单的解决方案，因为运算符逻辑没有改变。

class CustomBoolean(sqlalchemy.types.TypeDecorator):
    """0 = False
    -1 or 1 = True
    NULL = None
    By default, SQLAlchemy only supports 0 and 1"""

    impl = Boolean
    cache_ok = True

    # In select or group by clause, replace col with expression: (CASE WHEN col != 0 THEN 1 ELSE col end)
    def column_expression(self, col):
        expr = col.expr
        if isinstance(expr.type, CustomBoolean):
            res = sqlalchemy.case(
                (type_coerce(expr, Integer()) != sqlalchemy.literal_column("0", type_=Integer), sqlalchemy.literal_column("1", type_=Integer)),
                else_=type_coerce(expr, Integer())
            )
            return type_coerce(res, Boolean())

        expr.type = Boolean()
        return expr

    # In where clause, replace col with expression: (CASE WHEN col != 0 THEN 1 ELSE col end)
    class comparator_factory(Boolean.Comparator):
        def operate(self, op, other, **kwargs):
            op1 = self
            if hasattr(op1, "__clause_element__"):
                clause = op1.__clause_element__()
                if isinstance(clause.type, CustomBoolean):
                    op1 = self.normalise(clause)

            op2 = other
            if hasattr(op2, "__clause_element__"):
                clause = op2.__clause_element__()
                if isinstance(clause.type, CustomBoolean):
                    op2 = self.normalise(clause)

            return op(
                op1, op2, **kwargs
            )

        @staticmethod
        def normalise(expr):
            res = sqlalchemy.case(
                (type_coerce(expr, Integer()) != sqlalchemy.literal_column("0", type_=Integer), sqlalchemy.literal_column("1", type_=Integer)),
                else_=type_coerce(expr, Integer())
            )
            return type_coerce(res, Boolean())