【问题标题】:Simpler way to traverse complex dictionary in python?在python中遍历复杂字典的更简单方法?
【发布时间】:2015-02-25 02:24:24
【问题描述】:

我正在迭代作为字典加载到 python 中的复杂 json 对象。以下是 json 文件的示例。对感兴趣的数据进行了注释。

{  
   "name":"ns1:timeSeriesResponseType",
   "nil":false,
   "value":{  
      "queryInfo":{  },
      "timeSeries":[  
         {  
            "variable":{  },
            "values":[  
               {  
                  "qualifier":[  ],
                  "censorCode":[  ],
                  "value":[  
                     {  
                        "codedVocabularyTerm":null,
                        "censorCode":null,
                        "offsetTypeID":null,
                        "accuracyStdDev":null,
                        "timeOffset":null,
                        "qualifiers":[  
                           "P",                      # data of interest
                           "Ice"                     # data of interest
                        ],
                        "qualityControlLevelCode":null,
                        "sampleID":null,
                        "dateTimeAccuracyCd":null,
                        "methodCode":null,
                        "codedVocabulary":null,
                        "sourceID":null,
                        "oid":null,
                        "dateTimeUTC":null,
                        "offsetValue":null,
                        "metadataTime":null,
                        "labSampleCode":null,
                        "methodID":null,
                        "value":"-999999",
                        "dateTime":"2015-02-24T03:30:00.000-05:00",
                        "offsetTypeCode":null,
                        "sourceCode":null
                     },
                     {  
                        "codedVocabularyTerm":null,
                        "censorCode":null,
                        "offsetTypeID":null,
                        "accuracyStdDev":null,
                        "timeOffset":null,
                        "qualifiers":[  ],
                        "qualityControlLevelCode":null,
                        "sampleID":null,
                        "dateTimeAccuracyCd":null,
                        "methodCode":null,
                        "codedVocabulary":null,
                        "sourceID":null,
                        "oid":null,
                        "dateTimeUTC":null,
                        "offsetValue":null,
                        "metadataTime":null,
                        "labSampleCode":null,
                        "methodID":null,
                        "value":"-999999",                          # data of interest
                        "dateTime":"2015-02-24T04:00:00.000-05:00", # data of interest
                        "offsetTypeCode":null,
                        "sourceCode":null
                     }
                  ],
                  "sample":[  ],
                  "source":[  ],
                  "offset":[  ],
                  "units":null,
                  "qualityControlLevel":[  ],
                  "method":[  ]
               }
            ],
            "sourceInfo":{  },
            "name":"USGS:03193000:00060:00011"
         },
         {  },  # more data need is stored in here
         {  },  # more data need is stored in here
         {  }   # more data need is stored in here
      ]
   },
   "declaredType":"org.cuahsi.waterml.TimeSeriesResponseType",
   "scope":"javax.xml.bind.JAXBElement$GlobalScope",
   "globalScope":true,
   "typeSubstituted":false
}

这是我的代码,用于单步执行/迭代字典以获取我想要的数据并将其存储在格式更简单的字典中:

# Setting up blank variables to store results
outputDict = {}
outputList = []
dateTimeList = []
valueList = []
qualifiersList = [[]]


for key in result["value"]["timeSeries"]:
    for key2 in key:
        if key2 == "values":
            for key3 in key.get(key2):
                for key4 in key3:
                    if key4 == "value":
                        for key5 in key3.get(key4):
                            for key6 in key5:
                                if key6 == "value":
                                    valueList.append(key5.get(key6))
                                if key6 == "dateTime":
                                    dateTimeList.append(key5.get(key6))
                        #print key.get("name")
                        #outputDict[key.get("name")]["dateTime"] = dateTimeList
                        #outputDict[key.get("name")]["values"] = valueList

        if key2 == "name":
            outputList.append(key.get(key2))
            outputDict[key.get(key2)]={"dateTime":None, "values":None, "qualifiers":None}
            outputDict[key.get("name")]["dateTime"] = dateTimeList
            outputDict[key.get("name")]["values"] = valueList
            del dateTimeList[:]
            del valueList[:]

