整理一系列try ...语句以外的内容? [英] Tidying up series of try... except statements?

问题描述

我已经编写了一个脚本，用于将Kustomer API中的数据传递到我们的数据库中，尽管它工作得很好，但有点混乱，但我想知道是否有一个更优雅的解决方案.我将结果行定义为字典，然后推入MySQL，但是当其中的某些值并非始终在JSON中可用时，就会出现混乱的情况.

I've written a script to pipe through data from the Kustomer API to our database, and although it works fine its a bit messy, was wondering if there's a more elegant solution to this. I'm defining the row of results im pushing through as a dictionary then pushing to MySQL, but the messy part comes when some of these values aren't available in the JSON all the time.

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

This has resulted in a try / except statements for each data point that may or may not be missing.

是否有更好的方法?下面的代码.

Is there a better way of doing this? Code below.

    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

推荐答案

首先，在可能的情况下，您应尝试使用指定了默认值的dict.get.接下来，您可以考虑 contextmanager 来使您的代码显着清洁工.考虑一下:

First you should try, where possible, to use dict.get with a default value specified. Next you can consider contextmanager to make your code significantly cleaner. Consider this:

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

现在重新编写，您可以确保一致错误处理，而无需重复逻辑:

Now rewritten, you can ensure consistent error handling without repeating logic:

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.

You can define an arbitrary number of functions like error_handling for various rules you wish to apply.

这篇关于整理一系列try ...语句以外的内容?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！