1. 程式人生 > 其它 >Python學習:json物件快速訪問(續)

Python學習:json物件快速訪問(續)

技術標籤:pythonpython

問題

我們再次回到jsonpath的問題,起初使用jsonpath-ng實現快速訪問json物件節點。但jsonpath-ng的效能問題,導致這個方法實在是糟糕,無法使用,悲劇的是已經寫了很多程式碼,再用全路徑訪問json工作量有點太大,有點進退兩難。

回過頭思考,起初我的目的,就是不必要寫全路徑訪問json節點,通過jsonpath只需要寫一個節點進行模糊匹配,實現節點訪問。實際上該需求並不複雜,並沒有用到jsonpath的複雜功能。能不能不用jsonpath,也能實現?

幸運的是方案是存在的,而且實現起來也很容易。python語言天生就支援json物件,不需要再進行物件的序列化,可以直接利用這個語言特性。

方案

節點物件獲取

首先我們解決json節點獲取問題。我們寫下如下的程式碼:

def get_one_json_node(data: dict, field: str):
    if isinstance(data, list):
        for it in data:
            ret = get_one_json_node(it, field)
            if ret is not None:
                return ret

    if not isinstance(data, dict):
        return None
    if field in data.keys():
        return data[field]

    for it in data.keys():
        ret = get_one_json_node(data[it], field)
        if ret:
            return ret
    return None

測試程式碼:

class TestJsonGet:
    js = {'a': 10, 'b': 20, 'c': {'e': 10, 'f': 'string'}, 'c1': {'e': {'f1': 30}, 'f': 'string'},
  'c2': {'e': 10, 'f': 'string'}, 'z': [{'z1': 10}, {'z1': 20}]}

    def test_get1(self):
        assert 10 == get_one_json_node(self.js, 'a')
        assert 10 == get_one_json_node(self.js, 'e')

    def test_get2(self):
        assert 10  == get_one_json_node(self.js, 'c.e')

通過pytest測試發現,test_get2測試失敗。原因是有多個e節點,我們需要支援c.e的訪問,於是再封裝一個上層方法:

def get_json_node(data: dict, field: str):
    node_path = field.split(".")
    node = data
    find = False
    for it in node_path:
        node = get_one_json_node(node, it)
        if not node:
            return None
        else:
            find = True

    if find:
        return node
    else:
        return None

測試程式碼:

class TestJsonGet:
    js = {'a': 10, 'b': 20, 'c': {'e': 10, 'f': 'string'}, 'c1': {'e': {'f1': 30}, 'f': 'string'},
          'c2': {'e': 10, 'f': 'string'}, 'z': [{'z1': 10}, {'z1': 20}]}

    def test_get3(self):
        assert 10 == get_json_node(self.js, 'a')
        assert 10 == get_json_node(self.js, 'e')
        assert 10 == get_json_node(self.js, 'c.e')

    def test_get4(self):
        assert 10 == get_json_node(self.js, 'z.z1')

這次pytest測試就全部通過了(實際上,我也測試了陣列節點的訪問。為了簡單起見,匹配第一個節點時get就直接返回,無需返回所有匹配節點)。節點的get方法執行時間0.1ms左右,相比jsonpath-ng提升近百倍了,可以接受。測試結果如下:

 pytest -s test/test_jsonhelper.py::TestJsonGet
======================================================= test session starts =======================================================
platform darwin -- Python 3.8.2, pytest-6.1.2, py-1.9.0, pluggy-0.13.1
rootdir: /Users/xxx/data/code/python
collected 4 items                                                                                                                 

test/test_jsonhelper.py ..func:{get_json_node} exec time is:{0.00317} ms
func:{get_json_node} exec time is:{0.08789} ms
func:{get_json_node} exec time is:{0.00195} ms
.func:{get_json_node} exec time is:{0.00317} ms

節點的寫入

同理,對於json節點設定可以參考get方法,寫兩個函式,具體程式碼如下:

def set_one_json_node(data: dict, field: str, value):
    if isinstance(data, list):
        for it in data:
            ret = set_one_json_node(it, field, value)
            if ret is not None:
                return ret

    if not isinstance(data, dict):
        return None
    if field in data.keys():
        data[field] = value
        return data

    for it in data.keys():
        ret = set_one_json_node(data[it], field, value)
        if ret:
            return ret
    return None

def set_json_node(data: dict, field: str, value):
    pos = field.find('.')
    if pos != -1:
        parent = field[0:pos]
        node = get_json_node(data, parent)
        if node is None:
            return None
        else:
            return set_one_json_node(node, field[pos + 1:], value)
    else:
        return set_one_json_node(data, field, value)

測試程式碼如下:

