bin/addons/base/ir/ir_rule.py

   1 ##############################################################################
   2 #
   3 # Copyright (c) 2004-2008 TINY SPRL. (http://tiny.be) All Rights Reserved.
   4 #
   5 # $Id$
   6 #
   7 # WARNING: This program as such is intended to be used by professional
   8 # programmers who take the whole responsability of assessing all potential
   9 # consequences resulting from its eventual inadequacies and bugs
  10 # End users who are looking for a ready-to-use solution with commercial
  11 # garantees and support are strongly adviced to contract a Free Software
  12 # Service Company
  13 #
  14 # This program is Free Software; you can redistribute it and/or
  15 # modify it under the terms of the GNU General Public License
  16 # as published by the Free Software Foundation; either version 2
  17 # of the License, or (at your option) any later version.
  18 #
  19 # This program is distributed in the hope that it will be useful,
  20 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  21 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22 # GNU General Public License for more details.
  23 #
  24 # You should have received a copy of the GNU General Public License
  25 # along with this program; if not, write to the Free Software
  26 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  27 #
  28 ##############################################################################
  29
  30 from osv import fields,osv
  31 import time
  32 import tools
  33
  34
  35 class ir_rule_group(osv.osv):
  36         _name = 'ir.rule.group'
  37
  38         _columns = {
  39                 'name': fields.char('Name', size=128, select=1),
  40                 'model_id': fields.many2one('ir.model', 'Model',select=1, required=True),
  41                 'global': fields.boolean('Global', select=1, help="Make the rule global or it needs to be put on a group or user"),
  42                 'rules': fields.one2many('ir.rule', 'rule_group', 'Tests', help="The rule is satisfied if at least one test is True"),
  43                 'groups': fields.many2many('res.groups', 'group_rule_group_rel', 'rule_group_id', 'group_id', 'Groups'),
  44                 'users': fields.many2many('res.users', 'user_rule_group_rel', 'rule_group_id', 'user_id', 'Users'),
  45         }
  46
  47         _order = 'model_id, global DESC'
  48
  49         _defaults={
  50                 'global': lambda *a: True,
  51         }
  52
  53         def unlink(self, cr, uid, ids, context=None):
  54                 res = super(ir_rule_group, self).unlink(cr, uid, ids, context=context)
  55                 # Restart the cache on the domain_get method of ir.rule
  56                 self.pool.get('ir.rule').domain_get()
  57                 return res
  58
  59         def create(self, cr, user, vals, context=None):
  60                 res = super(ir_rule_group, self).create(cr, user, vals, context=context)
  61                 # Restart the cache on the domain_get method of ir.rule
  62                 self.pool.get('ir.rule').domain_get()
  63                 return res
  64
  65         def write(self, cr, uid, ids, vals, context=None):
  66                 if not context:
  67                         context={}
  68                 res = super(ir_rule_group, self).write(cr, uid, ids, vals, context=context)
  69                 # Restart the cache on the domain_get method of ir.rule
  70                 self.pool.get('ir.rule').domain_get()
  71                 return res
  72
  73 ir_rule_group()
  74
  75
  76 class ir_rule(osv.osv):
  77         _name = 'ir.rule'
  78         _rec_name = 'field_id'
  79
  80         def _operand(self,cr,uid,context):
  81
  82                 def get(object, level=3, recur=None, root_tech='', root=''):
  83                         res = []
  84                         if not recur:
  85                                 recur = []
  86                         fields = self.pool.get(object).fields_get(cr,uid)
  87                         key = fields.keys()
  88                         key.sort()
  89                         for k in key:
  90
  91                                 if fields[k]['type'] in ('many2one'):
  92                                         res.append((root_tech+'.'+k+'.id',
  93                                                 root+'/'+fields[k]['string']))
  94
  95                                 elif fields[k]['type'] in ('many2many', 'one2many'):
  96                                         res.append(('\',\'.join(map(lambda x: str(x.id), '+root_tech+'.'+k+'))',
  97                                                 root+'/'+fields[k]['string']))
  98
  99                                 else:
 100                                         res.append((root_tech+'.'+k,
 101                                                 root+'/'+fields[k]['string']))
 102
 103                                 if (fields[k]['type'] in recur) and (level>0):
 104                                         res.extend(get(fields[k]['relation'], level-1,
 105                                                 recur, root_tech+'.'+k, root+'/'+fields[k]['string']))
 106
 107                         return res
 108
 109                 res = [("False", "False"), ("True", "True"), ("user.id", "User")]
 110                 res += get('res.users', level=1,
 111                                 recur=['many2one'], root_tech='user', root='User')
 112                 return res
 113
 114         def _domain_force_get(self, cr, uid, ids, field_name, arg, context={}):
 115                 res = {}
 116                 for rule in self.browse(cr, uid, ids, context):
 117                         if rule.domain_force:
 118                                 res[rule.id] = eval(rule.domain_force, {'user': self.pool.get('res.users').browse(cr, 1, uid),
 119                                                         'time':time})
 120                         else:
 121                                 if rule.operator in ('in', 'child_of'):
 122                                         dom = eval("[('%s', '%s', [%s])]" % (rule.field_id.name, rule.operator,
 123                                                 rule.operand), {'user': self.pool.get('res.users').browse(cr, 1, uid),
 124                                                         'time':time})
 125                                 else:
 126                                         dom = eval("[('%s', '%s', %s)]" % (rule.field_id.name, rule.operator,
 127                                                 rule.operand), {'user': self.pool.get('res.users').browse(cr, 1, uid),
 128                                                         'time':time})
 129                                 res[rule.id] = dom
 130                 return res
 131
 132         _columns = {
 133                 'field_id': fields.many2one('ir.model.fields', 'Field',domain= "[('model_id','=', parent.model_id)]", select=1, required=True),
 134                 'operator':fields.selection((('=', '='), ('<>', '<>'), ('<=', '<='), ('>=', '>='), ('in', 'in'), ('child_of', 'child_of')), 'Operator', required=True),
 135                 'operand':fields.selection(_operand,'Operand', size=64, required=True),
 136                 'rule_group': fields.many2one('ir.rule.group', 'Group', select=2, required=True, ondelete="cascade"),
 137                 'domain_force': fields.char('Force Domain', size=250),
 138                 'domain': fields.function(_domain_force_get, method=True, string='Domain', type='char', size=250)
 139         }
 140
 141         def onchange_all(self, cr, uid, ids, field_id, operator, operand):
 142                 if not (field_id or operator or operand):
 143                         return {}
 144
 145         def domain_get(self, cr, uid, model_name):
 146                 # root user above constraint
 147                 if uid == 1:
 148                         return '', []
 149
 150                 cr.execute("""SELECT r.id FROM
 151                         ir_rule r
 152                                 JOIN (ir_rule_group g
 153                                         JOIN ir_model m ON (g.model_id = m.id))
 154                                         ON (g.id = r.rule_group)
 155                                 WHERE m.model = %s
 156                                 AND (g.id IN (SELECT rule_group_id FROM group_rule_group_rel g_rel
 157                                                         JOIN res_groups_users_rel u_rel ON (g_rel.group_id = u_rel.gid)
 158                                                         WHERE u_rel.uid = %d) OR g.global)""", (model_name, uid))
 159                 ids = map(lambda x:x[0], cr.fetchall())
 160                 if not ids:
 161                         return '', []
 162                 obj = self.pool.get(model_name)
 163                 add = []
 164                 add_str = []
 165                 sub = []
 166                 sub_str = []
 167                 clause={}
 168                 clause_global={}
 169                 for rule in self.browse(cr, uid, ids):
 170                         dom = rule.domain
 171                         if rule.rule_group['global']:
 172                                 clause_global.setdefault(rule.rule_group.id, [])
 173                                 clause_global[rule.rule_group.id].append(obj._where_calc(cr, uid, dom, active_test=False))
 174                         else:
 175                                 clause.setdefault(rule.rule_group.id, [])
 176                                 clause[rule.rule_group.id].append(obj._where_calc(cr, uid, dom, active_test=False))
 177
 178                 def _query(clause, test):
 179                         query = ''
 180                         val = []
 181                         for g in clause.values():
 182                                 if not g:
 183                                         continue
 184                                 if len(query):
 185                                         query += ' '+test+' '
 186                                 query += '('
 187                                 first = True
 188                                 for c in g:
 189                                         if not first:
 190                                                 query += ' AND '
 191                                         first = False
 192                                         query += '('
 193                                         first2 = True
 194                                         for clause in c[0]:
 195                                                 if not first2:
 196                                                         query += ' AND '
 197                                                 first2 = False
 198                                                 query += clause
 199                                         query += ')'
 200                                         val += c[1]
 201                                 query += ')'
 202                         return query, val
 203
 204                 query, val = _query(clause, 'OR')
 205                 query_global, val_global = _query(clause_global, 'OR')
 206                 if query_global:
 207                         if query:
 208                                 query = '('+query+') OR '+query_global
 209                                 val.extend(val_global)
 210                         else:
 211                                 query = query_global
 212                                 val = val_global
 213
 214
 215                 if query:
 216                         query = '('+query+')'
 217                 return query, val
 218         domain_get = tools.cache()(domain_get)
 219
 220         def unlink(self, cr, uid, ids, context=None):
 221                 res = super(ir_rule, self).unlink(cr, uid, ids, context=context)
 222                 # Restart the cache on the domain_get method of ir.rule
 223                 self.domain_get()
 224                 return res
 225
 226         def create(self, cr, user, vals, context=None):
 227                 res = super(ir_rule, self).create(cr, user, vals, context=context)
 228                 # Restart the cache on the domain_get method of ir.rule
 229                 self.domain_get()
 230                 return res
 231
 232         def write(self, cr, uid, ids, vals, context=None):
 233                 if not context:
 234                         context={}
 235                 res = super(ir_rule, self).write(cr, uid, ids, vals, context=context)
 236                 # Restart the cache on the domain_get method
 237                 self.domain_get()
 238                 return res
 239
 240 ir_rule()