我的问题是——对 python 有点陌生,谁能指出我的代码中任何明显的低效率?我可以指望 json 文件在几个月(可能是几年)内不会改变结构,所以我相信我最初使用 for 键在 result["value"]["timeSeries"]: 很好,但我不确定很多很多 for 循环是否不必要或效率低下。有没有一种简单的方法可以从这样的分层字典中搜索并返回键:值对,字典列表位于字典列表中?

编辑:

基于@Alex Martelli 提供的解决方案,这里是新的、更高效、精简的代码版本:

# Building the output dictionary
for key in result["value"]["timeSeries"]:
    if "values" in key:
        for key2 in key.get("values"):
            if "value" in key2:
                for key3 in key2.get("value"):
                    if "value" in key3:
                        valueList.append(key3.get("value"))
                    if "dateTime" in key3:
                        dateTimeList.append(key3.get("dateTime"))
                    if "qualifiers" in key3:
                        qualifiersList.append(key3.get("qualifiers"))

    if "name" in key:
        outputList.append(key.get("name"))
        outputDict[key.get("name")]={"dateTime":None, "values":None, "qualifiers":None}
        outputDict[key.get("name")]["dateTime"] = dateTimeList[:]    # passing the items in the list rather
        outputDict[key.get("name")]["values"] = valueList[:]         # than a reference to the list so the delete works
        outputDict[key.get("name")]["qualifiers"] = qualifiersList[:]         # than a reference to the list so the delete works
        del dateTimeList[:]
        del valueList[:]
        del qualifiersList[:]

工作原理相同,删除了 4 行代码。运行时间更快。不错。

编辑:

基于@Two-Bit Alchemist 提出的解决方案,这同样有效:

# Building the output dictionary
    for key in result["value"]["timeSeries"]:
        print key
        for value in key["values"][0]["value"]:
            # qualifiers is a list containing ["P", "Ice"]
            qualifiersList.append(value['qualifiers'])
            valueList.append(value['value'])
            dateTimeList.append(value['dateTime'])


        if "name" in key:
            outputList.append(key.get("name"))
            outputDict[key.get("name")]={"dateTime":None, "values":None, "qualifiers":None}
            outputDict[key.get("name")]["dateTime"] = dateTimeList[:]    # passing the items in the list rather
            outputDict[key.get("name")]["values"] = valueList[:]         # than a reference to the list so the delete works
            outputDict[key.get("name")]["qualifiers"] = qualifiersList[:]         # than a reference to the list so the delete works
            del dateTimeList[:]
            del valueList[:]
            del qualifiersList[:]

我看到的唯一问题是我无法完全确定 ["values"] 列表中的第一个位置是我想要的。而且我丢失了“if”语句提供的检查,这些检查应该确保在错误的查询语句返回值时不会引入错误。

编辑:

