- Python 3.8 or later
pip install requests- A free EmergencyAPI key (sign up here, no credit card)
This guide walks you through fetching real-time Australian emergency incidents with Python. By the end you will have working code that queries incidents, filters by state and type, searches by location, and handles pagination.
pip install requestsThe simplest call. Fetch the first 5 active incidents across Australia.
import requests
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://emergencyapi.com/api/v1"
response = requests.get(
f"{BASE_URL}/incidents",
headers={"Authorization": f"Bearer {API_KEY}"},
params={"limit": 5}
)
response.raise_for_status()
data = response.json()
print(f"Total active incidents: {data['metadata']['total']}")
for feature in data["features"]:
props = feature["properties"]
# geometry can be null (e.g. historical incidents that logged no location)
geom = feature["geometry"]
coords = geom["coordinates"] if geom else None
print(f" {props['title']}")
print(f" State: {props['source']['state'].upper()}")
print(f" Type: {props['eventType']}, Severity: {props['severity']}")
if coords:
print(f" Location: {coords[1]:.4f}, {coords[0]:.4f}")
print()The response is a GeoJSON FeatureCollection. Each feature has a geometry (Point or Polygon, or null when the source logged no location) and properties with title, event type, severity, warning level, source agency, and timestamps.
Add query parameters to narrow results. Multiple values are comma-separated.
# NSW bushfires only
response = requests.get(
f"{BASE_URL}/incidents",
headers={"Authorization": f"Bearer {API_KEY}"},
params={
"state": "nsw",
"eventType": "bushfire",
}
)
# Multiple states, multiple types
response = requests.get(
f"{BASE_URL}/incidents",
headers={"Authorization": f"Bearer {API_KEY}"},
params={
"state": "nsw,vic,qld",
"eventType": "bushfire,flood,storm",
"severity": "moderate,major,severe",
}
)Available filters: state, eventType, severity, warningLevel, bbox. See the full API docs for all options.
Use the /v1/incidents/nearby endpoint to search within a radius of a coordinate.
# Incidents within 50km of Sydney CBD
response = requests.get(
f"{BASE_URL}/incidents/nearby",
headers={"Authorization": f"Bearer {API_KEY}"},
params={
"lat": -33.8688,
"lng": 151.2093,
"radius": 50, # kilometres
}
)
data = response.json()
for feature in data["features"]:
dist = feature["properties"].get("distance_km", "?")
print(f" {feature['properties']['title']} ({dist} km away)")The /v1/events endpoint returns spatially clustered incidents with boundary polygons, affected suburbs, and auto-generated titles.
# Fetch all active events
response = requests.get(
f"{BASE_URL}/events",
headers={"Authorization": f"Bearer {API_KEY}"},
)
data = response.json()
for feature in data["features"]:
props = feature["properties"]
print(f" {props['title']}")
print(f" Incidents: {props['incidentCount']}, Severity: {props['severity']}")
print(f" Suburbs: {', '.join(props.get('affectedSuburbs', []))}")
print()
# Drill into a single event with its linked incidents
event_id = data["features"][0]["id"]
detail = requests.get(
f"{BASE_URL}/events/{event_id}",
headers={"Authorization": f"Bearer {API_KEY}"},
params={"include_incidents": "true"},
).json()EmergencyAPI uses cursor-based pagination. The response metadata includes anextCursor value when more results are available.
all_incidents = []
cursor = None
while True:
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
response = requests.get(
f"{BASE_URL}/incidents",
headers={"Authorization": f"Bearer {API_KEY}"},
params=params,
)
response.raise_for_status()
data = response.json()
all_incidents.extend(data["features"])
cursor = data["metadata"].get("nextCursor")
if not cursor:
break
print(f"Fetched {len(all_incidents)} total incidents")Handle rate limits (429) and other errors gracefully.
import time
def fetch_incidents(params=None):
for attempt in range(3):
response = requests.get(
f"{BASE_URL}/incidents",
headers={"Authorization": f"Bearer {API_KEY}"},
params=params or {},
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
wait = int(response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {wait}s...")
time.sleep(wait)
elif response.status_code == 401:
raise Exception("Invalid API key")
else:
response.raise_for_status()
raise Exception("Max retries exceeded")The free tier allows 500 calls per day. The rate limit headers (X-RateLimit-Remaining) tell you how many calls you have left.