LocalProxy objects in Flask-SQLAlchemy

Question

I use many werkzeug.local.LocalProxy objects in my Flask application. They should be ideal for objects, but this is actually not the case, as they do not respond to type () or instanceof () correctly.

They do not like SQLAlchemy at all. If I create LocalProxy to write SQLAlchemy, SQLAlchemy considers this None. If you pass it LocalProxy to a simpler type, it just says that it is the wrong type.

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

How do you guys deal with this problem? Just call _get_current_object () a lot? It would be great if SQLAlchemy or Flask-SQLAlchemy could automatically handle these LocalProxy objects more gracefully, especially considering that Flask-Login uses them and almost everyone uses it, right?

I am considering adding this function to my project to deal with it, and wrapping any of my local processes in it before passing them to sqlalchemy:

from werkzeug.local import LocalProxy def real(obj): if isinstance(obj, LocalProxy): return obj._get_current_object() return obj 

Answer 1

I fixed the drivers used by SQLAlchemy , but I'm afraid this is not the most general solution.

 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 

The source can be found here .