Poner en orden una serie de bashs … ¿excepto declaraciones?

He escrito un script para pasar a través de los datos de la API de Kustomer a nuestra base de datos, y aunque funciona bien, es un poco complicado, me preguntaba si existe una solución más elegante para esto. Estoy definiendo la fila de resultados que estoy enviando a través de un diccionario y luego empujando a MySQL, pero la parte complicada llega cuando algunos de estos valores no están disponibles en el JSON todo el tiempo.

Esto ha dado lugar a una statement de prueba / excepción para cada punto de datos que puede o no estar ausente.

¿Hay una mejor manera de hacer esto? Código abajo.

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 

Primero debe intentar, cuando sea posible, usar dict.get con un valor predeterminado especificado. A continuación, puede considerar contextmanager para que su código sea significativamente más limpio. Considera esto:

 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 

Ahora reescrito, puede garantizar un manejo de errores consistente sin repetir la lógica:

 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'] 

Puede definir un número arbitrario de funciones, como error_handling para varias reglas que desee aplicar.

Puede usar la función, que le da un elemento de los dicts nesteds, y no genera una excepción, si no existe.

Como este borrador rápido:

 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 

Entonces, en lugar de:

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

Simplemente escribe:

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

Las diferentes convenciones de nomenclatura en los lados de entrada y salida hacen que sea difícil superar la claridad de las asignaciones explícitas. Preservar la semántica exacta de su versión ( p . Ej. , Que no asigna a conversation_category en ausencia de una typeStr incluso si categoryTree está disponible) excluye ciertas opciones (como hacer que una estructura de datos se typeStr con un try / except en cada acceso) ; es posible que pueda hacerlo mejor con más suposiciones sobre sus datos de entrada.

No obstante, además del dict.get ya mencionado , puede utilizar los elementos incorporados ( any or , y dict ) e introducir una función auxiliar y algunas variables temporales para hacer que el código sea mucho más legible:

 # 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 

Si bien es posible que tenga una política en contra, en este caso recomiendo escribir las cláusulas restantes except KeyError: pass en una línea cada una: ayuda al corchete visual del código tentativo.