Batch Map Export: Automation via API

Exporting a single map manually is easy. But what if you need to download 50 cities in GeoJSON, 200 neighborhoods in DXF, or update data daily for 10 regions? That's what the REST API with batch export support is for.

API Overview

The osm2cdr.ru API follows REST principles. Key endpoints:

API access requires an API key, available from your dashboard. It goes in the X-API-Key header.

Single Export: curl

Let's start with a simple request to export one area. Note the shape of the body: bbox is an object with min_lon/min_lat/max_lon/max_lat, and formats is a list (1 to 5 formats at a time).

curl -X POST https://osm2cdr.ru/api/render \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76},
    "formats": ["geojson"],
    "map_style": "standard",
    "detail_level": 4
  }'

The response carries a task id — the file itself arrives later, in a separate request:

{
  "task_id": "abc123-def456",
  "status": "pending",
  "created_at": "2026-08-07T13:56:03.726383Z",
  "formats": ["geojson"]
}

Checking status and downloading:

# Check status
curl -H "X-API-Key: YOUR_API_KEY" \
  https://osm2cdr.ru/api/status/abc123-def456

# Download result (when status = "completed"); the format is required in the path
curl -H "X-API-Key: YOUR_API_KEY" \
  -o moscow_center.geojson \
  https://osm2cdr.ru/api/download/abc123-def456/geojson

Batch Export: batch endpoint

The batch endpoint accepts an array of areas and returns a single batch_id for the whole pack. This is more efficient than sending requests one by one — the server optimizes execution and can reuse the PostGIS cache, and the result comes back as one archive.

Each item is its own area with its own list of formats; render settings go into that item's config.

curl -X POST https://osm2cdr.ru/api/batch/export \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "items": [
      {"bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76}, "formats": ["geojson"]},
      {"bbox": {"min_lon": 30.28, "min_lat": 59.92, "max_lon": 30.36, "max_lat": 59.96}, "formats": ["geojson"]},
      {"bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76}, "formats": ["dxf"]}
    ],
    "archive_format": "zip"
  }'

Response (HTTP 202):

{
  "batch_id": "b7f1c0e2-9a44-4f2b-8d31-0e5a1c9b7a20",
  "status": "pending",
  "items_total": 3,
  "items_completed": 0,
  "items_failed": 0
}

Batch limits:

On top of that: up to 5 formats per area, and archive_format is either zip or tar.gz. Without authentication the batch endpoint answers 403 — anonymous batch export is closed.

Python Automation

Exporting a List of Cities

import requests
import time
from pathlib import Path

API_URL = "https://osm2cdr.ru/api"
API_KEY = "YOUR_API_KEY"
HEADERS = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

# bbox order: [min_lon, min_lat, max_lon, max_lat]
cities = {
    "moscow": [37.32, 55.57, 37.95, 55.92],
    "saint_petersburg": [30.08, 59.83, 30.55, 60.09],
    "berlin": [13.08, 52.34, 13.76, 52.68],
    "paris": [2.22, 48.81, 2.47, 48.90],
    "london": [-0.35, 51.38, 0.15, 51.60],
}

def as_bbox(values: list) -> dict:
    """[min_lon, min_lat, max_lon, max_lat] -> the bbox object the API expects."""
    keys = ("min_lon", "min_lat", "max_lon", "max_lat")
    return dict(zip(keys, values))

def export_city(name: str, bbox: list, fmt: str = "geojson") -> str:
    """Create an export task and wait for the result."""
    resp = requests.post(f"{API_URL}/render", headers=HEADERS, json={
        "bbox": as_bbox(bbox),
        "formats": [fmt],
        "map_style": "standard",
        "detail_level": 4
    })
    resp.raise_for_status()
    task_id = resp.json()["task_id"]
    print(f"[{name}] Task created: {task_id}")

    for _ in range(120):
        status = requests.get(f"{API_URL}/status/{task_id}", headers=HEADERS).json()
        if status["status"] == "completed":
            break
        if status["status"] == "failed":
            raise RuntimeError(f"Export failed: {status.get('error')}")
        time.sleep(5)

    output = Path(f"output/{name}.{fmt}")
    output.parent.mkdir(exist_ok=True)
    dl = requests.get(f"{API_URL}/download/{task_id}/{fmt}", headers=HEADERS)
    output.write_bytes(dl.content)
    print(f"[{name}] Saved to {output} ({len(dl.content)} bytes)")
    return str(output)

for city, bbox in cities.items():
    export_city(city, bbox, "geojson")

Batch Export: one pack, one archive

Batch export works differently from a loop of single tasks: you submit a list of areas in one request, get one batch_id, and at the end pick up a single archive with all the files.

