pyparsing 语法提取 Python 片段的部分
pyparsing grammer to extract portions of Python snippet
我有一个 Python 片段,如下所示:
fn = HiveExecOperator(
task_id="abc",
hql="abc.hql",
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
dag=main_dag,
)
我想定义一个语法,让我可以在函数 HiveExecOperator
的参数列表中查看键值对,而无需分解嵌套的参数。例如 - 我有兴趣取回列表:
[task_id="abc",
hql="abc.hql",
...
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
...]
我尝试执行以下操作:
assignment = variable + '=' + "HiveExecOperator" + nestedExpr('(', ')').setParseAction(lambda x: print(x))
parameters.transformString(python_snippet)
setParseAction
的输出是:
['fn', '=', 'HiveExecOperator(']
['task_id', '=', '"abc",']
['hql', '=', '"abc.hql",']
['dq_sql', '=', 'DQCheck("stage.billing_associations")']
['lower', '=', '1.0,']
['upper', '=', 'float("inf"),']
...
我们将不胜感激。
如 mkrieger1 所述,您可以使用 ast
内置的 Python 库。
在Python 3.9(或更高版本)中,有ast.unparse
函数可以将ast.Node
转换为字符串。
import ast
mycode = """\
fn = HiveExecOperator(
task_id="abc",
hql="abc.hql",
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
dag=main_dag,
)
"""
root = ast.parse(mycode)
calls = [n for n in ast.walk(root) if isinstance(n, ast.Call)]
first_call = calls[0]
target_list = [(k.arg, ast.unparse(k.value)) for k in first_call.keywords]
print(target_list)
这给出了
[
('task_id', "'abc'"),
('hql', "'abc.hql'"),
('dq_sql', "DQCheck('pqr').is_within_range('COUNT(DISTINCT billing_association_type)', 'type_cnts', lower=1.0, upper=float('inf')).build()"),
('dag', 'main_dag')
]
我有一个 Python 片段,如下所示:
fn = HiveExecOperator(
task_id="abc",
hql="abc.hql",
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
dag=main_dag,
)
我想定义一个语法,让我可以在函数 HiveExecOperator
的参数列表中查看键值对,而无需分解嵌套的参数。例如 - 我有兴趣取回列表:
[task_id="abc",
hql="abc.hql",
...
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
...]
我尝试执行以下操作:
assignment = variable + '=' + "HiveExecOperator" + nestedExpr('(', ')').setParseAction(lambda x: print(x))
parameters.transformString(python_snippet)
setParseAction
的输出是:
['fn', '=', 'HiveExecOperator(']
['task_id', '=', '"abc",']
['hql', '=', '"abc.hql",']
['dq_sql', '=', 'DQCheck("stage.billing_associations")']
['lower', '=', '1.0,']
['upper', '=', 'float("inf"),']
...
我们将不胜感激。
如 mkrieger1 所述,您可以使用 ast
内置的 Python 库。
在Python 3.9(或更高版本)中,有ast.unparse
函数可以将ast.Node
转换为字符串。
import ast
mycode = """\
fn = HiveExecOperator(
task_id="abc",
hql="abc.hql",
dq_sql=DQCheck("pqr")
.is_within_range(
"COUNT(DISTINCT billing_association_type)",
"type_cnts",
lower=1.0,
upper=float("inf"),
)
.build(),
dag=main_dag,
)
"""
root = ast.parse(mycode)
calls = [n for n in ast.walk(root) if isinstance(n, ast.Call)]
first_call = calls[0]
target_list = [(k.arg, ast.unparse(k.value)) for k in first_call.keywords]
print(target_list)
这给出了
[
('task_id', "'abc'"),
('hql', "'abc.hql'"),
('dq_sql', "DQCheck('pqr').is_within_range('COUNT(DISTINCT billing_association_type)', 'type_cnts', lower=1.0, upper=float('inf')).build()"),
('dag', 'main_dag')
]