Python Quickstart: Australian Emergency Data in 5 Minutes

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.

Prerequisites
  • Python 3.8 or later
  • pip install requests
  • A free EmergencyAPI key (sign up here, no credit card)
1. Fetch All Active Incidents

The simplest call. Fetch the first 5 active incidents across Australia.

Python
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.

2. Filter by State and Event Type

Add query parameters to narrow results. Multiple values are comma-separated.

Python
# 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.

3. Find Incidents Near a Location

Use the /v1/incidents/nearby endpoint to search within a radius of a coordinate.

Python
# 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)")
4. Get Clustered Events

The /v1/events endpoint returns spatially clustered incidents with boundary polygons, affected suburbs, and auto-generated titles.

Python
# 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()
5. Pagination

EmergencyAPI uses cursor-based pagination. The response metadata includes anextCursor value when more results are available.

Python
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")
6. Error Handling

Handle rate limits (429) and other errors gracefully.

Python
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.

Next Steps