2010-05-21 34 views
14

我一直在寻找他们的维基上的sqlalchemy食谱,但不知道哪一个是最好实施我想要做的。预过滤用户访问sqlalchemy查询的最佳方式是什么?

我表格中的每一行都有一个与它关联的user_id。现在,对于每个查询,我都会查询当前登录的用户的id,然后按照我感兴趣的标准进行查询。我担心的是开发人员可能会忘记将此过滤器添加到查询中(巨大的安全风险)。因此,我想根据当前用户的管理权限设置一个全局过滤器来过滤登录用户可以看到的内容。

感谢您的帮助。谢谢。

+3

你有没有想过使用数据库安全系统来实现这一目标?例如,Oracle通过其VPD功能提供此功能(http://www.oracle.com/technology/deploy/security/database-security/virtual-private-database/index.html)。您也可以在其他数据库中实现类似的功能(例如http://ddj.com/database/215900773和http://technet.microsoft.com/en-gb/library/cc966395.aspx)。只需谷歌“行级安全RDBSName”。 – stephan 2010-05-26 15:07:22

+0

,或在所有添加适当限制的表上设置视图或规则,并剥夺用户直接读取底层表的权限。 http://www.postgresql.org/docs/8.4/interactive/rules.html – 2010-09-23 21:42:54

回答

2

下面是简化的重定义查询构造函数,用于过滤所有模型查询(包括关系)。您可以将它传递给query_cls参数sessionmaker。只要会话在可用时构建,用户ID参数就不需要是全局的。

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 

它仅适用于单一模型查询,并且有很多工作使其适用于其他情况。我希望看到一个详尽的版本,因为它对大多数Web应用程序都必须具有功能。它使用存储在每个模型类中的固定条件,因此您必须根据需要进行修改。

+1

因此,我最终结合使用http://www.sqlalchemy.org/trac/wiki/UsageRecipes/PreFilteredQuery进行了解答。有没有一种方法可以在模型类级别为sessionmaker指定query_cls?并非所有模型都有我想过滤的这一列。 – steve 2010-12-25 18:28:42

+0

噢,我在模型中查询= Session.query_property(FilterByUserQuery)。非常感谢。 – steve 2010-12-25 20:02:56

1

这是一个非常天真实现,假设有用户已登录的属性/属性self.current_user已存储。

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) 
1

我写了一个SQLAlchemy的扩展,我觉得做你所描述的:https://github.com/mwhite/multialchemy

它通过进行代理更改Query._from_objQueryContext._froms性能,这哪里是表,从最终获取设置选择执行此操作。

相关问题