Obiekty LocalProxy w Flask-SQLAlchemy

Question

Używam wielu obiektów werkzeug.local.LocalProxy w mojej aplikacji Flask. Mają być doskonałymi stand-inami dla obiektów, ale tak naprawdę nie są, ponieważ nie odpowiadają poprawnie na type() lub instanceof().

SQLAlchemy w ogóle ich nie lubi. Jeśli zrobię LocalProxy do rekordu SQLAlchemy, SQLAlchemy uważa go za Brak. Jeśli przekazuję go LocalProxy do prostszego typu, po prostu mówi, że jest to niewłaściwy typ.

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

Jak sobie z tym radzicie? Po prostu często dzwonisz do _get_current_object()? Byłoby całkiem fajnie, gdyby SQLAlchemy lub Flask-SQLAlchemy mogły z większą łatwością obsłużyć te obiekty LocalProxy, szczególnie biorąc pod uwagę, że Flask-Login używa ich i prawie każdy z nich to używa, prawda?

jestem rozważa dodanie tej funkcji do mojego projektu, aby sobie z tym poradzić, a owijanie żadnego z moich localproxies w nim przed przekazaniem ich do sqlalchemy:

from werkzeug.local import LocalProxy 

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

Answer 1

I poprawione sterowniki używane przez SQLAlchemy, ale obawiam się, że nie jest to rozwiązanie najbardziej ogólne.

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 

Oryginalne źródło można znaleźć here.