Flask-SQLAlchemy 中的 LocalProxy 对象

Posted 2023-03-16

【中文标题】Flask-SQLAlchemy 中的 LocalProxy 对象

【英文标题】：LocalProxy objects in Flask-SQLAlchemy

【发布时间】：2014-12-25 22:26:56

【问题描述】：

我在我的 Flask 应用程序中使用了很多 werkzeug.local.LocalProxy 对象。它们应该是对象的完美替代品，但实际上并非如此，因为它们不能正确响应 type() 或 instanceof()。

SQLAlchemy 根本不喜欢它们。如果我为 SQLAlchemy 记录创建 LocalProxy，则 SQLAlchemy 将其视为无。如果我将 LocalProxy 传递给更简单的类型，它只会说它是错误的类型。

Here's an example of Flask-SQLAlchemy having a bad time with LocalProxy.

你们是如何解决这个问题的？只是调用 _get_current_object() 很多？如果 SQLAlchemy 或 Flask-SQLAlchemy 可以更优雅地自动处理这些 LocalProxy 对象，那就太酷了，尤其是考虑到 Flask-Login 使用它们，并且几乎每个人都使用它，对吧？

我正在考虑将此函数添加到我的项目中来处理它，并在将我的任何本地代理传递给 sqlalchemy 之前将它们包装在其中：

from werkzeug.local import LocalProxy

def real(obj):
    if isinstance(obj, LocalProxy):
        return obj._get_current_object()
    return obj

【参考方案1】：

我修补了SQLAlchemy使用的驱动程序，但我担心这不是最通用的解决方案。

from flask_sqlalchemy import SQLAlchemy as FlaskSQLAlchemy
from sqlalchemy.engine import Engine
from werkzeug.local import LocalProxy


class SQLAlchemy(FlaskSQLAlchemy):
    """Implement or overide extension methods."""

    def apply_driver_hacks(self, app, info, options):
        """Called before engine creation."""
        # Don't forget to apply hacks defined on parent object.
        super(SQLAlchemy, self).apply_driver_hacks(app, info, options)

        if info.drivername == 'sqlite':
            from sqlite3 import register_adapter

            def adapt_proxy(proxy):
                """Get current object and try to adapt it again."""
                return proxy._get_current_object()

            register_adapter(LocalProxy, adapt_proxy)

        elif info.drivername == 'postgresql+psycopg2':  # pragma: no cover
            from psycopg2.extensions import adapt, register_adapter

            def adapt_proxy(proxy):
                """Get current object and try to adapt it again."""
                return adapt(proxy._get_current_object())

            register_adapter(LocalProxy, adapt_proxy)

        elif info.drivername == 'mysql+pymysql':  # pragma: no cover
            from pymysql import converters

            def escape_local_proxy(val, mapping):
                """Get current object and try to adapt it again."""
                return converters.escape_item(
                    val._get_current_object(),
                    self.engine.dialect.encoding,
                    mapping=mapping,
                )

            converters.encoders[LocalProxy] = escape_local_proxy

原文可以在here找到。