def custom_join()

in pipelines/iot_analytics/iot_analytics_pipeline/pipeline.py [0:0]

• 11 lines of code
• 1 McCabe index (conditional complexity)

def custom_join(left: Dict[str, Any], right: Dict[str, Any]):
  enriched = {}
  enriched["vehicle_id"] = left["vehicle_id"]
  enriched["max_temperature"] = left["max_temperature"]
  enriched["max_vibration"] = left["max_vibration"]
  enriched["latest_timestamp"] = left["max_timestamp"]
  enriched["avg_mileage"] = left["avg_mileage"]
  enriched["last_service_date"] = right["maintenance"]["last_service_date"]
  enriched["maintenance_type"] = right["maintenance"]["maintenance_type"]
  enriched["model"] = right["maintenance"]["model"]
  return enriched