python - 在sqlalchemy中使用自定义列类型时如何查找列名

Question

我正在尝试在 sqlalchemy 中定义我自己的列类型:

class NonNegativeInt(TypeDecorator):
    impl = Integer

    def process_bind_param(self, value, dialect):
        if value is not None:
            if not isinstance(value, int) or value < 0:
                raise TypeError('Expect non-negative integer, find %s instead.' % value)
        return value

class Game(Model):
    score = Column(NonNegativeInt, default=0)

当我尝试将负整数绑定(bind)到 NonNegativeInt 列时，例如分数，它会按预期引发错误:

sqlalchemy.exc.StatementError: (exceptions.TypeError) Expect non-negative integer, find -1 instead.

但是没有指定列的名称，所以当列很多的时候，调试起来不太方便。使用原始 Integer 类型时，我得到了更具体的错误消息:

sqlalchemy.exc.InternalError: (pymysql.err.InternalError) (1366, u"Incorrect integer value: 'aaaa' for column 'score' at row 1")

如何找到有错误的列的名称(例如分数)？此外，是否可以在分配错误值时引发错误:game.score = -1，而不是提交时？

Answer 1

这是一个使用 inspect 模块的解决方案。由于它使用检查模块，因此它可能对 SA 升级不是很稳健(如文档字符串中所述)。

此代码假定该值作为变量 key 在 TypeDecorator.process_bind_param 之上两个堆栈级别可用(因此索引 [3]在 get_column_name 中)。如果堆栈深度或变量名称在 SA 内部发生变化，则此代码将中断，这就是它使用“bare except”的原因。但是，异常将被记录为包含堆栈跟踪的 DEBUG 消息，因此它应该保持可调试状态。

不过，我确实认为 SA 的那部分不太可能发生变化。

import logging

LOG = logging.getLogger(__name__)


def get_column_name() -> str:
    """
    Tries to determine the field/column-name from inside a TypeDecorator
    method.

    This is done by inspecting the current Python stack and may not always work
    and is not guaranteed to survive SQLAlchemy upgrades. In that case, it will
    return ``"<unknown>"`` as value so you should not use this for any other
    use-case than debugging or logging!
    """
    from inspect import getmembers, getouterframes, currentframe
    frame = currentframe()
    column_name = '<unknown>'
    try:
        target_frame = getouterframes(frame)[3]
        frame_locals = dict(getmembers(target_frame))['frame'].f_locals
        column_name = frame_locals['key']
    except:  # pylint: disable=bare-except
        LOG.debug('Unable to retrieve the column name', exc_info=True)
    return column_name

然后可以在 TypeDecorator 中使用此函数:

class NonNegativeInt(TypeDecorator):
    impl = Integer

    def process_bind_param(self, value, dialect):
        column = get_column_name()
        if value is not None:
            if not isinstance(value, int) or value < 0:
                raise TypeError('Expect non-negative integer in column %r, find %s instead.' % (column, value))
        return value