def process_event()

in functions/source/asset-model-converter/handler.py [0:0]

• 63 lines of code
• 17 McCabe index (conditional complexity)

def process_event(event, integration_points):
    if not event['type'] == 'lifecycle':
        return None

    asset = None
    integration_point_ids = [integration['Id'] for integration in integration_points]

    if event['reading']['et'] == 'model_updated':
        model_converter.setup_model(event['reading']['asset_model'])

    elif event['reading']['et'] == 'root_updated':
        asset = asset_converter.setup_asset( 
            obj_type="root",
            external_id=event['reading']['id'], 
            name=event['reading']['name'], 
            model_name="UrbanIO Root", 
            hierarchy_name=None
        )

    elif event['reading']['et'] == 'operator_updated' and event['reading']['id'] in integration_point_ids:
        # Remove superoperator name from ancestors
        ancestor_names = event['reading']['ancestor_names']
        ancestor_names.pop(0)
        asset = asset_converter.setup_asset( 
            obj_type="operator",
            external_id=event['reading']['id'],
            name=' / '.join(ancestor_names), 
            model_name="UrbanIO Operator", 
            hierarchy_name='Operators'
        )

    elif event['reading']['et'] == 'customer_updated':
        parent_operators = [integration for integration in integration_points if integration['Id'] in event['metadata']['ref']['o']]
        parent_operator_names = parent_operators[0]['Reading']['ancestor_names'] if len(parent_operators) > 0 else []
        sub_operator_names = [x for x in event['reading']['ancestor_names'] if x not in parent_operator_names]
        customer_name = f"{'/'.join(sub_operator_names)}/{event['reading']['name']}" if len(sub_operator_names) > 0 else event['reading']['name']
        asset = asset_converter.setup_asset( 
            obj_type="customer",
            external_id=event['reading']['id'], 
            name= customer_name, 
            model_name="UrbanIO Customer", 
            hierarchy_name='Customers'
        )

    elif event['reading']['et'] == 'location_updated':
        asset = asset_converter.setup_asset( 
            obj_type="location",
            external_id=event['reading']['id'], 
            name= f"{event['reading']['name']} ({event['reading']['id']})", 
            model_name="UrbanIO Location", 
            hierarchy_name='Locations'
        )

    elif event['reading']['et'] == 'category_updated':
        asset = asset_converter.setup_asset( 
            obj_type="category",
            external_id=event['reading']['id'], 
            name= f"{event['reading']['name']} ({event['metadata']['ref']['l']})", 
            model_name=f"UrbanIO {event['reading']['name'].capitalize()} Sensors", 
            hierarchy_name=f"{event['reading']['name'].capitalize()} Devices"
        )       

    elif event['reading']['et'] == 'device_updated':
        asset = asset_converter.setup_asset( 
            obj_type="device",
            external_id=event['reading']['id'], 
            name= f"{event['reading']['name']} ({event['reading']['id']})", 
            model_name=f"UrbanIO {event['reading']['device_type']} Device", 
            hierarchy_name=f"{event['reading']['device_type']} Devices",
            datastream_descriptions=event['reading']['datastreams']
        )

    return asset