try:

    # requests.get returns a "file-like" object
    # in this case it is a JSON object because of the settings in the query
    response = requests.get(url=query)


    # if-else ladder that only performs the parsing of the returned JSON object
    # when the HTTP status code indicates a successful query execution
    if(response.status_code == 200):

        # parsing the
        result = response.json()

        # Setting up blank variables to store results
        outputDict = {}
        outputList = []
        dateTimeList = []
        valueList = []
        qualifiersList = []


        # Building the output dictionary
        for key in result["value"]["timeSeries"]:
            print key
            for value in key["values"][0]["value"]:
                # qualifiers is a list containing ["P", "Ice"]
                qualifiersList.append(value['qualifiers'])
                valueList.append(value['value'])
                dateTimeList.append(value['dateTime'])

            # OLD CODE   
            # if "values" in key:
            #     for key2 in key.get("values"):
            #         if "value" in key2:
            #             for key3 in key2.get("value"):
            #                 if "value" in key3:
            #                     valueList.append(key3.get("value"))
            #                 if "dateTime" in key3:
            #                     dateTimeList.append(key3.get("dateTime"))
            #                 if "qualifiers" in key3:
            #                     qualifiersList.append(key3.get("qualifiers"))

            if "name" in key:
                outputList.append(key.get("name"))
                outputDict[key.get("name")]={"dateTime":None, "values":None, "qualifiers":None}
                outputDict[key.get("name")]["dateTime"] = dateTimeList[:]    # passing the items in the list rather
                outputDict[key.get("name")]["values"] = valueList[:]         # than a reference to the list so the delete works
                outputDict[key.get("name")]["qualifiers"] = qualifiersList[:]         # than a reference to the list so the delete works
                del dateTimeList[:]
                del valueList[:]
                del qualifiersList[:]


        # Tracking how long it took to process the data
        elapsed = time.time() - now
        print "Runtime: " + str(elapsed)

        out = {"Status": 'ok', "Results": [[{"myResult": outputDict}]]}

    elif(response.status_code == 400):
        raise Exception("Bad Request, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    elif(response.status_code== 403):
        raise Exception("Access Forbidden, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    elif(response.status_code == 404):
        raise Exception("Gage location(s) not Found, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    elif(response.status_code == 500):
        raise Exception("Internal Server Error, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    elif(response.status_code == 503):
        raise Exception("Service Unavailable, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    else:
        raise Exception("Unknown Response, "+ datetime.now().strftime('%Y-%m-%d %H:%M:%S'))



except:
    out = {"Status": 'Error', "Message": str(sys.exc_info()[1])}


print out

【问题讨论】:

  • 我想你只需要访问它,对吧?如果dct 是您的json 结构,则类似于dct['value']['timeSeries'][0]['values'][0]['value'][0]['qualifiers']。这可能并不完全正确,因为,是的,这件事非常复杂!
  • 所以,如果我知道对我的查询可能有多少响应,这将起作用。但是,任何时候都可能有 1 到几十个响应,我不知道有多少响应,所以我不知道在您的解决方案中放置(或循环)多少 i。如果我总是知道回复的总数,你的建议会奏效。但我不确定它是否可以。
  • 你将这个结构的哪一部分称为“响应”,其中有可变数量?
  • 啊,是的 - 我不够清楚。整个结构 - 加载到 python 字典中的 json 对象 - 是对 Web 服务查询的响应。内部两个主要部分“timeSeries”(包含我感兴趣的所有数据)和“queryInfo”(我不感兴趣)将根据我针对 Web 服务执行的查询的特定属性而增长.我需要编写此代码来处理可能返回“timeSeries”下的字典列表中包含 1 到几十个成员的列表的查询。
  • 我仍然很困惑为什么这个问题不能用 one for loop 解决——我坦率地承认我可能仍然误解了这个问题。类似for data in json_structure['value']['timeSeries']: data['values'][0]['value'][0]['qualifiers'] # etc 的东西——如果存储在timeSeries 的列表长度未知,只需遍历它并重复该过程。

标签: python dictionary iteration


【解决方案1】:

您问“我的代码中是否存在任何明显的低效率”——答案是肯定的,特别是在您遍历字典的地方(因此按顺序获取它们的所有键,即 O(N),即,所花费的时间与字典中键的数量)而不是仅仅将它们用作字典(这需要时间O(1),即恒定时间——也很快)。

例如你在哪里

for key2 in key:
    if key2 == "values":
       ...use key.get(key2)...
    if key2 == "name":
       ...use key.get(key2)...

你应该有:

if 'values' in key:
   ...use key['values']...
if 'name' in key:
   ...use key['name']...

以及更深层次的类似结构。可以进一步优化事物,例如:

values = key.get('values')
if values is not None:
    ...use values...
name = key.get('name')
if name is not None:
    ...use name...

避免重复索引(同样,更深入)。

【讨论】:

  • 谢谢!我还不太明白,但我会努力的!
  • in 运算符,当 RHS 是字典时,如果 LHS 是该字典中的键,则返回 true。字典的get 方法在使用不是该字典中的键的参数调用时返回None。这就是它的全部内容!顺便说一句,请记住最终接受答案,即单击它左侧的复选标记轮廓......除非你得到一个更好的答案,现在看起来不太可能:-)
  • 即将发布我的实现 - 期待看看我是否错过了任何效率提升。非常感谢!
  • @traggatmot,不客气——如果您将其作为单独的 Q 发布,您无疑会得到进一步优化的答案!
  • @AlexMartelli 我认为很多 OP 的代码是由于对公认的复杂数据结构的一种睁大眼睛的反应,以及对 access/EAFP 的误解。这可以更简单地完成。 (见我的回答和更新。)
【解决方案2】:

就我理解这个问题的方式而言,我相信您最初对该问题的令人困惑的方法,这种方法过于矫枉过正,使一个非常简单的解决方案变得模糊不清。如果我仍然误解这一点并且过于简单化,请纠正我。尽管这个结构非常复杂,但如果它的可变部分是timeSeries 处的列表长度,您可以访问该列表并对其进行迭代,同时反复抓取您的“感兴趣的数据”。我不知道这些数据是什么,可以为您提供一个很好的示例数据结构,甚至是体面的变量名称,以便在以后的程序中使用它,所以我只是将它存储在一个大列表中列出只是为了向您展示我的意思:

data_of_interest = []
for data in json_structure['value']['timeSeries']:
    value_list = data['values'][0]['value']
    # qualifiers is a list containing ["P", "Ice"]
    qualifiers = value_list[0]['qualifiers']
    value = value_list[1]['value']
    dateTime = value_list[1]['dateTime']
    data_of_interest.append([qualifiers, value, dateTime])

如果在我硬编码索引 0 的其他地方有重复,只需在那里引入 for 循环,例如

for data in json_structure['value']['timeSeries']:
    for value_set in data['values']:
        for value_list in value_set['value']:
            # etc

如果您担心某些值不存在,只需准备从 dict 或等效项中捕获 KeyError

例如,我写的地方:

value = value_list[1]['value']

如果value_list 没有>= 2 个值,这可能会引发IndexError,如果它的第二个元素不是将“值”映射到某物的字典,则可能会引发KeyError。您可以抓住其中一个或两个,将它们一起或单独处理,或者直接忽略它们并继续。

try:
    value = value_list[1]['value']
except KeyError:    # catch only one
    # do something

try:
    value = value_list[1]['value']
except (IndexError, KeyError):    # catch both
    # handle together

try:
    value = value_list[1]['value']
except IndexError:
    # handle IndexError
except KeyError:
    # handle KeyError

你的# handle whatever 代码很可能是pass——这只是意味着“我知道这可能会发生,但不要惊慌。继续阅读。”如果您捕获它们,异常将“冒泡”到执行上下文的顶部并使您的程序崩溃。

【讨论】:

  • 更新了问题以显示您的解决方案的实施 - 我仍然担心它,因为它丢失了 if 语句提供的检查。我相信,尽管我不完全确定,当查询结果不包含预期值时,这些 if 语句应该可以防止代码失败。
  • 字典的好处在于,如果这些值不存在,它们会引发KeyError。因此,只需用 try/except 包围您的访问并抓住它。我会更新我的答案。
  • 顺便说一句,使用 if 语句进行保护称为“跳前检查”或 LBYL,而使用我向您展示的 try/except 称为“请求宽恕比许可更容易”或 EAFP。后者在 Python 中是首选,因为它在多线程环境等方面具有优势。
  • 我已经在 try/catch 语句中捕获了这个代码。我将发布整个代码。
  • 我需要弄清楚如何将您所说的那些 KeyErrors 传递给更大的 try/catch。
猜你喜欢
  • 2022-10-17
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2019-12-20
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多