python - 查找在 sqlalchemy 用户定义类型中发生的操作

Question

我们已经成功实现了用户定义的数据类型。事实上，有一种本地化策略可以将可本地化的值保存到 JSONB 字段中，并添加一个 LocalizedString 类型，我们在 postgresql 函数的帮助下使用 sqlAlchemy 用户定义类型实现获取/设置它们的值。

class LocalizedString(JSONB):

    def __init__(self):
        super(LocalizedString, self).__init__()


    def column_expression(self, colexpr):
        locale_id = thread_locale[threading.current_thread().ident]
        return func.delocalized(colexpr,locale_id)

    def bind_expression(self, bindvalue):
        locale_id = thread_locale[threading.current_thread().ident]
        val = type_coerce(bindvalue, String)
        return func.localized(locale_id, val)

问题是我必须查明在 bind_expression 中发生了哪个操作以在更新中添加以前的子项。

{'en' : 'Try'} 更新后 -> {'en' : 'Try', 'fr' : 'essayer'}

Answer 1

最后，我结束了像 before_update 和 after_update 这样的 session 事件。实际上我发现没有直接或间接的方法来检查当前的 Action /命令。这是设计使然，因为运行 bind_expression 和 column_expression 的阶段不同于命令创建阶段。也就是说，你甚至不知道在这个阶段可能会产生哪个命令。因此，我在 before_update 中保存了一个上下文绑定(bind)变量，然后在需要时使用它。

@event.listens_for(Base, 'before_update', propagate=True)
def receive_before_update(mapper, connection, target):
    if type(col.type) == LocalizedString and threading.current_thread().ident in thread_data:
        thread_data[threading.current_thread().ident].action = 'update'