LocalProxy-Objekte in Flask-SQLAlchemy

Question

Ich benutze viele Objekte von werkzeug.local.LocalProxy in meiner Flask-App. Sie sollen perfekte Stellvertreter für Objekte sein, sind aber nicht wirklich, da sie nicht auf type() oder instanceof() richtig reagieren.

SQLAlchemy mag sie überhaupt nicht. Wenn ich ein LocalProxy zu einem SQLAlchemy-Eintrag mache, betrachtet SQLAlchemy es als None. Wenn ich einen LocalProxy an einen einfacheren Typ übergebe, sagt es nur, dass es der falsche Typ ist.

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

Wie geht es euch mit diesem Problem? Rufen Sie einfach _get_current_object() viel auf? Es wäre ziemlich cool, wenn SQLAlchemy oder Flask-SQLAlchemy diese LocalProxy-Objekte automatisch besser handhaben könnten, besonders wenn man bedenkt, dass Flask-Login sie verwendet, und so ziemlich jeder benutzt das, oder?

Ich betrachte diese Funktion zu meinem Projekt hinzufügen, damit umzugehen, und in ihm jeder meiner localproxies Einwickeln, bevor sie an sqlalchemy vorbei:

from werkzeug.local import LocalProxy 

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

Answer 1

ich die Treiber von SQLAlchemy verwendet geflickt, aber ich Angst, dass es nicht die allgemeinste Lösung ist.

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 

Ursprüngliche Quelle kann here gefunden werden.