def batch_export(cities: dict, fmt: str = "shp", archive: str = "zip"):
    """Batch export via the batch endpoint: a pack -> an archive."""
    items = [
        {"bbox": as_bbox(bbox), "formats": [fmt]}
        for bbox in cities.values()
    ]

    resp = requests.post(f"{API_URL}/batch/export", headers=HEADERS, json={
        "items": items,
        "archive_format": archive
    })
    resp.raise_for_status()
    batch = resp.json()
    batch_id = batch["batch_id"]
    print(f"Batch created: {batch_id}, {batch['items_total']} items")

    for _ in range(240):  # up to 20 minutes
        st = requests.get(f"{API_URL}/batch/status/{batch_id}", headers=HEADERS).json()
        print(f"{st['items_completed']}/{st['items_total']} done, "
              f"{st['items_failed']} failed")
        if st["status"] in ("completed", "failed"):
            break
        time.sleep(5)

    suffix = "zip" if archive == "zip" else "tar.gz"
    path = Path(f"output/batch_{batch_id}.{suffix}")
    path.parent.mkdir(exist_ok=True)
    dl = requests.get(f"{API_URL}/batch/download/{batch_id}", headers=HEADERS)
    path.write_bytes(dl.content)
    print(f"Saved to {path} ({len(dl.content)} bytes)")
    return path

batch_export(cities, "shp")

In the status response, results carries one entry per area: item_index (its position in your original list), status, files and error — match the output back to your city list by item_index.

Parallel Polling for Multiple Tasks

For long-running tasks (large areas, complex formats) the easiest way to track progress is parallel polling of the REST endpoint /api/status/{task_id}. That is the path for a set of independent /api/render tasks; a pack created via /api/batch/export has its own aggregate status at /api/batch/status/{batch_id}. The WebSocket endpoint /ws/task/{task_id} exists only as a stub (it immediately closes with code 1001) and is reserved for a future implementation via Celery -> Redis Pub/Sub — do not rely on it in production scripts.

Basic single-task polling pattern:

import requests
import time

def poll_until_done(task_id: str, interval: int = 5, timeout: int = 600) -> dict:
    """Poll GET /api/status/{task_id} until completed/failed."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        st = requests.get(f"{API_URL}/status/{task_id}", headers=HEADERS).json()
        progress = st.get("progress", 0)
        step = st.get("step", "")
        print(f"[{task_id[:8]}] {st['status']} {progress}% — {step}")
        if st["status"] == "completed":
            return st
        if st["status"] == "failed":
            raise RuntimeError(f"Export failed: {st.get('error')}")
        time.sleep(interval)
    raise TimeoutError(f"Task {task_id} did not finish in {timeout}s")

Parallel batch monitoring via ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor, as_completed

def poll_batch(task_ids: list[str], max_workers: int = 5) -> dict:
    """Poll a list of task_ids in parallel; return {task_id: final_status}."""
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(poll_until_done, tid): tid for tid in task_ids}
        for fut in as_completed(futures):
            tid = futures[fut]
            try:
                results[tid] = fut.result()
            except Exception as e:
                results[tid] = {"status": "failed", "error": str(e)}
    return results

# Usage: collect task_ids from several /api/render calls and wait for all in parallel
# results = poll_batch(task_ids, max_workers=5)

On HTTP 429 (rate limit), increase interval or lower max_workers — the sliding window rate limiter counts every request made with your API key, including /api/status/{task_id} calls.

Rate Limiting

The API uses sliding window rate limiting. When limits are exceeded, HTTP 429 is returned with a Retry-After header:

resp = requests.post(f"{API_URL}/render", headers=HEADERS, json=payload)
if resp.status_code == 429:
    wait = int(resp.headers.get("Retry-After", 60))
    print(f"Rate limited. Waiting {wait}s...")
    time.sleep(wait)

Best Formats for Automation

Not all formats are equally suited for batch scenarios:

For batch operations, we recommend GeoJSON (universal) or FlatGeobuf (fast and compact).

Example: Daily Data Updates

import schedule

def daily_update():
    """Update data for key regions."""
    regions = {
        "moscow_roads": {"bbox": [37.3, 55.5, 37.9, 55.9], "format": "geojson"},
        "spb_buildings": {"bbox": [30.1, 59.8, 30.5, 60.1], "format": "shp"},
    }
    for name, params in regions.items():
        try:
            export_city(name, params["bbox"], params["format"])
        except Exception as e:
            print(f"Error exporting {name}: {e}")

schedule.every().day.at("03:00").do(daily_update)

while True:
    schedule.run_pending()
    time.sleep(60)

Automating exports via the API saves hours of manual work. The batch endpoint, parallel status polling, and a single archive per pack let you process hundreds of areas in minutes.