class TestJsonSet:
    js = {'a': 10, 'b': 20, 'c': {'e': 10, 'f': 'string'}, 'c1': {'e': {'f1': 30}, 'f': 'string'},
          'c2': {'e': 10, 'f': 'string'}, 'z': [{'z1': 10}, {'z1': 20}]}

    def test_set1(self):
        js = copy.deepcopy(self.js)
        set_one_json_node(js, 'a', 20)
        set_one_json_node(js, 'e', 30)
        assert 20 == get_one_json_node(js, 'a')
        assert 30 == get_one_json_node(js, 'e')

    def test_set2(self):
        js = copy.deepcopy(self.js)
        set_one_json_node(js, 'c.e', 20)
        assert None == get_one_json_node(self.js, 'c.e')

    def test_set3(self):
        js = copy.deepcopy(self.js)
        set_json_node(js, 'a', 20)
        set_json_node(js, 'e', 30)
        assert 20 == get_json_node(js, 'a')
        assert 30 == get_json_node(js, 'e')
        set_json_node(js, 'c.e', 40)
        assert 40 == get_json_node(js, 'c.e')

    def test_set4(self):
        js = copy.deepcopy(self.js)
        set_json_node(js, 'z.z1', 100)
        assert 100 == get_json_node(js, 'z.z1')

pytest測試,set方法的執行時間和get方法執行時間類似,沒有出現大的變動,相比jsonpath-ng也有巨大的提升,能夠滿足要求。執行結果如下:

pytest -s test/test_jsonhelper.py::TestJsonSet
======================================================= test session starts =======================================================
platform darwin -- Python 3.8.2, pytest-6.1.2, py-1.9.0, pluggy-0.13.1
rootdir: /Users/xxx/data/code/python
collected 4 items                                                                                                                 

test/test_jsonhelper.py func:{test_set1} exec time is:{0.03638} ms
..func:{set_json_node} exec time is:{0.00293} ms
func:{set_json_node} exec time is:{0.00293} ms
func:{get_json_node} exec time is:{0.00195} ms
func:{get_json_node} exec time is:{0.00195} ms
func:{get_json_node} exec time is:{0.00098} ms
func:{set_json_node} exec time is:{0.00513} ms
func:{get_json_node} exec time is:{0.00098} ms
.func:{get_json_node} exec time is:{0.00195} ms
func:{set_json_node} exec time is:{0.01514} ms
func:{get_json_node} exec time is:{0.02588} ms

討論

我們簡單實現了json訪問的get和set方法,效能得到比較好的提升,對於頻繁訪問json物件,可以不使用jsonpath-ng,直接使用上面的方法,從而規避jsonpath-ng的效能問題。

有的時候,對於自身的需求,一定要多思考,也許存在一種簡單的方法,就能解決。不一定殺雞得用牛刀,而且可能是一把生鏽的牛刀,盲目使用第三方庫,可能會走不少彎路。

結合之前的總結,完善JsonHelper型別,這樣就可以使用了(當然程式碼的健壯性,還有不少工作,依據實際需求,大家有興趣自行完善即可)。輔助型別JsonHelper程式碼如下:

class JsonHelper:
    def __init__(self, buffer: dict):
        self.__dict__['_buffer'] = buffer

    def get(self, field: str):
        ret = self.__get_json_node(self.__dict__['_buffer'], field)

        if not ret:
            raise Exception("field:{} is not exist".format(field))
        else:
            return ret

    def set(self, field: str, value):
        ret = self.__set_json_node(self.__dict__['_buffer'], field, value)
        if ret is None:
            raise Exception("field:{} is not exist".format(field))
        return self

    def __get_one_json_node(self, data: dict, field: str):
        if isinstance(data, list):
            for it in data:
                ret = self.__get_one_json_node(it, field)
                if ret is not None:
                    return ret

        if not isinstance(data, dict):
            return None
        if field in data.keys():
            return data[field]

        for it in data.keys():
            ret = self.__get_one_json_node(data[it], field)
            if ret:
                return ret
        return None

    def __get_json_node(self, data: dict, field: str):
        node_path = field.split(".")
        node = data
        find = False
        for it in node_path:
            node = self.__get_one_json_node(node, it)
            if not node:
                return None
            else:
                find = True

        if find:
            return node
        else:
            return None

    def __set_one_json_node(self, data: dict, field: str, value):
        if isinstance(data, list):
            for it in data:
                ret = self.__set_one_json_node(it, field, value)
                if ret is not None:
                    return ret

        if not isinstance(data, dict):
            return None
        if field in data.keys():
            data[field] = value
            return data

        for it in data.keys():
            ret = self.__set_one_json_node(data[it], field, value)
            if ret:
                return ret
        return None

    def __set_json_node(self, data: dict, field: str, value):
        pos = field.find('.')
        if pos != -1:
            parent = field[0:pos]
            node = self.__get_json_node(data, parent)
            if node is None:
                return None
            else:
                return self.__set_one_json_node(node, field[pos + 1:], value)
        else:
            return self.__set_one_json_node(data, field, value)