def parser(self):
join_type = (pp.Literal("LEFT") | pp.Literal("RIGHT") | pp.Literal("INNER") | pp.Literal("OUTER"))
node_name = pp.Word(pp.alphas, pp.alphanums + "_$")
col_name = pp.Word(pp.alphas, pp.alphanums + "_$")
col_name_list = pp.Group(pp.delimitedList(col_name, delim=","))
l_brac = pp.Suppress("[")
r_brac = pp.Suppress("]")
single_join = (join_type + pp.Suppress("(") + node_name + l_brac +
col_name_list + r_brac + pp.Suppress("==>") + node_name +
l_brac + col_name_list + r_brac + pp.Suppress(")"))
single_join.addParseAction(lambda x: self._add_join(join_type=x[0],
child_node_name=x[1],
child_cols=x[2],
parent_node_name=x[3],
parent_cols=x[4]))
join_block = pp.OneOrMore(single_join)
return join_block
python类Group()的实例源码
def __init__(self, identifier_parser=None):
"""
:param IdentifierParser identifier_parser: An identifier parser for checking the 3P and 5P partners
"""
self.identifier_parser = identifier_parser if identifier_parser is not None else IdentifierParser()
identifier = self.identifier_parser.language
reference_seq = oneOf(['r', 'p', 'c'])
coordinate = pyparsing_common.integer | '?'
missing = Keyword('?')
range_coordinate = missing(FUSION_MISSING) | (
reference_seq(FUSION_REFERENCE) + Suppress('.') + coordinate(FUSION_START) + Suppress('_') + coordinate(
FUSION_STOP))
self.language = fusion_tags + nest(Group(identifier)(PARTNER_5P), Group(range_coordinate)(RANGE_5P),
Group(identifier)(PARTNER_3P), Group(range_coordinate)(RANGE_3P))
super(FusionParser, self).__init__(self.language)
def _parse_atat_lattice(lattice_in):
"""Parse an ATAT-style `lat.in` string.
The parsed string will be in three groups: (Coordinate system) (lattice) (atoms)
where the atom group is split up into subgroups, each describing the position and atom name
"""
float_number = Regex(r'[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?').setParseAction(lambda t: [float(t[0])])
vector = Group(float_number + float_number + float_number)
angles = vector
vector_line = vector + Suppress(LineEnd())
coord_sys = Group((vector_line + vector_line + vector_line) | (vector + angles + Suppress(LineEnd())))
lattice = Group(vector + vector + vector)
atom = Group(vector + Group(OneOrMore(Word(alphas + '_'))))
atat_lattice_grammer = coord_sys + lattice + Group(OneOrMore(atom))
# parse the input string and convert it to a POSCAR string
return atat_lattice_grammer.parseString(lattice_in)
def expression(self):
expression = pyparsing.Forward()
# (1 + (2 + 3))
nested_expression = pyparsing.nestedExpr(
"(", ")", expression).setParseAction(self._combine_lists)
# FOO(2 , 3)
function_call = (
_TOKEN().setResultsName("function")
+ _OPEN_PARENTHESIS()
+ pyparsing.delimitedList(
pyparsing.Combine(expression, adjacent=False, joinString=" "),
delim=",").setResultsName("func_args")
+ _CLOSE_PARENTHESIS()
)
expression << pyparsing.OneOrMore(
function_call.setParseAction(self._is_known_function)
| pyparsing.Group(nested_expression)
| _TOKEN()
| _NOT_TOKEN()
)
return pyparsing.Combine(expression, adjacent=False, joinString=" ")
def _maybe_attributes(self):
"""Possibly match some attributes.
The syntax of attributes is described here:
https://gcc.gnu.org/onlinedocs/gcc/Attribute-Syntax.html
"""
return pyparsing.Group(
pyparsing.ZeroOrMore(
_ATTRIBUTE
+ _DOUBLE_OPEN_PARENTHESIS
+ pyparsing.delimitedList(
pyparsing.Group(
self._identifier()("name")
+ pyparsing.Optional(
_OPEN_PARENTHESIS
+ parsers.anything_beetween("()")("args")
+ _CLOSE_PARENTHESIS
)
)
)
+ _DOUBLE_CLOSE_PARENTHESIS
).setParseAction(self._make_attribute)
)("attributes")
def _create_simple_statements():
global binary, ident, rvalue, simple_statement, semi, comp, number, slot_id, callrpc_stmt, generic_statement, streamer_stmt, stream, selector
if simple_statement is not None:
return
meta_stmt = Group(Literal('meta').suppress() + ident + Literal('=').suppress() + rvalue + semi).setResultsName('meta_statement')
require_stmt = Group(Literal('require').suppress() + ident + comp + rvalue + semi).setResultsName('require_statement')
set_stmt = Group(Literal('set').suppress() - (ident | number) - Literal("to").suppress() - (rvalue | binary) - Optional(Literal('as').suppress() + config_type) + semi).setResultsName('set_statement')
callrpc_stmt = Group(Literal("call").suppress() + (ident | number) + Literal("on").suppress() + slot_id + Optional(Literal("=>").suppress() + stream('explicit_stream')) + semi).setResultsName('call_statement')
streamer_stmt = Group(Optional(Literal("manual")('manual')) + Optional(oneOf(u'encrypted signed')('security')) + Optional(Literal(u'realtime')('realtime')) + Literal('streamer').suppress() -
Literal('on').suppress() - selector('selector') - Optional(Literal('to').suppress() - slot_id('explicit_tile')) - Optional(Literal('with').suppress() - Literal('streamer').suppress() - number('with_other')) - semi).setResultsName('streamer_statement')
copy_stmt = Group(Literal("copy").suppress() - Optional(oneOf("all count average")('modifier')) - Optional(stream('explicit_input') | number('constant_input')) - Literal("=>") - stream("output") - semi).setResultsName('copy_statement')
trigger_stmt = Group(Literal("trigger") - Literal("streamer") - number('index') - semi).setResultsName('trigger_statement')
simple_statement = meta_stmt | require_stmt | set_stmt | callrpc_stmt | streamer_stmt | trigger_stmt | copy_stmt
# In generic statements, keep track of the location where the match started for error handling
locator = Empty().setParseAction(lambda s, l, t: l)('location')
generic_statement = Group(locator + Group(ZeroOrMore(Regex(u"[^{};]+")) + Literal(u';'))('match')).setResultsName('unparsed_statement')
def _create_block_bnf():
global block_bnf, time_interval, slot_id, statement, block_id, ident, stream
if block_bnf is not None:
return
trigger_clause = Group(stream_trigger | Group(stream).setResultsName('stream_always') | Group(ident).setResultsName('identifier'))
every_block_id = Group(Literal(u'every').suppress() - time_interval).setResultsName('every_block')
when_block_id = Group(Literal(u'when').suppress() + Literal("connected").suppress() - Literal("to").suppress() - slot_id).setResultsName('when_block')
latch_block_id = Group(Literal(u'when').suppress() - stream_trigger).setResultsName('latch_block')
config_block_id = Group(Literal(u'config').suppress() - slot_id).setResultsName('config_block')
on_block_id = Group(Literal(u'on').suppress() - trigger_clause.setResultsName('triggerA') - Optional((Literal("and") | Literal("or")) - trigger_clause.setResultsName('triggerB'))).setResultsName('on_block')
block_id = every_block_id | when_block_id | latch_block_id | config_block_id | on_block_id
block_bnf = Forward()
statement = generic_statement | block_bnf
block_bnf << Group(block_id + Group(Literal(u'{').suppress() + ZeroOrMore(statement) + Literal(u'}').suppress())).setResultsName('block')
def __init__(self, identifier_parser=None):
"""
:param IdentifierParser identifier_parser: An identifier parser for checking the 3P and 5P partners
"""
self.identifier_parser = identifier_parser if identifier_parser is not None else IdentifierParser()
gmod_default_ns = oneOf(list(language.gmod_namespace.keys())).setParseAction(self.handle_gmod_default)
gmod_identifier = Group(self.identifier_parser.identifier_qualified) | Group(gmod_default_ns)
self.language = gmod_tag + nest(gmod_identifier(IDENTIFIER))
super(GmodParser, self).__init__(self.language)
def build_legacy_fusion(identifier, reference):
break_start = (ppc.integer | '?').setParseAction(fusion_break_handler_wrapper(reference, start=True))
break_end = (ppc.integer | '?').setParseAction(fusion_break_handler_wrapper(reference, start=False))
res = identifier(PARTNER_5P) + WCW + fusion_tags + nest(identifier(PARTNER_3P) + Optional(
WCW + Group(break_start)(RANGE_5P) + WCW + Group(break_end)(RANGE_3P)))
res.setParseAction(fusion_legacy_handler)
return res
def __init__(self, identifier_parser=None):
"""
:param IdentifierParser identifier_parser: An identifier parser for checking the 3P and 5P partners
"""
identifier_parser = identifier_parser if identifier_parser is not None else IdentifierParser()
super(LocationParser, self).__init__(Group(location_tag + nest(identifier_parser.language))(LOCATION))
def parse_filter_str(self, filter_str):
"""
method to parse filter string
"""
prop = pp.WordStart(pp.alphas) + pp.Word(pp.alphanums +
"_").setResultsName("prop")
value = (pp.QuotedString("'") | pp.QuotedString('"') | pp.Word(
pp.printables, excludeChars=",")).setResultsName("value")
types_ = pp.oneOf("re eq ne gt ge lt le").setResultsName("types")
flags = pp.oneOf("C I").setResultsName("flags")
comma = pp.Literal(',')
quote = (pp.Literal("'") | pp.Literal('"')).setResultsName("quote")
type_exp = pp.Group(pp.Literal("type") + pp.Literal(
"=") + quote + types_ + quote).setResultsName("type_exp")
flag_exp = pp.Group(pp.Literal("flag") + pp.Literal(
"=") + quote + flags + quote).setResultsName("flag_exp")
semi_expression = pp.Forward()
semi_expression << pp.Group(pp.Literal("(") +
prop + comma + value +
pp.Optional(comma + type_exp) +
pp.Optional(comma + flag_exp) +
pp.Literal(")")
).setParseAction(
self.parse_filter_obj).setResultsName("semi_expression")
expr = pp.Forward()
expr << pp.operatorPrecedence(semi_expression, [
("not", 1, pp.opAssoc.RIGHT, self.not_operator),
("and", 2, pp.opAssoc.LEFT, self.and_operator),
("or", 2, pp.opAssoc.LEFT, self.or_operator)
])
result = expr.parseString(filter_str)
return result
def __init__(self, ffilter, queue_out):
FuzzQueue.__init__(self, queue_out)
Thread.__init__(self)
self.setName('filter_thread')
self.queue_out = queue_out
if PYPARSING:
element = oneOf("c l w h")
digits = "XB0123456789"
integer = Word( digits )#.setParseAction( self.__convertIntegers )
elementRef = Group(element + oneOf("= != < > >= <=") + integer)
operator = oneOf("and or")
definition = elementRef + ZeroOrMore( operator + elementRef)
nestedformula = Group(Suppress(Optional(Literal("("))) + definition + Suppress(Optional(Literal(")"))))
self.finalformula = nestedformula + ZeroOrMore( operator + nestedformula)
elementRef.setParseAction(self.__compute_element)
nestedformula.setParseAction(self.__compute_formula)
self.finalformula.setParseAction(self.__myreduce)
self.res = None
self.hideparams = ffilter
if "XXX" in self.hideparams['codes']:
self.hideparams['codes'].append("0")
self.baseline = None
def __init__(self):
if PYPARSING:
category = Word( alphas + "_-*", alphanums + "_-*" )
operator = oneOf("and or ,")
neg_operator = "not"
elementRef = category
definition = elementRef + ZeroOrMore( operator + elementRef)
nestedformula = Group(Suppress(Optional(Literal("("))) + definition + Suppress(Optional(Literal(")"))))
neg_nestedformula = Optional(neg_operator) + nestedformula
self.finalformula = neg_nestedformula + ZeroOrMore( operator + neg_nestedformula)
elementRef.setParseAction(self.__compute_element)
neg_nestedformula.setParseAction(self.__compute_neg_formula)
nestedformula.setParseAction(self.__compute_formula)
self.finalformula.setParseAction(self.__myreduce)
def getchunk():
"""
Using pyparsing, create chunk reader for chunk strings.
"""
slot = pp.Word("".join([pp.alphas, "_"]), "".join([pp.alphanums, "_"]))
special_value = pp.Group(pp.oneOf([ACTRVARIABLE, "".join([ACTRNEG, ACTRVARIABLE]), ACTRNEG, VISIONGREATER, VISIONSMALLER, "".join([VISIONGREATER, ACTRVARIABLE]), "".join([VISIONSMALLER, ACTRVARIABLE])])\
+ pp.Word("".join([pp.alphanums, "_", '"', "'"])))
strvalue = pp.QuotedString('"', unquoteResults=False)
strvalue2 = pp.QuotedString("'", unquoteResults=False)
varvalue = pp.Word("".join([pp.alphanums, "_"]))
value = varvalue | special_value | strvalue | strvalue2
chunk_reader = pp.OneOrMore(pp.Group(slot + value))
return chunk_reader
def getrule():
"""
Using pyparsing, get rule out of a string.
"""
arrow = pp.Literal("==>")
buff = pp.Word(pp.alphas, "".join([pp.alphanums, "_"]))
special_valueLHS = pp.oneOf([x for x in _LHSCONVENTIONS.keys()])
end_buffer = pp.Literal(">")
special_valueRHS = pp.oneOf([x for x in _RHSCONVENTIONS.keys()])
chunk = getchunk()
rule_reader = pp.Group(pp.OneOrMore(pp.Group(special_valueLHS + buff + end_buffer + pp.Group(pp.Optional(chunk))))) + arrow + pp.Group(pp.OneOrMore(pp.Group(special_valueRHS + buff + end_buffer + pp.Group(pp.Optional(chunk)))))
return rule_reader
def group(cls, expr):
def group_action(s, l, t):
try:
lst = t[0].asList()
except (IndexError, AttributeError):
lst = t
return [cls(lst)]
return Group(expr).setParseAction(group_action)
def __init__(self, ffilter, queue_out):
FuzzQueue.__init__(self, queue_out)
Thread.__init__(self)
self.setName('filter_thread')
self.queue_out = queue_out
if PYPARSING:
element = oneOf("c l w h")
digits = "XB0123456789"
integer = Word( digits )#.setParseAction( self.__convertIntegers )
elementRef = Group(element + oneOf("= != < > >= <=") + integer)
operator = oneOf("and or")
definition = elementRef + ZeroOrMore( operator + elementRef)
nestedformula = Group(Suppress(Optional(Literal("("))) + definition + Suppress(Optional(Literal(")"))))
self.finalformula = nestedformula + ZeroOrMore( operator + nestedformula)
elementRef.setParseAction(self.__compute_element)
nestedformula.setParseAction(self.__compute_formula)
self.finalformula.setParseAction(self.__myreduce)
self.res = None
self.hideparams = ffilter
if "XXX" in self.hideparams['codes']:
self.hideparams['codes'].append("0")
self.baseline = None
def __init__(self):
if PYPARSING:
category = Word( alphas + "_-*", alphanums + "_-*" )
operator = oneOf("and or ,")
neg_operator = "not"
elementRef = category
definition = elementRef + ZeroOrMore( operator + elementRef)
nestedformula = Group(Suppress(Optional(Literal("("))) + definition + Suppress(Optional(Literal(")"))))
neg_nestedformula = Optional(neg_operator) + nestedformula
self.finalformula = neg_nestedformula + ZeroOrMore( operator + neg_nestedformula)
elementRef.setParseAction(self.__compute_element)
neg_nestedformula.setParseAction(self.__compute_neg_formula)
nestedformula.setParseAction(self.__compute_formula)
self.finalformula.setParseAction(self.__myreduce)
interpreter.py 文件源码
项目：Python_Master-the-Art-of-Design-Patterns
作者: PacktPublishing
项目源码
文件源码
阅读 27
收藏 0
点赞 0
评论 0
def main():
word = Word(alphanums)
command = Group(OneOrMore(word))
token = Suppress("->")
device = Group(OneOrMore(word))
argument = Group(OneOrMore(word))
event = command + token + device + Optional(token + argument)
gate = Gate()
garage = Garage()
airco = Aircondition()
heating = Heating()
boiler = Boiler()
fridge = Fridge()
tests = ('open -> gate',
'close -> garage',
'turn on -> aircondition',
'turn off -> heating',
'increase -> boiler temperature -> 5 degrees',
'decrease -> fridge temperature -> 2 degrees')
open_actions = {'gate':gate.open, 'garage':garage.open, 'aircondition':airco.turn_on,
'heating':heating.turn_on, 'boiler temperature':boiler.increase_temperature,
'fridge temperature':fridge.increase_temperature}
close_actions = {'gate':gate.close, 'garage':garage.close, 'aircondition':airco.turn_off,
'heating':heating.turn_off, 'boiler temperature':boiler.decrease_temperature,
'fridge temperature':fridge.decrease_temperature}
for t in tests:
if len(event.parseString(t)) == 2: # no argument
cmd, dev = event.parseString(t)
cmd_str, dev_str = ' '.join(cmd), ' '.join(dev)
if 'open' in cmd_str or 'turn on' in cmd_str:
open_actions[dev_str]()
elif 'close' in cmd_str or 'turn off' in cmd_str:
close_actions[dev_str]()
elif len(event.parseString(t)) == 3: # argument
cmd, dev, arg = event.parseString(t)
cmd_str, dev_str, arg_str = ' '.join(cmd), ' '.join(dev), ' '.join(arg)
num_arg = 0
try:
num_arg = int(arg_str.split()[0]) # extract the numeric part
except ValueError as err:
print("expected number but got: '{}'".format(arg_str[0]))
if 'increase' in cmd_str and num_arg > 0:
open_actions[dev_str](num_arg)
elif 'decrease' in cmd_str and num_arg > 0:
close_actions[dev_str](num_arg)
def build_parser(self):
parsed_term = pyparsing.Group(pyparsing.Combine(pyparsing.Word(pyparsing.alphanums) + \
pyparsing.Suppress('*'))).setResultsName('wildcard') | \
pyparsing.Group(pyparsing.Combine(pyparsing.Word(pyparsing.alphanums+"._") + \
pyparsing.Word(':') + pyparsing.Group(pyparsing.Optional("\"") + \
pyparsing.Optional("<") + pyparsing.Optional(">") + pyparsing.Optional("=") + \
pyparsing.Optional("-") + pyparsing.Word(pyparsing.alphanums+"._/") + \
pyparsing.Optional("&") + pyparsing.Optional("<") + pyparsing.Optional(">") + \
pyparsing.Optional("=") + pyparsing.Optional("-") + \
pyparsing.Optional(pyparsing.Word(pyparsing.alphanums+"._/")) + \
pyparsing.Optional("\"")))).setResultsName('fields') | \
pyparsing.Group(pyparsing.Combine(pyparsing.Suppress('-')+ \
pyparsing.Word(pyparsing.alphanums+"."))).setResultsName('not_term') | \
pyparsing.Group(pyparsing.Word(pyparsing.alphanums)).setResultsName('term')
parsed_or = pyparsing.Forward()
parsed_quote_block = pyparsing.Forward()
parsed_quote_block << ((parsed_term + parsed_quote_block) | parsed_term )
parsed_quote = pyparsing.Group(pyparsing.Suppress('"') + parsed_quote_block + \
pyparsing.Suppress('"')).setResultsName("quotes") | parsed_term
parsed_parenthesis = pyparsing.Group((pyparsing.Suppress("(") + parsed_or + \
pyparsing.Suppress(")"))).setResultsName("parenthesis") | parsed_quote
parsed_and = pyparsing.Forward()
parsed_and << (pyparsing.Group(parsed_parenthesis + pyparsing.Suppress(pyparsing.Keyword("and")) + \
parsed_and).setResultsName("and") | \
pyparsing.Group(parsed_parenthesis + pyparsing.OneOrMore(~pyparsing.oneOf("or and") + \
parsed_and)).setResultsName("and") | parsed_parenthesis)
parsed_or << (pyparsing.Group(parsed_and + pyparsing.Suppress(pyparsing.Keyword("or")) + \
parsed_or).setResultsName("or") | parsed_and)
return parsed_or.parseString
def parse_filter_str(self, filter_str):
"""
method to parse filter string
"""
prop = pp.WordStart(pp.alphas) + pp.Word(pp.alphanums +
"_").setResultsName("prop")
value = (pp.QuotedString("'") | pp.QuotedString('"') | pp.Word(
pp.printables, excludeChars=",")).setResultsName("value")
types_ = pp.oneOf("re eq ne gt ge lt le").setResultsName("types")
flags = pp.oneOf("C I").setResultsName("flags")
comma = pp.Literal(',')
quote = (pp.Literal("'") | pp.Literal('"')).setResultsName("quote")
type_exp = pp.Group(pp.Literal("type") + pp.Literal(
"=") + quote + types_ + quote).setResultsName("type_exp")
flag_exp = pp.Group(pp.Literal("flag") + pp.Literal(
"=") + quote + flags + quote).setResultsName("flag_exp")
semi_expression = pp.Forward()
semi_expression << pp.Group(pp.Literal("(") +
prop + comma + value +
pp.Optional(comma + type_exp) +
pp.Optional(comma + flag_exp) +
pp.Literal(")")
).setParseAction(
self.parse_filter_obj).setResultsName("semi_expression")
expr = pp.Forward()
expr << pp.operatorPrecedence(semi_expression, [
("not", 1, pp.opAssoc.RIGHT, self.not_operator),
("and", 2, pp.opAssoc.LEFT, self.and_operator),
("or", 2, pp.opAssoc.LEFT, self.or_operator)
])
result = expr.parseString(filter_str)
return result
def entry():
name = Word(alphas, alphas+'_')
value = Word(nums, nums+".").setResultsName('nominal_value')
uncert = Word(nums).setResultsName('std_dev')
value_uncert = value + Suppress("(") + uncert + Suppress(")")
return Dict( Group(name + Suppress("=") + value_uncert ) )
def parser(cls):
lpar = pp.Suppress("(")
rpar = pp.Suppress(")")
mutate = pp.Suppress('mutate')
col_name = pp.Word(pp.alphas, pp.alphanums + "_$")
expr_evaluator = Evaluator(deferred_eval=True)
col_expr = expr_evaluator.parser()
mutation = col_name + pp.Suppress("=") + col_expr
mutation.setParseAction(lambda x: {'col_name': x[0], 'col_expr': x[1]})
mutations = pp.Group(pp.delimitedList(mutation))
parser = mutate + lpar + mutations + rpar
parser.setParseAction(lambda x: Mutate(mutations=x))
return parser
def parser(cls):
rename = pp.Suppress("rename")
rename_kwarg = common_parsers.column + pp.Suppress("=") + common_parsers.column
rename_kwarg.setParseAction(lambda x: {x[0]: x[1]})
kwargs = pp.Group(pp.delimitedList(rename_kwarg))
kwargs.setParseAction(lambda x: {k: v for d in x for k, v in d.items()})
parser = rename + pp.Suppress("(") + kwargs + pp.Suppress(")")
parser.setParseAction(lambda x: Rename(columns=x[0]))
return parser
def parser(cls):
select = pp.Suppress("select")
column = common_parsers.column
parser = select + pp.Suppress("(") + pp.Group(pp.delimitedList(column)) + pp.Suppress(")")
parser.setParseAction(lambda x: Select(columns=x[0]))
return parser
def parser(cls):
remove = pp.Suppress("remove")
column = common_parsers.column
parser = remove + pp.Suppress("(") + pp.Group(pp.delimitedList(column)) + pp.Suppress(")")
parser.setParseAction(lambda x: Remove(columns=x[0]))
return parser
def parser(cls):
unpack = pp.Suppress("unpack")
packed_col_name = common_parsers.column
dict_key = pp.Suppress("[") + pp.QuotedString(quoteChar="'") + pp.Suppress("]")
dict_key_grp = pp.Group(pp.OneOrMore(dict_key))
new_col_name = common_parsers.column
unpack_arg = new_col_name + pp.Suppress("=") + packed_col_name + dict_key_grp
unpack_arg.setParseAction(lambda x: {'packed_col': x[1], 'key_list': x[2], 'new_col_name': x[0]})
parser = unpack + pp.Suppress("(") + pp.delimitedList(unpack_arg) + pp.Suppress(")")
parser.setParseAction(lambda x: Unpack(unpack_list=x))
return parser
def parser(self):
query_key = pp.Keyword("QUERY")
query_value = pp.Suppress("|") + pp.SkipTo(pp.Suppress(";"), include=True)
fields_key = pp.Keyword("FIELDS")
field_name = common_parsers.column
field_name_list = pp.Group(pp.delimitedList(field_name, delim=",")).setParseAction(lambda x: x.asList())
fields_block = (pp.Suppress(fields_key) + field_name_list)
connector_name = pp.Word(pp.alphas, pp.alphanums + "_$")
using_block = pp.Suppress("USING") + connector_name
then_key = pp.Suppress("THEN")
manipulation_set = pp.Suppress("|") + pp.SkipTo(pp.Suppress(";"), include=True)
then_block = then_key + manipulation_set
as_key = pp.Suppress("AS")
node_name = pp.Word(pp.alphas, pp.alphanums + "_$")
as_block = as_key + node_name
query_node_block = (pp.Suppress(query_key) + query_value + pp.Optional(fields_block, default=None) + using_block + pp.Optional(then_block, default=None) + as_block)
query_node_block.setParseAction(lambda x: self._add_query_node(query_value=x[0],
connector_name=x[2],
node_name=x[4],
fields=x[1],
manipulation_set=x[3]))
single_query_node = query_node_block + pp.Optional(pp.Suppress("---"))
retrieve_block = pp.OneOrMore(single_query_node)
return retrieve_block
def parser(self):
# Define punctuation as suppressed literals.
lparen, rparen, lbrack, rbrack, lbrace, rbrace, colon = \
map(pp.Suppress, "()[]{}:")
integer = pp.Combine(pp.Optional(pp.oneOf("+ -")) + pp.Word(pp.nums)) \
.setName("integer") \
.setParseAction(lambda toks: int(toks[0]))
real = pp.Combine(pp.Optional(pp.oneOf("+ -")) + pp.Word(pp.nums) + "." +
pp.Optional(pp.Word(pp.nums)) +
pp.Optional(pp.oneOf("e E") + pp.Optional(pp.oneOf("+ -")) + pp.Word(pp.nums))) \
.setName("real") \
.setParseAction(lambda toks: float(toks[0]))
_datetime_arg = (integer | real)
datetime_args = pp.Group(pp.delimitedList(_datetime_arg))
_datetime = pp.Suppress(pp.Literal('datetime') + pp.Literal("(")) + datetime_args + pp.Suppress(")")
_datetime.setParseAction(lambda x: self._make_datetime(x[0]))
tuple_str = pp.Forward()
list_str = pp.Forward()
dict_str = pp.Forward()
list_item = real | integer | _datetime | pp.quotedString.setParseAction(pp.removeQuotes) | \
pp.Group(list_str) | tuple_str | dict_str
tuple_str << (pp.Suppress("(") + pp.Optional(pp.delimitedList(list_item)) +
pp.Optional(pp.Suppress(",")) + pp.Suppress(")"))
tuple_str.setParseAction(lambda toks : tuple(toks.asList()))
list_str << (lbrack + pp.Optional(pp.delimitedList(list_item) +
pp.Optional(pp.Suppress(","))) + rbrack)
dict_entry = pp.Group(list_item + colon + list_item)
dict_str << (lbrace + pp.Optional(pp.delimitedList(dict_entry) +
pp.Optional(pp.Suppress(","))) + rbrace)
dict_str.setParseAction(lambda toks: dict(toks.asList()))
return list_item
def __setattr__(self, k, v):
if isinstance(v, pp.ParserElement):
Grammar.grm[k] = pp.Group(v).setResultsName(k)
