# Step 1: Install what we need
%pip install azure-eventhub requests
# Step 2: Import libraries
import json
import requests
import pandas as pd
from datetime import datetime
from azure.eventhub import EventHubProducerClient, EventData
import time
# Step 3: Your connection details
eventhub_namespace = "esehlnrljzmeus6ltvsfen.servicebus.windows.net"
eventhub_name = "xxx"
sas_key_name = "xxx"
sas_key_value = "xxx"
connection_str = (
f"Endpoint=sb://{eventhub_namespace}/;"
f"SharedAccessKeyName={sas_key_name};"
f"SharedAccessKey={sas_key_value};"
f"EntityPath={eventhub_name}"
)
api_params = {
"resource_id": "292f788f-4339-455b-8cc0-153e14509d4d",
"filters": json.dumps({"Region": "South West"})
}
api_headers = {
"Authorization": (
"xxx"
"xxx"
"xxx"
)
}
print(f"Setup complete at {datetime.now()}")
# Step 4: Get data from National Grid API with proper error handling
print(f"Fetching data at {datetime.now()}")
records = []
api_success = False
try:
# Add timeout and proper error handling
response = requests.get(api_url, params=api_params, headers=api_headers, timeout=30)
response.raise_for_status() # Raises exception for bad status codes
data = response.json()
records = data.get('result', {}).get('records', [])
print(f"Found {len(records)} records")
if not records:
print("No records found - notebook will complete without processing")
else:
api_success = True
except requests.exceptions.Timeout:
print("API request timed out - notebook will complete without processing")
except requests.exceptions.RequestException as e:
print(f"API request failed: {e} - notebook will complete without processing")
except json.JSONDecodeError as e:
print(f"JSON decode error: {e} - notebook will complete without processing")
except Exception as e:
print(f"Unexpected error fetching data: {e} - notebook will complete without processing")
def clean_and_stringify_record(record):
"""
Cleans the record by:
- Removing spaces from keys
- Converting all values to strings (or empty string if None)
- Adding notebook_processed_time timestamp
"""
try:
cleaned = {k.replace(" ", ""): str(v) if v is not None else "" for k, v in record.items()}
# Add timestamp when record was processed by this notebook
cleaned['notebook_processed_time'] = datetime.now().isoformat()
return cleaned
except Exception as e:
print(f"Error cleaning record: {e}")
return {}
# Step 5: Send to Event Hub with proper error handling and batching
def send_to_eventhub_in_batches(records, batch_size=100):
"""Send records to Event Hub in batches with proper error handling"""
try:
producer = EventHubProducerClient.from_connection_string(connection_str)
total_sent = 0
with producer:
for i in range(0, len(records), batch_size):
batch_records = records[i:i + batch_size]
try:
event_batch = producer.create_batch()
for record in batch_records:
cleaned_record = clean_and_stringify_record(record)
if cleaned_record: # Only add non-empty records
try:
event_batch.add(EventData(json.dumps(cleaned_record)))
except Exception as e:
print(f"Failed to add record to batch: {e}")
continue
if len(event_batch) > 0:
producer.send_batch(event_batch)
batch_num = i // batch_size + 1
print(f"Sent batch {batch_num}: {len(batch_records)} records")
total_sent += len(batch_records)
else:
print(f"Batch {i // batch_size + 1} was empty - skipping")
except Exception as e:
print(f"Error processing batch {i // batch_size + 1}: {e}")
continue
# Small delay between batches to prevent overwhelming the service
time.sleep(0.1)
print(f"Successfully sent {total_sent} records to Event Hub")
return total_sent
except Exception as e:
print(f"Event Hub connection error: {e}")
return 0
# Execute the sending process
if records and api_success:
print(f"Starting to send {len(records)} records to Event Hub")
sent_count = send_to_eventhub_in_batches(records)
if sent_count > 0:
print(f"Successfully processed {sent_count} records")
execution_status = "SUCCESS"
else:
print("Failed to send any records")
execution_status = "FAILED"
else:
print("No records to process or API call failed")
execution_status = "SKIPPED"
# Final completion message
print(f"Notebook execution completed with status: {execution_status} at {datetime.now()}")
print("All done! Event Hub ingestion process complete.")
# Force completion by ensuring all variables are properly cleaned up
del records, api_success, execution_status
print("Notebook execution finished.")
