mirror of
https://github.com/ansible/awx.git
synced 2026-09-03 10:28:29 -02:30
basic fact search grammar
* Establish a base grammar for handling json path specification and value matching. With boolean logic support and parenthesis grouping i.e. (a.b.c="value") and ((a.b="foo") or (a="bar")) * generate Q() results for passing to Host.objects.filter() * Hooked up via /api/v1/hosts?host_filter=... * DynamicFilterField added to store host filter as string with grammar parser attached as static methods for later use by DynamicInventory & continued use by host_filter=...
This commit is contained in:
@@ -1,7 +1,10 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved.
|
||||
|
||||
# Python
|
||||
import json
|
||||
import re
|
||||
from pyparsing import infixNotation, opAssoc, Word, Optional, Literal
|
||||
|
||||
# Django
|
||||
from django.db.models.signals import (
|
||||
@@ -18,6 +21,7 @@ from django.db.models.fields.related import (
|
||||
ReverseManyRelatedObjectsDescriptor,
|
||||
)
|
||||
from django.utils.encoding import smart_text
|
||||
from django.db.models import Q
|
||||
|
||||
# Django-JSONField
|
||||
from jsonfield import JSONField as upstream_JSONField
|
||||
@@ -27,7 +31,7 @@ from awx.main.models.rbac import batch_role_ancestor_rebuilding, Role
|
||||
from awx.main.utils import get_current_apps
|
||||
|
||||
|
||||
__all__ = ['AutoOneToOneField', 'ImplicitRoleField', 'JSONField']
|
||||
__all__ = ['AutoOneToOneField', 'ImplicitRoleField', 'JSONField', 'DynamicFilterField']
|
||||
|
||||
|
||||
class JSONField(upstream_JSONField):
|
||||
@@ -292,3 +296,189 @@ class ImplicitRoleField(models.ForeignKey):
|
||||
child_ids = [x for x in Role_.parents.through.objects.filter(to_role_id__in=role_ids).distinct().values_list('from_role_id', flat=True)]
|
||||
Role_.objects.filter(id__in=role_ids).delete()
|
||||
Role.rebuild_role_ancestor_list([], child_ids)
|
||||
|
||||
|
||||
|
||||
unicode_printables = u''.join(unichr(c) for c in xrange(65536) if not unichr(c).isspace())
|
||||
unicode_printables += u'.'
|
||||
unicode_printables_spaces = u''.join(unichr(c) for c in xrange(65536))
|
||||
|
||||
|
||||
def string_to_type(t):
|
||||
if t == 'true':
|
||||
return True
|
||||
elif t == 'false':
|
||||
return False
|
||||
|
||||
if re.search('^[-+]?[0-9]+$',t):
|
||||
return int(t)
|
||||
|
||||
if re.search('^[-+]?[0-9]+\.[0-9]+$',t):
|
||||
return float(t)
|
||||
|
||||
return t
|
||||
|
||||
|
||||
class DynamicFilterField(models.TextField):
|
||||
|
||||
|
||||
class BoolOperand(object):
|
||||
def __init__(self, t):
|
||||
#print("Got t %s" % t)
|
||||
kwargs = dict()
|
||||
k, v = self._extract_key_value(t)
|
||||
k, v = self._json_path_to_contains(k, v)
|
||||
kwargs[k] = v
|
||||
self.result = Q(**kwargs)
|
||||
|
||||
'''
|
||||
TODO: We should be able to express this in the grammar and let
|
||||
pyparsing do the heavy lifting.
|
||||
TODO: separate django filter requests from our custom json filter
|
||||
request so we don't process the key any. This could be
|
||||
accomplished using a whitelist or introspecting the
|
||||
relationship refered to to see if it's a jsonb type.
|
||||
'''
|
||||
def _json_path_to_contains(self, k, v):
|
||||
pieces = k.split('__')
|
||||
|
||||
flag_first_arr_found = False
|
||||
|
||||
assembled_k = ''
|
||||
assembled_v = v
|
||||
|
||||
last_kv = None
|
||||
last_v = None
|
||||
|
||||
contains_count = 0
|
||||
for i, piece in enumerate(pieces):
|
||||
if flag_first_arr_found is False and piece.endswith('[]'):
|
||||
assembled_k += '%s__contains' % (piece[0:-2])
|
||||
contains_count += 1
|
||||
flag_first_arr_found = True
|
||||
elif flag_first_arr_found is False and i == len(pieces) - 1:
|
||||
assembled_k += '%s' % piece
|
||||
elif flag_first_arr_found is False:
|
||||
assembled_k += '%s__' % piece
|
||||
elif flag_first_arr_found is True:
|
||||
new_kv = dict()
|
||||
if piece.endswith('[]'):
|
||||
new_v = []
|
||||
new_kv[piece[0:-2]] = new_v
|
||||
else:
|
||||
new_v = dict()
|
||||
new_kv[piece] = new_v
|
||||
|
||||
|
||||
if last_v is None:
|
||||
last_v = []
|
||||
assembled_v = last_v
|
||||
|
||||
if type(last_v) is list:
|
||||
last_v.append(new_kv)
|
||||
elif type(last_v) is dict:
|
||||
last_kv[last_kv.keys()[0]] = new_kv
|
||||
|
||||
last_v = new_v
|
||||
last_kv = new_kv
|
||||
contains_count += 1
|
||||
|
||||
if contains_count > 1:
|
||||
if type(last_v) is list:
|
||||
last_v.append(v)
|
||||
if type(last_v) is dict:
|
||||
last_kv[last_kv.keys()[0]] = v
|
||||
|
||||
return (assembled_k, assembled_v)
|
||||
|
||||
def _extract_key_value(self, t):
|
||||
t_len = len(t)
|
||||
|
||||
k = None
|
||||
v = None
|
||||
|
||||
# key
|
||||
# "something"=
|
||||
v_offset = 2
|
||||
if t_len >= 2 and t[0] == "\"" and t[2] == "\"":
|
||||
k = t[1]
|
||||
v_offset = 4
|
||||
# something=
|
||||
else:
|
||||
k = t[0]
|
||||
|
||||
# value
|
||||
# ="something"
|
||||
if t_len > (v_offset + 2) and t[v_offset] == "\"" and t[v_offset + 2] == "\"":
|
||||
v = t[v_offset + 1]
|
||||
# empty ""
|
||||
elif t_len > (v_offset + 1):
|
||||
v = ""
|
||||
# no ""
|
||||
else:
|
||||
v = string_to_type(t[v_offset])
|
||||
|
||||
return (k, v)
|
||||
|
||||
|
||||
class BoolBinOp(object):
|
||||
def __init__(self, t):
|
||||
self.left = t[0][0].result
|
||||
self.right = t[0][2].result
|
||||
|
||||
self.result = self.execute_logic(self.left, self.right)
|
||||
|
||||
|
||||
class BoolAnd(BoolBinOp):
|
||||
def execute_logic(self, left, right):
|
||||
return left & right
|
||||
|
||||
|
||||
class BoolOr(BoolBinOp):
|
||||
def execute_logic(self, left, right):
|
||||
return left | right
|
||||
|
||||
|
||||
class BoolNot(object):
|
||||
def __init__(self,t):
|
||||
self.right = t[0][1]
|
||||
self.result = ~self.right
|
||||
|
||||
|
||||
@classmethod
|
||||
def filter_string_to_q(cls, filter_string):
|
||||
|
||||
'''
|
||||
TODO:
|
||||
* handle values with " via: a.b.c.d="hello\"world"
|
||||
* handle keys with " via: a.\"b.c="yeah"
|
||||
* handle key with __ in it
|
||||
|
||||
* add not support
|
||||
|
||||
* transform [] into contains via: a.b.c[].d[].e.f[]="blah"
|
||||
|
||||
* handle optional value quoted: a.b.c=""
|
||||
|
||||
'''
|
||||
atom = Word(unicode_printables, excludeChars=['(', ')', '=', '"'])
|
||||
atom_inside_quotes = Word(unicode_printables_spaces, excludeChars=['"'])
|
||||
atom_quoted = Literal('"') + Optional(atom_inside_quotes) + Literal('"')
|
||||
EQUAL = Literal('=')
|
||||
|
||||
grammar = ((atom_quoted | atom) + EQUAL + Optional((atom_quoted | atom)))
|
||||
grammar.setParseAction(cls.BoolOperand)
|
||||
|
||||
boolExpr = infixNotation(grammar, [
|
||||
#("not", 1, opAssoc.RIGHT, cls.BoolNot),
|
||||
("and", 2, opAssoc.LEFT, cls.BoolAnd),
|
||||
("or", 2, opAssoc.LEFT, cls.BoolOr),
|
||||
])
|
||||
|
||||
res = boolExpr.parseString('(' + filter_string + ')')
|
||||
if len(res) > 0:
|
||||
return res[0].result
|
||||
|
||||
raise RuntimeError("Parsing the filter_string %s wen't terribly wrong" % filter_string)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user