cubicweb: comparison cubicweb/rqlrewrite.py

equal deleted inserted replaced

-:39c9e548f0ce
+:9f668acfa6c3
 tree.
 This is used for instance for read security checking in the repository.
 """
 from six import text_type, string_types
 from rql import nodes as n, stmts, TypeResolverException
 from rql.utils import common_parent
 from logilab.common import tempattr
 from logilab.common.graph import has_path
 from cubicweb import Unauthorized
-from cubicweb.schema import RRQLExpression
 def cleanup_solutions(rqlst, solutions):
 for sol in solutions:
 for vname in list(sol):
 if not (vname in rqlst.defined_vars or vname in rqlst.aliases):
 # XXX could be factorized with add_etypes_restriction from rql 0.31
 for varname in sorted(allpossibletypes):
 var = newroot.defined_vars[varname]
 stinfo = var.stinfo
 if stinfo.get('uidrel') is not None:
-continue # eid specified, no need for additional type specification
+continue  # eid specified, no need for additional type specification
 try:
 typerel = rqlst.defined_vars[varname].stinfo.get('typerel')
 except KeyError:
 assert varname in rqlst.aliases
 continue
 else:
 # variable has already some strict types restriction. new
 # possible types can only be a subset of existing ones, so only
 # remove no more possible types
 for cst in mytyperel.get_nodes(n.Constant):
-if not cst.value in possibletypes:
+if cst.value not in possibletypes:
 cst.parent.remove(cst)
 else:
 # we have to add types restriction
 if stinfo.get('scope') is not None:
 rel = var.scope.add_type_restriction(var, possibletypes)
 def _expand_selection(terms, selected, aliases, select, newselect):
 for term in terms:
 for vref in term.iget_nodes(n.VariableRef):
-if not vref.name in selected:
+if vref.name not in selected:
 select.append_selected(vref)
 colalias = newselect.get_variable(vref.name, len(aliases))
 aliases.append(n.VariableRef(colalias))
 selected.add(vref.name)
 for etype in etypes:
 for ttype in ttypes_func(etype):
 if rdef(etype, ttype).cardinality[cardindex] in '+*':
 return True
 return False
 def _compatible_relation(relations, stmt, sniprel):
 """Search among given rql relation nodes if there is one 'compatible' with the
 snippet relation, and return it if any, else None.
 class Unsupported(Exception):
 """raised when an rql expression can't be inserted in some rql query
 because it create an unresolvable query (eg no solutions found)
 """
 class VariableFromSubQuery(Exception):
 """flow control exception to indicate that a variable is coming from a
 subquery, and let parent act accordingly
 """
 newvar = newselect.get_variable(var.name)
 newvar.stinfo.setdefault('ftirels', set()).add(rel)
 newvar.stinfo.setdefault('relations', set()).add(rel)
 newselect.set_orderby(sortterms)
 _expand_selection(select.orderby, selected, aliases, select, newselect)
-select.orderby = () # XXX dereference?
+select.orderby = ()  # XXX dereference?
 if select.groupby:
 newselect.set_groupby([g.copy(newselect) for g in select.groupby])
 _expand_selection(select.groupby, selected, aliases, select, newselect)
-select.groupby = () # XXX dereference?
+select.groupby = ()  # XXX dereference?
 if select.having:
 newselect.set_having([g.copy(newselect) for g in select.having])
 _expand_selection(select.having, selected, aliases, select, newselect)
-select.having = () # XXX dereference?
+select.having = ()  # XXX dereference?
 if select.limit:
 newselect.limit = select.limit
 select.limit = None
 if select.offset:
 newselect.offset = select.offset
 assert len(newsolutions) >= len(solutions), (
 'rewritten rql %s has lost some solutions, there is probably '
 'something wrong in your schema permission (for instance using a '
 'RQLExpression which inserts a relation which doesn\'t exist in '
 'the schema)\nOrig solutions: %s\nnew solutions: %s' % (
 select, solutions, newsolutions))
 if len(newsolutions) > len(solutions):
 newsolutions = self.remove_ambiguities(snippets, newsolutions)
 assert newsolutions
 select.solutions = newsolutions
 add_types_restriction(self.schema, select)
 for varmap, rqlexprs in snippets:
 if isinstance(varmap, dict):
 varmap = tuple(sorted(varmap.items()))
 else:
 assert isinstance(varmap, tuple), varmap
-if varexistsmap is not None and not varmap in varexistsmap:
+if varexistsmap is not None and varmap not in varexistsmap:
 continue
 self.insert_varmap_snippets(varmap, rqlexprs, varexistsmap)
 def init_from_varmap(self, varmap, varexistsmap=None):
 self.varmap = varmap
 vi['rhs_rels'] = {}
 for rel in sti.get('rhsrelations', []):
 vi['rhs_rels'].setdefault(rel.r_type, []).append(rel)
 vi['lhs_rels'] = {}
 for rel in sti.get('relations', []):
-if not rel in sti.get('rhsrelations', []):
+if rel not in sti.get('rhsrelations', []):
 vi['lhs_rels'].setdefault(rel.r_type, []).append(rel)
 else:
 vi['rhs_rels'] = vi['lhs_rels'] = {}
 def _subquery_variable(self, selectvar):
 exists = varexistsmap[varmap]
 if self.exists_snippet.get(rqlexpr) is exists:
 self.insert_snippet(varmap, rqlexpr.snippet_rqlst, exists)
 if varexistsmap is None and not inserted:
 # no rql expression found matching rql solutions. User has no access right
