整理一系列尝试...除了语句？

Question

我编写了一个脚本来将 Kustomer API 中的数据通过管道传输到我们的数据库，虽然它工作正常但有点混乱，我想知道是否有更优雅的解决方案。我将结果行定义为我作为字典推送然后推送到 MySQL，但是当其中一些值在 JSON 中始终不可用时，就会出现混乱的部分。

这导致每个可能丢失或可能不丢失的数据点的 try / except 语句。

有没有更好的方法来做到这一点？代码如下。

    try:
        record_data = {
            'id': record['id'],
            'created_at': str(datetime.strptime(record['attributes']['createdAt'], '%Y-%m-%dT%H:%M:%S.%fZ'))[:-7],
            'last_activity_at': str(datetime.strptime(record['attributes']['lastActivityAt'], '%Y-%m-%dT%H:%M:%S.%fZ'))[:-7],
            'first_marked_done': None,
            'last_marked_done': None,
            'assigned_team': record['attributes']['assignedTeams'][0] if record['attributes']['assignedTeams'] != [] else None,
            'conversation_type': None,
            'conversation_category': None,
            'conversation_subcategory': None,
            'message_count': record['attributes']['messageCount'],
            'note_count': record['attributes']['noteCount'],
            'satisfaction': record['attributes']['satisfaction'],
            'status': None,
            'email': 1 if len(list(filter(lambda x: x == 'email', record['attributes']['channels']))) > 0 else 0,
            'chat': 1 if len(list(filter(lambda x: x == 'chat', record['attributes']['channels']))) > 0 else 0,
            'priority': record['attributes']['priority'],
            'direction': 'outbound' if record['attributes']['direction'] == 'out' else 'in',
            'nlp_score': None,
            'nlp_sentiment': None,
            'waiting_for': None,
            'sla_breach': None,
            'sla_status': None,
            'breached_sla': None,
            'breached_at': None
        }
        try:
            record_data['status'] = record['attributes']['status']
        except KeyError:
            pass
        try:
            record_data['conversation_type'] = record['attributes']['custom']['typeStr']
            record_data['conversation_category'] = str(record['attributes']['custom']['categoryTree']).split('.')[0]
            record_data['conversation_subcategory'] = str(record['attributes']['custom']['categoryTree']).split('.')[1] if len(str(record['attributes']['custom']['categoryTree']).split('.')) > 1 else None
        except KeyError:
            pass
        try:
            record_data['waiting_for'] = record['attributes']['custom']['typeStr']
        except KeyError:
            pass
        try:
            record_data['first_marked_done'] = str(datetime.strptime(record['attributes']['firstDone']['createdAt'], '%Y-%m-%dT%H:%M:%S.%fZ'))[:-7]
            record_data['last_marked_done'] = str(datetime.strptime(record['attributes']['lastDone']['createdAt'], '%Y-%m-%dT%H:%M:%S.%fZ'))[:-7]

        except KeyError:
            pass
        try:
            record_data['sla_breach'] = 0 if record['attributes']['sla']['breached'] is False else 1
            record_data['sla_status'] = record['attributes']['sla']['status']
            if record_data['sla_breach'] == 1:
                try:
                    record_data['breached_sla'] = record['attributes']['sla']['breach']['metric']
                    record_data['breached_at'] = record['attributes']['sla']['breach']['at']
                except KeyError:
                    for m in record['attributes']['sla']['metrics']:
                        try:
                            if record['attributes']['sla']['metrics'][m]['breachAt'] == record['attributes']['sla']['summary']['firstBreachAt']:
                                record_data['breached_sla'] = m
                                record_data['breached_at'] = str(datetime.strptime(record['attributes']['sla']['summary']['firstBreachAt'], '%Y-%m-%dT%H:%M:%S.%fZ'))[:-7]
                        except KeyError:
                            pass
        except KeyError:
            record_data['sla_breach'] = 0
        print(record_data)
        self.db.insert_update(KustomerConversations(**record_data))

    except KeyError:
        pass

Answer 1

首先，您应该尽可能尝试使用 dict.get 并指定默认值。接下来，您可以考虑使用contextmanager 来使您的代码更加简洁。考虑一下：

try:
    record_data['status'] = record['attributes']['status']
except KeyError:
    pass
try:
    record_data['conversation_type'] = record['attributes']['custom']['typeStr']
except KeyError:
    pass
try:
    record_data['waiting_for'] = record['attributes']['custom']['typeStr']
except KeyError:
    pass
try:
    record_data['first_marked_done'] = record['attributes']['firstDone']['createdAt']
except KeyError:
    pass

现在重写，可以保证一致错误处理，无需重复逻辑：

from contextlib import contextmanager

