python - 整理一系列 try...except 语句？

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。