¿Cuál es la mejor forma de acceder previamente al usuario de filtro para las consultas de sqlalchemy?

Question

He estado viendo las recetas de sqlalchemy en su wiki, pero no sé cuál es la mejor para implementar lo que estoy tratando de hacer.

Cada fila en mis tablas tiene un user_id asociado. En este momento, para cada consulta, realicé una consulta con el ID del usuario que está actualmente conectado, y luego realizo una consulta según los criterios en los que estoy interesado. Me preocupa que los desarrolladores olviden agregar este filtro a la consulta (un gran riesgo de seguridad) Por lo tanto, me gustaría establecer un filtro global basado en los derechos de administrador del usuario actual para filtrar lo que el usuario conectado podría ver.

Agradecemos su ayuda. Gracias.

Answer 1

A continuación se simplifica el constructor de consultas redefinido para filtrar todas las consultas del modelo (incluidas las relaciones). Puede pasarlo como parámetro query_cls al sessionmaker. El parámetro ID de usuario no necesita ser global en lo que respecta a la sesión cuando ya está disponible.

class HackedQuery(Query): 

    def get(self, ident): 
     # Use default implementation when there is no condition 
     if not self._criterion: 
      return Query.get(self, ident) 
     # Copied from Query implementation with some changes. 
     if hasattr(ident, '__composite_values__'): 
      ident = ident.__composite_values__() 
     mapper = self._only_mapper_zero(
        "get() can only be used against a single mapped class.") 
     key = mapper.identity_key_from_primary_key(ident) 
     if ident is None: 
      if key is not None: 
       ident = key[1] 
     else: 
      from sqlalchemy import util 
      ident = util.to_list(ident) 
     if ident is not None: 
      columns = list(mapper.primary_key) 
      if len(columns)!=len(ident): 
       raise TypeError("Number of values doen't match number " 
           'of columns in primary key') 
      params = {} 
      for column, value in zip(columns, ident): 
       params[column.key] = value 
      return self.filter_by(**params).first() 


def QueryPublic(entities, session=None): 
    # It's not directly related to the problem, but is useful too. 
    query = HackedQuery(entities, session).with_polymorphic('*') 
    # Version for several entities needs thorough testing, so we 
    # don't use it yet. 
    assert len(entities)==1, entities 
    cls = _class_to_mapper(entities[0]).class_ 
    public_condition = getattr(cls, 'public_condition', None) 
    if public_condition is not None: 
     query = query.filter(public_condition) 
    return query 

Funciona solo para consultas de modelo único, y hay mucho trabajo para hacerlo adecuado para otros casos. Me gustaría ver una versión elaborada ya que DEBE TENER funcionalidad para la mayoría de las aplicaciones web. Utiliza una condición fija almacenada en cada clase de modelo, por lo que debe modificarla según sus necesidades.

Answer 2

Aquí hay una implementación muy ingenua que supone que existe el atributo/propiedad self.current_user que el usuario ha almacenado.

class YourBaseRequestHandler(object): 

    @property 
    def current_user(self): 
     """The current user logged in.""" 
     pass 

    def query(self, session, entities): 
     """Use this method instead of :method:`Session.query() 
     <sqlalchemy.orm.session.Session.query>`. 

     """ 
     return session.query(entities).filter_by(user_id=self.current_user.id) 

Answer 3

me escribió una extensión SQLAlchemy que creo que hace lo que usted está describiendo: https://github.com/mwhite/multialchemy

Esto se logra mediante proxy a los cambios en las Query._from_obj y QueryContext._froms propiedades, que es donde las mesas para seleccionar de última instancia conseguir el sistema .