@contextmanager
def error_handling():
    try:
        yield
    except KeyError:
        pass

with error_handling():
    record_data['status'] = record['attributes']['status']
with error_handling():
    record_data['conversation_type'] = record['attributes']['custom']['typeStr']
with error_handling():
    record_data['waiting_for'] = record['attributes']['custom']['typeStr']
with error_handling():
    record_data['first_marked_done'] = record['attributes']['firstDone']['createdAt']

您可以为您希望应用的各种规则定义任意数量的函数，例如 error_handling。

Answer 2

您可以使用函数，它为您提供嵌套字典中的元素，并且不会引发异常（如果它不存在）。

喜欢这个速写：

def get_nested_dict_value(src_dict, *nested_keys, **kwargs):
    """
    Get value of some nested dict by series of keys with default value.
    Example:
    instead of: 
        x = data['a']['b']['c']['d']
    use
        x = get_nested_dict_value(data, 'a', 'b', 'c', 'd')
    or, if you need some non-None default value, add default=xxx kwarg:
        x = get_nested_dict_value(data, 'a', 'b', 'c', 'd', default=0)
    """
    default = kwargs.get("default", None)
    pointer = src_dict
    i = 0
    for key in nested_keys:
        i += 1
        if key in pointer:
            pointer = pointer[key]
            if i == len(nested_keys):
                return pointer
        else:
            return default

所以，而不是：

try:    
    record_data['conversation_type'] = record['attributes']['custom']['typeStr']
except Exception:
    pass

你只需输入：

record_data['conversation_type'] = get_nested_dict_value(record, 'attributes', 'custom', 'typeStr')

Answer 3

输入端和输出端的不同命名约定使得显式分配的清晰度难以超越。保留您的版本的确切语义（例如，即使categoryTree 可用，它也不会在没有typeStr 的情况下分配conversation_category）排除某些选择（例如在 each 访问时使用 try/except 循环创建数据结构）；对输入数据进行更多假设可能会做得更好。

不过，除了 dict.get already mentioned，您还可以使用内置函数（any、or 和 dict）并引入一个辅助函数和一些临时变量来创建代码更具可读性：

# this gives one digit of the hour for me...?
def ptime(s): return str(datetime.strptime(s,'%Y-%m-%dT%H:%M:%S.%fZ'))[:-7]

try:
    attr=record['attributes']
    cust=attr.get('custom',{})  # defer KeyErrors into the below
    record_data = dict(
        id = record['id'],
        created_at = ptime(attr['createdAt']),
        last_activity_at = ptime(attr['lastActivityAt']),
        first_marked_done = None,
        last_marked_done = None,
        assigned_team = attr['assignedTeams'][0] or None,
        conversation_type = None,
        conversation_category = None,
        conversation_subcategory = None,
        message_count = attr['messageCount'],
        note_count = attr['noteCount'],
        satisfaction = attr['satisfaction'],
        status = attr.get('status'),
        email = int(any(x == 'email' for x in attr['channels'])),
        chat = int(any(x == 'chat' for x in attr['channels'])),
        priority = attr['priority'],
        direction = 'outbound' if attr['direction'] == 'out' else 'in',
        nlp_score = None,
        nlp_sentiment = None,
        waiting_for = cust.get('typeStr'),
        sla_breach = 0,
        sla_status = None,
        breached_sla = None,
        breached_at = None
    )
    try:
        record_data['conversation_type'] = cust['typeStr']
        cat=str(cust['categoryTree']).split('.')
        record_data['conversation_category'] = cat[0]
        record_data['conversation_subcategory'] = cat[1] if len(cat) > 1 else None
    except KeyError: pass
    try:
        record_data['first_marked_done'] = ptime(attr['firstDone']['createdAt'])
        record_data['last_marked_done'] = ptime(attr['lastDone']['createdAt'])
    except KeyError: pass
    try:
        sla=attr['sla']
        record_data['sla_breach'] = 0 if sla['breached'] is False else 1
        record_data['sla_status'] = sla['status']
        if record_data['sla_breach'] == 1:
            try:
                record_data['breached_sla'] = sla['breach']['metric']
                record_data['breached_at'] = sla['breach']['at']
            except KeyError:
                for m,v in sla['metrics'].items():
                    try:
                        v=v['breachAt']
                        if v == sla['summary']['firstBreachAt']:
                            record_data['breached_sla'] = m
                            record_data['breached_at'] = ptime(v)
                    except KeyError: pass
    except KeyError: pass
    print(record_data)
    self.db.insert_update(KustomerConversations(**record_data))

except KeyError: pass

虽然您可能有反对它的政策，但在这种情况下，我建议将剩余的 except KeyError: pass 子句各写一行：它有助于对暂定代码进行可视化括号。