filter - usar - ¿Cuál es la mejor forma de acceso previo al usuario de filtro para las consultas sqlalchemy?

primeros pasos crm hubspot

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.

Aprecio tu ayuda. Gracias.

Answer 1

A continuación, se simplifica el constructor de consultas redefinido para filtrar todas las consultas de modelos (incluidas las relaciones). Puede pasarlo como parámetro sessionmaker a 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 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

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

Hace esto mediante un cambio de proxy a las propiedades Query._from_obj y QueryContext._froms , que es donde finalmente se establecen las tablas para seleccionar.