-raise Unauthorized() # XXX may also be because of bad constraints in schema definition
+raise Unauthorized()  # XXX may also be because of bad constraints in schema definition
 def insert_snippet(self, varmap, snippetrqlst, previous=None):
 new = snippetrqlst.where.accept(self)
 existing = self.existingvars
 self.existingvars = None
 if self._insert_scope is None and any(vi.get('stinfo', {}).get('optrelations')
 for vi in self.varinfos):
 assert previous is None
 self._insert_scope, new = self.snippet_subquery(varmap, new)
 self.insert_pending()
-#self._insert_scope = None
 return new
 new = self._inserted_root(new)
 if previous is None:
 insert_scope.add_restriction(new)
 else:
 ptypes = stmt.defined_vars[varname].stinfo['possibletypes']
 if len(ptypes) > 1:
 # XXX dunno how to handle this
 self.session.error(
 'cant check security of %s, ambigous type for %s in %s',
-stmt, varname, key[0]) # key[0] == the rql expression
+stmt, varname, key[0])  # key[0] == the rql expression
 raise Unauthorized()
 etype = next(iter(ptypes))
 eschema = self.schema.eschema(etype)
 if not eschema.has_perm(self.session, action):
 rqlexprs = eschema.get_rqlexprs(action)
 continue
 done.add(rel)
 rschema = get_rschema(rel.r_type)
 if rschema.final or rschema.inlined:
 subselect_vrefs = []
-rel.children[0].name = varname # XXX explain why
+rel.children[0].name = varname  # XXX explain why
 subselect.add_restriction(rel.copy(subselect))
 for vref in rel.children[1].iget_nodes(n.VariableRef):
 if isinstance(vref.variable, n.ColumnAlias):
 # XXX could probably be handled by generating the
 # subquery into the detected subquery
 # we can use vref here define in above for loop
 ostinfo = vref.variable.stinfo
 for orel in iter_relations(ostinfo):
 orschema = get_rschema(orel.r_type)
 if orschema.final or orschema.inlined:
-todo.append( (vref.name, ostinfo) )
+todo.append((vref.name, ostinfo))
 break
 if need_null_test:
 snippetrqlst = n.Or(
 n.make_relation(subselect.get_variable(selectvar), 'is',
 (None, None), n.Constant,
 def build_variantes(self, newsolutions):
 variantes = set()
 for sol in newsolutions:
 variante = []
 for key, newvar in self.rewritten.items():
-variante.append( (key, sol[newvar]) )
+variante.append((key, sol[newvar]))
 variantes.add(tuple(variante))
 # rebuild variantes as dict
 variantes = [dict(variante) for variante in variantes]
 # remove variable which have always the same type
 for key in self.rewritten:
 removed.add(vref.name)
 for key, newvar in list(self.rewritten.items()):
 if newvar in removed:
 del self.rewritten[key]
 def _may_be_shared_with(self, sniprel, target):
 """if the snippet relation can be skipped to use a relation from the
 original query, return that relation node
 """
 if sniprel.neged(strict=True):
-return None # no way
+return None  # no way
 rschema = self.schema.rschema(sniprel.r_type)
 stmt = self.current_statement()
 for vi in self.varinfos:
 try:
 if target == 'object':
 orels = vi['lhs_rels'][sniprel.r_type]
 cardindex = 0
 ttypes_func = rschema.objects
 rdef = rschema.rdef
-else: # target == 'subject':
+else:  # target == 'subject':
 orels = vi['rhs_rels'][sniprel.r_type]
 cardindex = 1
 ttypes_func = rschema.subjects
-rdef = lambda x, y: rschema.rdef(y, x)
+def rdef(x, y):
+return rschema.rdef(y, x)
 except KeyError:
 # may be raised by vi['xhs_rels'][sniprel.r_type]
 continue
 # if cardinality isn't in '?1', we can't ignore the snippet relation
 # and use variable from the original query
 return varname in self.existingvars
 if varname == 'U':
 return True
 vargraph = self.current_expr.vargraph
 for existingvar in self.existingvars:
-#path = has_path(vargraph, varname, existingvar)
+if varname not in vargraph or has_path(vargraph, varname, existingvar):
-if not varname in vargraph or has_path(vargraph, varname, existingvar):
 return True
 # no path from this variable to an existing variable
 return False
 def visit_relation(self, node):
 if node.r_type in ('has_add_permission', 'has_update_permission',
 'has_delete_permission', 'has_read_permission'):
 assert lhs.name == 'U'
 action = node.r_type.split('_')[1]
 key = (self.current_expr, self.varmap, rhs.name)
-self.pending_keys.append( (key, action) )
+self.pending_keys.append((key, action))
 return
 if isinstance(rhs, n.VariableRef):
 if self.existingvars and not self.keep_var(rhs.name):
 return
 if lhs.name in self.revvarmap and rhs.name != 'U':
 self.pending_keys = None
 rules = self.schema.rules_rqlexpr_mapping
 for relation in union.iget_nodes(n.Relation):
 if relation.r_type in rules:
 self.select = relation.stmt
-self.solutions = solutions = self.select.solutions[:]
+self.solutions = self.select.solutions[:]
 self.current_expr = rules[relation.r_type]
 self._insert_scope = relation.scope
 self.rewritten = {}
 lhs, rhs = relation.get_variable_parts()
 varmap = {lhs.name: 'S', rhs.name: 'O'}

branch	3.25
changeset 12087	9f668acfa6c3
parent 12086	39c9e548f0ce
child 12089	54b518367617