在 python 上实施查询语言

implement query language on python

我正在寻找一种方法在我的工作场所向其他开发人员和可选的客户公开过滤功能。

问题

我想基于用户定义的过滤器对我的数据(python 字典)实施一种简单的查询语言,该过滤器公开给我的其他开发人员,然后再公开给我们的客户。

通过我的字典/json 数据公开 SQL 接口会很棒(我不想设置服务器)

示例场景

db = [
  {'first': 'john', 'last': 'doe', 'likes': ['cookies', 'http']},
  {'first': 'jane', 'last': 'doe', 'likes': ['cookies', 'donuts']},
  {'first': 'danny', 'last': 'foo', 'likes': ['http', 'donuts']},
]

query = '(first == "john" or last == "doe") and likes contains "cookies"'
results = run_query(db, query)

这应该 return(在结果中):

[
  {'first': 'john', 'last': 'doe', 'likes': ['cookies', 'http']},
  {'first': 'jane', 'last': 'doe', 'likes': ['cookies', 'donuts']},
]

注意:我不介意更改操作员名称,例如or -> OR contains -> inside 或其他任何东西,只要它是人类可读的并且保持相同的语言表达能力

我尝试过的解决方案

DSL

我查看了一些像 PLY 这样的 DSL 库,但在我看来它们太复杂并且涉及一些魔术来完成事情(不确定从哪里开始以及是否值得)

插件

没有找到任何插件系统来为我的用户公开沙盒功能(即更安全的评估)

JSON 查询包

我查看了 TinyDB 和其他在 json 上实现了某种 SQL 的东西,但找不到没有大量定制的东西。 我还查看了 pandasql ,它总体上看起来不错但未维护的库 :(

有一个基于 PLY 的 lucene 包解析器 - luqum 但它与我的语法树不同(它们有更多方法)并且库没有真正维护,(我确实考虑操纵这个库有点得到我想要的)

SQL网站

使用 SQLiteDB 加载我所有的数据(在内存中或不在内存中),然后 运行 SQL 对其进行查询。没有测试它,但这应该非常简单,缺点是将我的整个数据加载到 SQL 只是 运行 我不想做的数据。

我乐于接受建议,甚至是关于如何改进上述解决方案以使这项工作有效

SQL 是广为人知的,也是最终用户普遍要求的。我知道有几个选项可以在 python 中实现这一点。它们依赖于外部库,但得到了很好的支持。

小数据解决方案

将字典输入 pd.DataFrame,例如请参阅 pd.DataFrame.from_dict. Then query via pandasql 库。看来您已经尝试过了,但我提到它是因为(根据我的经验)它按照它说的做。

大数据解决方案

以 HDF5 格式保存您的数据。 pandas 数据帧和 numpy 数组可以通过 h5py library. Then use HDFql 库轻松存储为 HDF5 格式以查询 HDF5 文件。

在将 PLY 用于基于文本的查询之前,我会像这样从常规 Python 类 构建核心语言:

class Match:
    def __init__(self, **target):
        [[self.key, self.value]] = target.items()
    def __call__(self, obj):
        return self.key in obj and self.value == obj[self.key]

class Contains:        
    def __init__(self, **target):
        [[self.key, self.value]] = target.items()
    def __call__(self, obj):
        return self.key in obj and self.value in obj[self.key]        

class Or:
    def __init__(self, *predicates):
        self.predicates = predicates
    def __call__(self, record):
        return any(predicate(record) for predicate in self.predicates)

class And:
    def __init__(self, *predicates):
        self.predicates = predicates
    def __call__(self, record):
        return all(predicate(record) for predicate in self.predicates)

def run_query(db, query):
    return filter(query, db)

if __name__ == '__main__':

    db = [
      {'first': 'john', 'last': 'doe', 'likes': ['cookies', 'http']},
      {'first': 'jane', 'last': 'doe', 'likes': ['cookies', 'donuts']},
      {'first': 'danny', 'last': 'foo', 'likes': ['http', 'donuts']},
    ]
    query = And(Or(Match(first='john'), Match(last='doe')), Contains(likes='cookies'))
    for result in run_query(db, query):
        print(result)

这输出:

{'first': 'john', 'last': 'doe', 'likes': ['cookies', 'http']}
{'first': 'jane', 'last': 'doe', 'likes': ['cookies', 'donuts']}