API Docs¶

Raw Data API Python Client¶

Client¶

Get OSM data for a specified area.

This is a convenience wrapper around RawDataClient.get_osm_data.

Parameters:

Name	Type	Description	Default
`geometry`	`dict[str, Any] \| str`	GeoJSON geometry object or string	required
`output_options`	`RawDataOutputOptions`	Options for controlling output behavior	`default()`
`**kwargs`		Additional parameters for customizing the request - fileName: Name for the export file (default: "osm_export") - outputType: Format of the output (default: "geojson") - bindZip: Whether to retrieve results as a zip file (default: False) - filters: Dictionary of filters to apply - geometryType: List of geometry types to include	`{}`

Returns:

Type	Description
`RawDataResult`	Object containing metadata, plus a filepath or data.

Raises:

Type	Description
`ValidationError`	If inputs are invalid
`APIRequestError`	If the API request fails
`TaskPollingError`	If polling the task status fails
`DownloadError`	If downloading data fails

Source code in osm_data_client/client.py

async def get_osm_data(
    geometry: dict[str, Any] | str,
    output_options: RawDataOutputOptions = RawDataOutputOptions.default(),
    **kwargs,
) -> RawDataResult:
    """
    Get OSM data for a specified area.

    This is a convenience wrapper around RawDataClient.get_osm_data.

    Args:
        geometry: GeoJSON geometry object or string
        output_options: Options for controlling output behavior
        **kwargs: Additional parameters for customizing the request
            - fileName: Name for the export file (default: "osm_export")
            - outputType: Format of the output (default: "geojson")
            - bindZip: Whether to retrieve results as a zip file (default: False)
            - filters: Dictionary of filters to apply
            - geometryType: List of geometry types to include

    Returns:
        Object containing metadata, plus a filepath or data.

    Raises:
        ValidationError: If inputs are invalid
        APIRequestError: If the API request fails
        TaskPollingError: If polling the task status fails
        DownloadError: If downloading data fails
    """
    config = RawDataClientConfig.default()
    client = RawDataClient(config=config)
    return await client.get_osm_data(geometry, output_options, **kwargs)

options: show_source: false heading_level: 3

Client for fetching OSM data via the HOTOSM Raw Data API.

This client provides a high-level interface for requesting and downloading OpenStreetMap data for a specified area with customizable filters.

Source code in osm_data_client/client.py

class RawDataClient:
    """
    Client for fetching OSM data via the HOTOSM Raw Data API.

    This client provides a high-level interface for requesting and downloading
    OpenStreetMap data for a specified area with customizable filters.
    """

    def __init__(self, config: RawDataClientConfig = RawDataClientConfig.default()):
        """
        Initialize the client.

        Args:
            config: Configuration for the client
        """
        self.config = config
        self.api = RawDataAPI(config)

    async def get_osm_data(
        self,
        geometry: dict[str, Any] | str,
        output_options: RawDataOutputOptions = RawDataOutputOptions.default(),
        **kwargs,
    ) -> RawDataResult | dict:
        """
        Get OSM data for a specified area.

        Args:
            geometry: GeoJSON geometry object or string
            output_options: Options for controlling output behavior
            **kwargs: Additional parameters for customizing the request
                - fileName: Name for the export file (default: "osm_export")
                - outputType: Format of the output (default: "geojson")
                - bindZip: Whether to retrieve results as a zip file (default: False)
                - filters: Dictionary of filters to apply
                - geometryType: List of geometry types to include

        Returns:
            Object containing metadata, plus a filepath or data.

        Raises:
            ValidationError: If inputs are invalid
            APIRequestError: If the API request fails
            TaskPollingError: If polling the task status fails
            DownloadError: If downloading data fails

        Examples:
            >>> data_path = await get_osm_data(
            ...     {"type": "Polygon", "coordinates": [...]},
            ...     fileName="my_buildings",
            ...     outputType="geojson",
            ...     filters={"tags": {"all_geometry": {"building": []}}}
            ... )
        """
        # Validate inputs
        geometry_input = GeometryInput.from_input(geometry)
        params = RequestParams.from_kwargs(**kwargs)

        if (
            params.output_type == "geojson"
            and params.bind_zip
            and not output_options.download_file
        ):
            log.info("Requesting OSM geojson data snapshot")
            return await self.api.request_plain_geojson_snapshot(geometry_input, params)

        # Request snapshot
        log.info("Requesting OSM data snapshot for %s", params.file_name)
        task_response = await self.api.request_snapshot(geometry_input, params)

        # Get task link for polling
        task_link = task_response.get("track_link")
        if not task_link:
            raise TaskPollingError("No task link found in API response")

        # Poll for task completion
        result = await self.api.poll_task_status(task_link)

        if result["status"] != "SUCCESS":
            # Handle failure
            error_msg = f"Task failed with status: {result['status']}"
            if result.get("result", {}).get("error_msg"):
                error_msg += f" - {result['result']['error_msg']}"
            raise DownloadError(error_msg)

        # Create metadata from the result
        metadata = RawDataApiMetadata.from_api_result(result, params)
        log.debug("Data metadata: %s", metadata)

        if output_options.download_file:
            # Download the data
            return await self.api.download_to_disk(metadata, output_options)

        # Skip download and return directly
        return RawDataResult(metadata=metadata, data=result.get("result", {}))

`init(config=RawDataClientConfig.default())` ¶

Initialize the client.

Parameters:

Name	Type	Description	Default
`config`	`RawDataClientConfig`	Configuration for the client	`default()`

Source code in osm_data_client/client.py

def __init__(self, config: RawDataClientConfig = RawDataClientConfig.default()):
    """
    Initialize the client.

    Args:
        config: Configuration for the client
    """
    self.config = config
    self.api = RawDataAPI(config)

`get_osm_data(geometry, output_options=RawDataOutputOptions.default(), **kwargs)` `async` ¶

Get OSM data for a specified area.

Parameters:

Name	Type	Description	Default
`geometry`	`dict[str, Any] \| str`	GeoJSON geometry object or string	required
`output_options`	`RawDataOutputOptions`	Options for controlling output behavior	`default()`
`**kwargs`		Additional parameters for customizing the request - fileName: Name for the export file (default: "osm_export") - outputType: Format of the output (default: "geojson") - bindZip: Whether to retrieve results as a zip file (default: False) - filters: Dictionary of filters to apply - geometryType: List of geometry types to include	`{}`

Returns:

Type	Description
`RawDataResult \| dict`	Object containing metadata, plus a filepath or data.

Raises:

Type	Description
`ValidationError`	If inputs are invalid
`APIRequestError`	If the API request fails
`TaskPollingError`	If polling the task status fails
`DownloadError`	If downloading data fails

Examples:

>>> data_path = await get_osm_data(
...     {"type": "Polygon", "coordinates": [...]},
...     fileName="my_buildings",
...     outputType="geojson",
...     filters={"tags": {"all_geometry": {"building": []}}}
... )

Source code in osm_data_client/client.py

async def get_osm_data(
    self,
    geometry: dict[str, Any] | str,
    output_options: RawDataOutputOptions = RawDataOutputOptions.default(),
    **kwargs,
) -> RawDataResult | dict:
    """
    Get OSM data for a specified area.

    Args:
        geometry: GeoJSON geometry object or string
        output_options: Options for controlling output behavior
        **kwargs: Additional parameters for customizing the request
            - fileName: Name for the export file (default: "osm_export")
            - outputType: Format of the output (default: "geojson")
            - bindZip: Whether to retrieve results as a zip file (default: False)
            - filters: Dictionary of filters to apply
            - geometryType: List of geometry types to include

    Returns:
        Object containing metadata, plus a filepath or data.

    Raises:
        ValidationError: If inputs are invalid
        APIRequestError: If the API request fails
        TaskPollingError: If polling the task status fails
        DownloadError: If downloading data fails

    Examples:
        >>> data_path = await get_osm_data(
        ...     {"type": "Polygon", "coordinates": [...]},
        ...     fileName="my_buildings",
        ...     outputType="geojson",
        ...     filters={"tags": {"all_geometry": {"building": []}}}
        ... )
    """
    # Validate inputs
    geometry_input = GeometryInput.from_input(geometry)
    params = RequestParams.from_kwargs(**kwargs)

    if (
        params.output_type == "geojson"
        and params.bind_zip
        and not output_options.download_file
    ):
        log.info("Requesting OSM geojson data snapshot")
        return await self.api.request_plain_geojson_snapshot(geometry_input, params)

    # Request snapshot
    log.info("Requesting OSM data snapshot for %s", params.file_name)
    task_response = await self.api.request_snapshot(geometry_input, params)

    # Get task link for polling
    task_link = task_response.get("track_link")
    if not task_link:
        raise TaskPollingError("No task link found in API response")

    # Poll for task completion
    result = await self.api.poll_task_status(task_link)

    if result["status"] != "SUCCESS":
        # Handle failure
        error_msg = f"Task failed with status: {result['status']}"
        if result.get("result", {}).get("error_msg"):
            error_msg += f" - {result['result']['error_msg']}"
        raise DownloadError(error_msg)

    # Create metadata from the result
    metadata = RawDataApiMetadata.from_api_result(result, params)
    log.debug("Data metadata: %s", metadata)

    if output_options.download_file:
        # Download the data
        return await self.api.download_to_disk(metadata, output_options)

    # Skip download and return directly
    return RawDataResult(metadata=metadata, data=result.get("result", {}))

options: show_source: false heading_level: 3

Client for the HOTOSM Raw Data API.

Source code in osm_data_client/client.py

class RawDataAPI:
    """Client for the HOTOSM Raw Data API."""

    def __init__(self, config: RawDataClientConfig = RawDataClientConfig.default()):
        """
        Initialize the API client.

        Args:
            config: Configuration for the client
        """
        self.config = config
        self.headers = {
            "accept": "application/json",
            "Content-Type": "application/json",
            "Referer": "raw-data-client-py",
        }

        if config.access_token:
            self.headers["Authorization"] = f"Bearer {config.access_token}"
            log.debug("Using access token for authentication")

    async def request_snapshot(
        self, geometry: GeometryInput, params: RequestParams
    ) -> dict[str, Any]:
        """
        Request a snapshot of OSM data.

        Args:
            geometry: Validated GeoJSON geometry object
            params: Validated request parameters

        Returns:
            API response with task tracking information

        Raises:
            APIRequestError: If the API request fails
        """
        payload = {
            **params.to_api_params(),
            "geometry": geometry.to_dict(),
        }

        log.debug("Requesting snapshot with params: %s", json.dumps(payload))

        async with ClientSession() as session:
            try:
                async with session.post(
                    f"{self.config.base_api_url}/snapshot/",
                    data=json.dumps(payload),
                    headers=self.headers,
                ) as response:
                    response_data = await response.json()
                    if response.status >= 400:
                        log.error(
                            "API request failed with status %d: %s",
                            response.status,
                            response_data,
                        )
                        raise APIRequestError(response.status, response_data)

                    # Log queue information if available
                    if "queue" in response_data:
                        queue_position = response_data.get("queue", 0)
                        if queue_position > 0:
                            log.info("Request queued at position %d", queue_position)

                    log.debug("Snapshot request successful: %s", response_data)
                    return response_data
            except ClientResponseError as ex:
                log.error("API client error: %s", str(ex))
                raise APIRequestError(ex.status, {}, str(ex)) from ex
            except Exception as ex:
                log.error("Unexpected error in API request: %s", str(ex))
                raise APIRequestError(0, {}, str(ex)) from ex

    async def request_plain_geojson_snapshot(
        self, geometry: GeometryInput, params: RequestParams
    ) -> dict[str, Any]:
        """
        Request a snapshot of OSM geojson data.

        Args:
            geometry: Validated GeoJSON geometry object
            params: Validated request parameters

        Returns:
            Plain geojson of osm features

        Raises:
            APIRequestError: If the API request fails
        """
        payload = {
            **params.to_api_params(),
            "geometry": geometry.to_dict(),
        }

        log.debug("Requesting snapshot with params: %s", json.dumps(payload))

        async with ClientSession() as session:
            try:
                async with session.post(
                    f"{self.config.base_api_url}/snapshot/plain/",
                    json=payload,
                    headers=self.headers,
                ) as response:
                    response_data = await response.json()
                    if response.status >= 400:
                        log.error(
                            "API request failed with status %d: %s",
                            response.status,
                            response_data,
                        )
                        raise APIRequestError(response.status, response_data)

                    # Log queue information if available
                    if "queue" in response_data:
                        queue_position = response_data.get("queue", 0)
                        if queue_position > 0:
                            log.info("Request queued at position %d", queue_position)

                    log.debug("Snapshot request successful: %s", response_data)
                    return response_data
            except ClientResponseError as ex:
                log.error("API client error: %s", str(ex))
                raise APIRequestError(ex.status, {}, str(ex)) from ex
            except Exception as ex:
                log.error("Unexpected error in API request: %s", str(ex))
                raise APIRequestError(0, {}, str(ex)) from ex

    async def poll_task_status(
        self, task_link: str, polling_interval: int = 2
    ) -> dict[str, Any]:
        """
        Poll the API to check task status until completion.

        Args:
            task_link: Task tracking URL
            polling_interval: Seconds between polling attempts

        Returns:
            Task status details

        Raises:
            TaskPollingError: If polling fails
        """
        log.info("Starting task polling: %s", task_link)

        # Track previous status to log changes
        previous_status = None

        async with ClientSession() as session:
            while True:
                try:
                    async with session.get(
                        url=f"{self.config.base_api_url}{task_link}",
                        headers=self.headers,
                    ) as response:
                        if response.status >= 400:
                            response_data = await response.json()
                            log.error(
                                "Polling failed with status %d: %s",
                                response.status,
                                response_data,
                            )
                            raise TaskPollingError(
                                f"Polling failed with status {response.status}: {response_data}"
                            )

                        result = await response.json()
                        current_status = result.get("status")

                        # Log status changes
                        if current_status != previous_status:
                            log.info("Task status: %s", current_status)
                            previous_status = current_status

                        if current_status in ["SUCCESS", "FAILED"]:
                            if current_status == "FAILED":
                                error_msg = result.get("result", {}).get(
                                    "error_msg", "Unknown error"
                                )
                                log.error("Task failed: %s", error_msg)
                            else:
                                log.info("Task completed successfully")
                            return result

                        log.debug(
                            "Task still processing, waiting %d seconds",
                            polling_interval,
                        )
                        await asyncio.sleep(polling_interval)
                except TaskPollingError:
                    raise
                except Exception as ex:
                    log.error("Error polling task status: %s", str(ex))
                    raise TaskPollingError(
                        f"Error polling task status: {str(ex)}"
                    ) from ex

    async def download_to_disk(
        self,
        data: RawDataApiMetadata,
        options: RawDataOutputOptions = RawDataOutputOptions.default(),
    ) -> RawDataResult:
        """
        Stream data from API to disk

        Args:
            data: Metadata containing download information
            options: Options for controlling extraction behavior

        Returns:
            RawDataResult with information about the downloaded file

        Raises:
            DownloadError: If downloading or processing fails
        """
        processor = OutputProcessor(self.config, options)
        file_path = processor.get_output_path(data)

        file_path.parent.mkdir(parents=True, exist_ok=True)
        log.info("Downloading data to %s (%s bytes)", file_path, data.size_bytes)

        try:
            async with ClientSession() as session:
                async with session.get(
                    data.download_url, headers=self.headers
                ) as response:
                    if response.status >= 400:
                        log.error("Download failed with status %d", response.status)
                        raise DownloadError(
                            f"Download failed with status {response.status}"
                        )

                    with open(file_path, "wb") as f:
                        log.debug("Streaming file contents using 1MB chunks")
                        downloaded_bytes = 0
                        async for chunk in response.content.iter_chunked(
                            1024 * 1024
                        ):  # 1MB chunks
                            f.write(chunk)
                            downloaded_bytes += len(chunk)
                            if (
                                data.size_bytes > 10 * 1024 * 1024
                                and downloaded_bytes % (10 * 1024 * 1024) == 0
                            ):
                                progress = (downloaded_bytes / data.size_bytes) * 100
                                log.info(
                                    "Download progress: %.1f%% (%d/%d bytes)",
                                    progress,
                                    downloaded_bytes,
                                    data.size_bytes,
                                )

                    log.info("Download complete: %s", file_path)

                    return await processor.process_download(file_path, data)

        except Exception as ex:
            log.error("Error downloading data: %s", str(ex))
            raise DownloadError(f"Error downloading data: {str(ex)}") from ex

`init(config=RawDataClientConfig.default())` ¶

Initialize the API client.

Parameters:

Name	Type	Description	Default
`config`	`RawDataClientConfig`	Configuration for the client	`default()`

Source code in osm_data_client/client.py

def __init__(self, config: RawDataClientConfig = RawDataClientConfig.default()):
    """
    Initialize the API client.

    Args:
        config: Configuration for the client
    """
    self.config = config
    self.headers = {
        "accept": "application/json",
        "Content-Type": "application/json",
        "Referer": "raw-data-client-py",
    }

    if config.access_token:
        self.headers["Authorization"] = f"Bearer {config.access_token}"
        log.debug("Using access token for authentication")

`download_to_disk(data, options=RawDataOutputOptions.default())` `async` ¶

Stream data from API to disk

Parameters:

Name	Type	Description	Default
`data`	`RawDataApiMetadata`	Metadata containing download information	required
`options`	`RawDataOutputOptions`	Options for controlling extraction behavior	`default()`

Returns:

Type	Description
`RawDataResult`	RawDataResult with information about the downloaded file

Raises:

Type	Description
`DownloadError`	If downloading or processing fails

Source code in osm_data_client/client.py

async def download_to_disk(
    self,
    data: RawDataApiMetadata,
    options: RawDataOutputOptions = RawDataOutputOptions.default(),
) -> RawDataResult:
    """
    Stream data from API to disk

    Args:
        data: Metadata containing download information
        options: Options for controlling extraction behavior

    Returns:
        RawDataResult with information about the downloaded file

    Raises:
        DownloadError: If downloading or processing fails
    """
    processor = OutputProcessor(self.config, options)
    file_path = processor.get_output_path(data)

    file_path.parent.mkdir(parents=True, exist_ok=True)
    log.info("Downloading data to %s (%s bytes)", file_path, data.size_bytes)

    try:
        async with ClientSession() as session:
            async with session.get(
                data.download_url, headers=self.headers
            ) as response:
                if response.status >= 400:
                    log.error("Download failed with status %d", response.status)
                    raise DownloadError(
                        f"Download failed with status {response.status}"
                    )

                with open(file_path, "wb") as f:
                    log.debug("Streaming file contents using 1MB chunks")
                    downloaded_bytes = 0
                    async for chunk in response.content.iter_chunked(
                        1024 * 1024
                    ):  # 1MB chunks
                        f.write(chunk)
                        downloaded_bytes += len(chunk)
                        if (
                            data.size_bytes > 10 * 1024 * 1024
                            and downloaded_bytes % (10 * 1024 * 1024) == 0
                        ):
                            progress = (downloaded_bytes / data.size_bytes) * 100
                            log.info(
                                "Download progress: %.1f%% (%d/%d bytes)",
                                progress,
                                downloaded_bytes,
                                data.size_bytes,
                            )

                log.info("Download complete: %s", file_path)

                return await processor.process_download(file_path, data)

    except Exception as ex:
        log.error("Error downloading data: %s", str(ex))
        raise DownloadError(f"Error downloading data: {str(ex)}") from ex

`poll_task_status(task_link, polling_interval=2)` `async` ¶

Poll the API to check task status until completion.

Parameters:

Name	Type	Description	Default
`task_link`	`str`	Task tracking URL	required
`polling_interval`	`int`	Seconds between polling attempts	`2`

Returns:

Type	Description
`dict[str, Any]`	Task status details

Raises:

Type	Description
`TaskPollingError`	If polling fails

Source code in osm_data_client/client.py

async def poll_task_status(
    self, task_link: str, polling_interval: int = 2
) -> dict[str, Any]:
    """
    Poll the API to check task status until completion.

    Args:
        task_link: Task tracking URL
        polling_interval: Seconds between polling attempts

    Returns:
        Task status details

    Raises:
        TaskPollingError: If polling fails
    """
    log.info("Starting task polling: %s", task_link)

    # Track previous status to log changes
    previous_status = None

    async with ClientSession() as session:
        while True:
            try:
                async with session.get(
                    url=f"{self.config.base_api_url}{task_link}",
                    headers=self.headers,
                ) as response:
                    if response.status >= 400:
                        response_data = await response.json()
                        log.error(
                            "Polling failed with status %d: %s",
                            response.status,
                            response_data,
                        )
                        raise TaskPollingError(
                            f"Polling failed with status {response.status}: {response_data}"
                        )

                    result = await response.json()
                    current_status = result.get("status")

                    # Log status changes
                    if current_status != previous_status:
                        log.info("Task status: %s", current_status)
                        previous_status = current_status

                    if current_status in ["SUCCESS", "FAILED"]:
                        if current_status == "FAILED":
                            error_msg = result.get("result", {}).get(
                                "error_msg", "Unknown error"
                            )
                            log.error("Task failed: %s", error_msg)
                        else:
                            log.info("Task completed successfully")
                        return result

                    log.debug(
                        "Task still processing, waiting %d seconds",
                        polling_interval,
                    )
                    await asyncio.sleep(polling_interval)
            except TaskPollingError:
                raise
            except Exception as ex:
                log.error("Error polling task status: %s", str(ex))
                raise TaskPollingError(
                    f"Error polling task status: {str(ex)}"
                ) from ex

`request_plain_geojson_snapshot(geometry, params)` `async` ¶

Request a snapshot of OSM geojson data.

Parameters:

Name	Type	Description	Default
`geometry`	`GeometryInput`	Validated GeoJSON geometry object	required
`params`	`RequestParams`	Validated request parameters	required

Returns:

Type	Description
`dict[str, Any]`	Plain geojson of osm features

Raises:

Type	Description
`APIRequestError`	If the API request fails

Source code in osm_data_client/client.py

async def request_plain_geojson_snapshot(
    self, geometry: GeometryInput, params: RequestParams
) -> dict[str, Any]:
    """
    Request a snapshot of OSM geojson data.

    Args:
        geometry: Validated GeoJSON geometry object
        params: Validated request parameters

    Returns:
        Plain geojson of osm features

    Raises:
        APIRequestError: If the API request fails
    """
    payload = {
        **params.to_api_params(),
        "geometry": geometry.to_dict(),
    }

    log.debug("Requesting snapshot with params: %s", json.dumps(payload))

    async with ClientSession() as session:
        try:
            async with session.post(
                f"{self.config.base_api_url}/snapshot/plain/",
                json=payload,
                headers=self.headers,
            ) as response:
                response_data = await response.json()
                if response.status >= 400:
                    log.error(
                        "API request failed with status %d: %s",
                        response.status,
                        response_data,
                    )
                    raise APIRequestError(response.status, response_data)

                # Log queue information if available
                if "queue" in response_data:
                    queue_position = response_data.get("queue", 0)
                    if queue_position > 0:
                        log.info("Request queued at position %d", queue_position)

                log.debug("Snapshot request successful: %s", response_data)
                return response_data
        except ClientResponseError as ex:
            log.error("API client error: %s", str(ex))
            raise APIRequestError(ex.status, {}, str(ex)) from ex
        except Exception as ex:
            log.error("Unexpected error in API request: %s", str(ex))
            raise APIRequestError(0, {}, str(ex)) from ex

`request_snapshot(geometry, params)` `async` ¶

Request a snapshot of OSM data.

Parameters:

Name	Type	Description	Default
`geometry`	`GeometryInput`	Validated GeoJSON geometry object	required
`params`	`RequestParams`	Validated request parameters	required

Returns:

Type	Description
`dict[str, Any]`	API response with task tracking information

Raises:

Type	Description
`APIRequestError`	If the API request fails

Source code in osm_data_client/client.py

async def request_snapshot(
    self, geometry: GeometryInput, params: RequestParams
) -> dict[str, Any]:
    """
    Request a snapshot of OSM data.

    Args:
        geometry: Validated GeoJSON geometry object
        params: Validated request parameters

    Returns:
        API response with task tracking information

    Raises:
        APIRequestError: If the API request fails
    """
    payload = {
        **params.to_api_params(),
        "geometry": geometry.to_dict(),
    }

    log.debug("Requesting snapshot with params: %s", json.dumps(payload))

    async with ClientSession() as session:
        try:
            async with session.post(
                f"{self.config.base_api_url}/snapshot/",
                data=json.dumps(payload),
                headers=self.headers,
            ) as response:
                response_data = await response.json()
                if response.status >= 400:
                    log.error(
                        "API request failed with status %d: %s",
                        response.status,
                        response_data,
                    )
                    raise APIRequestError(response.status, response_data)

                # Log queue information if available
                if "queue" in response_data:
                    queue_position = response_data.get("queue", 0)
                    if queue_position > 0:
                        log.info("Request queued at position %d", queue_position)

                log.debug("Snapshot request successful: %s", response_data)
                return response_data
        except ClientResponseError as ex:
            log.error("API client error: %s", str(ex))
            raise APIRequestError(ex.status, {}, str(ex)) from ex
        except Exception as ex:
            log.error("Unexpected error in API request: %s", str(ex))
            raise APIRequestError(0, {}, str(ex)) from ex

options: show_source: false heading_level: 3

Results¶

Result object containing processed file path and associated metadata.

Attributes:

Name	Type	Description
`path`	`Optional[Path]`	Path to the final processed file or directory
`data`	`Optional[dict]`	dict representation of data
`metadata`	`RawDataApiMetadata`	Original metadata from the API response
`extracted`	`bool`	Whether the file was extracted from an archive
`original_path`	`Optional[Path]`	Path to the original downloaded file (if different from path)
`extracted_files`	`Optional[list[Path]]`	List of files that were extracted (if applicable)

Source code in osm_data_client/processing.py

@dataclass
class RawDataResult:
    """
    Result object containing processed file path and associated metadata.

    Attributes:
        path: Path to the final processed file or directory
        data: dict representation of data
        metadata: Original metadata from the API response
        extracted: Whether the file was extracted from an archive
        original_path: Path to the original downloaded file (if different from path)
        extracted_files: List of files that were extracted (if applicable)
    """

    metadata: RawDataApiMetadata
    path: Optional[Path] = None
    data: Optional[dict] = None
    extracted: bool = False
    original_path: Optional[Path] = None
    extracted_files: Optional[list[Path]] = None

    def exists(self) -> bool:
        """Check if the result file or directory exists."""
        if not self.path:
            return False
        return self.path.exists()

    def suffix(self) -> str:
        """Get file type suffix, if path exists."""
        if not self.path:
            return ""
        return self.path.suffix

    def __str__(self) -> str:
        """Return string representation of the result."""
        if not self.path:
            return ""
        return str(self.path)

`str()` ¶

Return string representation of the result.

Source code in osm_data_client/processing.py

def __str__(self) -> str:
    """Return string representation of the result."""
    if not self.path:
        return ""
    return str(self.path)

`exists()` ¶

Check if the result file or directory exists.

Source code in osm_data_client/processing.py

def exists(self) -> bool:
    """Check if the result file or directory exists."""
    if not self.path:
        return False
    return self.path.exists()

`suffix()` ¶

Get file type suffix, if path exists.

Source code in osm_data_client/processing.py

def suffix(self) -> str:
    """Get file type suffix, if path exists."""
    if not self.path:
        return ""
    return self.path.suffix

options: show_source: false heading_level: 3

Models¶

Validated geometry input for OSM API requests.

Source code in osm_data_client/models.py

@dataclass
class GeometryInput:
    """Validated geometry input for OSM API requests."""

    type: str
    coordinates: list[Any]

    @classmethod
    def from_input(cls, geometry: dict[str, Any] | str) -> "GeometryInput":
        """
        Create a GeometryInput from either a dictionary or a JSON string.

        Args:
            geometry: GeoJSON geometry object or string

        Returns:
            Validated GeometryInput object

        Raises:
            ValidationError: If geometry is invalid
        """
        from .exceptions import ValidationError

        if isinstance(geometry, str):
            try:
                geometry_dict = json.loads(geometry)
                log.debug("Parsed geometry from JSON string")
            except json.JSONDecodeError:
                log.error("Failed to parse geometry JSON string")
                raise ValidationError("Invalid GeoJSON string")
        else:
            geometry_dict = geometry

        if (
            geometry_dict.get("type") == "FeatureCollection"
            and "features" in geometry_dict
        ):
            log.debug("Converting FeatureCollection to Geometry")
            if geometry_dict["features"]:
                feature = geometry_dict["features"][0]
                if "geometry" in feature:
                    geometry_dict = feature["geometry"]

        if "type" not in geometry_dict:
            log.error("Geometry missing 'type' field")
            raise ValidationError("Geometry must have a 'type' field")

        if "coordinates" not in geometry_dict:
            log.error("Geometry missing 'coordinates' field")
            raise ValidationError("Geometry must have a 'coordinates' field")

        valid_types = ["Polygon", "MultiPolygon"]
        if geometry_dict["type"] not in valid_types:
            log.error("Invalid geometry type: %s", geometry_dict["type"])
            raise ValidationError(f"Geometry type must be one of {valid_types}")

        # Check CRS if present (basic validation)
        if "crs" in geometry_dict:
            crs = geometry_dict.get("crs", {}).get("properties", {}).get("name")
            valid_crs = [
                "urn:ogc:def:crs:OGC:1.3:CRS84",
                "urn:ogc:def:crs:EPSG::4326",
                "WGS 84",
            ]
            if crs and crs not in valid_crs:
                log.warning("Unsupported CRS: %s. Raw Data API requires EPSG:4326", crs)
                raise ValidationError(
                    "Unsupported coordinate system. Raw Data API requires "
                    "GeoJSON in WGS84 (EPSG:4326) standard."
                )

        # Basic coordinate validation for first coordinate
        try:
            first_coord = cls._get_first_coordinate(geometry_dict["coordinates"])
            if first_coord and not cls._is_valid_coordinate(first_coord):
                log.error("Invalid coordinates: %s", first_coord)
                raise ValidationError(
                    "Coordinates appear to be invalid. Should be longitude/latitude "
                    "in the range of -180 to 180 and -90 to 90 respectively."
                )
        except (IndexError, TypeError):
            log.warning("Could not validate coordinates format")

        log.debug("Validated geometry of type %s", geometry_dict["type"])
        return cls(type=geometry_dict["type"], coordinates=geometry_dict["coordinates"])

    @staticmethod
    def _get_first_coordinate(coordinates):
        """Extract the first coordinate from nested arrays."""
        # For Polygon or MultiPolygon in GeoJSON, we need to navigate the nested structure
        # A Polygon coordinate is [[[x1,y1], [x2,y2], ...]]
        # A MultiPolygon coordinate is [[[[x1,y1], [x2,y2], ...]], ...]

        result = coordinates
        while isinstance(result, list) and isinstance(result[0], list):
            result = result[0]

        return result

    @staticmethod
    def _is_valid_coordinate(coord):
        """Check if a coordinate is valid (within expected range)."""
        if not isinstance(coord, list) or len(coord) < 2:
            return False
        return -180 <= coord[0] <= 180 and -90 <= coord[1] <= 90

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation."""
        return {"type": self.type, "coordinates": self.coordinates}

`from_input(geometry)` `classmethod` ¶

Create a GeometryInput from either a dictionary or a JSON string.

Parameters:

Name	Type	Description	Default
`geometry`	`dict[str, Any] \| str`	GeoJSON geometry object or string	required

Returns:

Type	Description
`GeometryInput`	Validated GeometryInput object

Raises:

Type	Description
`ValidationError`	If geometry is invalid

Source code in osm_data_client/models.py

@classmethod
def from_input(cls, geometry: dict[str, Any] | str) -> "GeometryInput":
    """
    Create a GeometryInput from either a dictionary or a JSON string.

    Args:
        geometry: GeoJSON geometry object or string

    Returns:
        Validated GeometryInput object

    Raises:
        ValidationError: If geometry is invalid
    """
    from .exceptions import ValidationError

    if isinstance(geometry, str):
        try:
            geometry_dict = json.loads(geometry)
            log.debug("Parsed geometry from JSON string")
        except json.JSONDecodeError:
            log.error("Failed to parse geometry JSON string")
            raise ValidationError("Invalid GeoJSON string")
    else:
        geometry_dict = geometry

    if (
        geometry_dict.get("type") == "FeatureCollection"
        and "features" in geometry_dict
    ):
        log.debug("Converting FeatureCollection to Geometry")
        if geometry_dict["features"]:
            feature = geometry_dict["features"][0]
            if "geometry" in feature:
                geometry_dict = feature["geometry"]

    if "type" not in geometry_dict:
        log.error("Geometry missing 'type' field")
        raise ValidationError("Geometry must have a 'type' field")

    if "coordinates" not in geometry_dict:
        log.error("Geometry missing 'coordinates' field")
        raise ValidationError("Geometry must have a 'coordinates' field")

    valid_types = ["Polygon", "MultiPolygon"]
    if geometry_dict["type"] not in valid_types:
        log.error("Invalid geometry type: %s", geometry_dict["type"])
        raise ValidationError(f"Geometry type must be one of {valid_types}")

    # Check CRS if present (basic validation)
    if "crs" in geometry_dict:
        crs = geometry_dict.get("crs", {}).get("properties", {}).get("name")
        valid_crs = [
            "urn:ogc:def:crs:OGC:1.3:CRS84",
            "urn:ogc:def:crs:EPSG::4326",
            "WGS 84",
        ]
        if crs and crs not in valid_crs:
            log.warning("Unsupported CRS: %s. Raw Data API requires EPSG:4326", crs)
            raise ValidationError(
                "Unsupported coordinate system. Raw Data API requires "
                "GeoJSON in WGS84 (EPSG:4326) standard."
            )

    # Basic coordinate validation for first coordinate
    try:
        first_coord = cls._get_first_coordinate(geometry_dict["coordinates"])
        if first_coord and not cls._is_valid_coordinate(first_coord):
            log.error("Invalid coordinates: %s", first_coord)
            raise ValidationError(
                "Coordinates appear to be invalid. Should be longitude/latitude "
                "in the range of -180 to 180 and -90 to 90 respectively."
            )
    except (IndexError, TypeError):
        log.warning("Could not validate coordinates format")

    log.debug("Validated geometry of type %s", geometry_dict["type"])
    return cls(type=geometry_dict["type"], coordinates=geometry_dict["coordinates"])

`to_dict()` ¶

Convert to dictionary representation.

Source code in osm_data_client/models.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary representation."""
    return {"type": self.type, "coordinates": self.coordinates}

options: show_source: false heading_level: 3

Validated parameters for OSM API requests.

Source code in osm_data_client/models.py

@dataclass
class RequestParams:
    """Validated parameters for OSM API requests."""

    file_name: str = "osm_export"
    output_type: str = "geojson"
    bind_zip: bool = True
    centroid: bool = False
    use_st_within: bool = True
    filters: Optional[FilterDict] = None
    geometry_type: Optional[list[str]] = None

    VALID_OUTPUT_TYPES = [
        "geojson",
        "shp",
        "kml",
        "mbtiles",
        "flatgeobuf",
        "csv",
        "geopackage",
        "pgdump",
    ]

    @classmethod
    def from_kwargs(cls, **kwargs) -> "RequestParams":
        """
        Create a RequestParams from keyword arguments.

        Args:
            **kwargs: Keyword arguments for request parameters

        Returns:
            Validated RequestParams object

        Raises:
            ValidationError: If parameters are invalid
        """
        from .exceptions import ValidationError

        # Convert to snake_case internally
        params = {}
        if "fileName" in kwargs:
            params["file_name"] = kwargs.pop("fileName")
        if "outputType" in kwargs:
            params["output_type"] = kwargs.pop("outputType")
        if "geometryType" in kwargs:
            params["geometry_type"] = kwargs.pop("geometryType")
        if "bindZip" in kwargs:
            params["bind_zip"] = kwargs.pop("bindZip")

        params.update(kwargs)

        if "output_type" in params and "bind_zip" in params:
            params["bind_zip"] = RequestParams.validate_bind_zip_compatibility(
                params["output_type"], params["bind_zip"]
            )

        instance = cls(**params)

        if instance.output_type not in cls.VALID_OUTPUT_TYPES:
            log.error("Invalid output type: %s", instance.output_type)
            raise ValidationError(f"outputType must be one of {cls.VALID_OUTPUT_TYPES}")

        return instance

    def to_api_params(self) -> dict[str, Any]:
        """Convert to API parameter dictionary."""
        # Convert to camelCase for API
        params = {
            "fileName": self.file_name,
            "outputType": self.output_type,
            "bindZip": self.bind_zip,
            "centroid": self.centroid,
            "useStWithin": self.use_st_within,
        }

        if self.filters:
            params["filters"] = self.filters

        if self.geometry_type:
            params["geometryType"] = self.geometry_type

        return params

    @staticmethod
    def validate_bind_zip_compatibility(output_type, bind_zip):
        """Validate if the output format is compatible with bindZip=False"""
        streaming_compatible_formats = [
            "geojson",
            "cog",
            "fgb",
        ]  # Cloud Optimized GeoTIFF, FlatGeoBuf

        if not bind_zip and output_type.lower() not in streaming_compatible_formats:
            log.warning(
                f"Format '{output_type}' requires ZIP packaging. "
                f"Automatically setting bindZip=True"
            )
            return True
        return bind_zip

`from_kwargs(**kwargs)` `classmethod` ¶

Create a RequestParams from keyword arguments.

Parameters:

Name	Type	Description	Default
`**kwargs`		Keyword arguments for request parameters	`{}`

Returns:

Type	Description
`RequestParams`	Validated RequestParams object

Raises:

Type	Description
`ValidationError`	If parameters are invalid

Source code in osm_data_client/models.py

@classmethod
def from_kwargs(cls, **kwargs) -> "RequestParams":
    """
    Create a RequestParams from keyword arguments.

    Args:
        **kwargs: Keyword arguments for request parameters

    Returns:
        Validated RequestParams object

    Raises:
        ValidationError: If parameters are invalid
    """
    from .exceptions import ValidationError

    # Convert to snake_case internally
    params = {}
    if "fileName" in kwargs:
        params["file_name"] = kwargs.pop("fileName")
    if "outputType" in kwargs:
        params["output_type"] = kwargs.pop("outputType")
    if "geometryType" in kwargs:
        params["geometry_type"] = kwargs.pop("geometryType")
    if "bindZip" in kwargs:
        params["bind_zip"] = kwargs.pop("bindZip")

    params.update(kwargs)

    if "output_type" in params and "bind_zip" in params:
        params["bind_zip"] = RequestParams.validate_bind_zip_compatibility(
            params["output_type"], params["bind_zip"]
        )

    instance = cls(**params)

    if instance.output_type not in cls.VALID_OUTPUT_TYPES:
        log.error("Invalid output type: %s", instance.output_type)
        raise ValidationError(f"outputType must be one of {cls.VALID_OUTPUT_TYPES}")

    return instance

`to_api_params()` ¶

Convert to API parameter dictionary.

Source code in osm_data_client/models.py

def to_api_params(self) -> dict[str, Any]:
    """Convert to API parameter dictionary."""
    # Convert to camelCase for API
    params = {
        "fileName": self.file_name,
        "outputType": self.output_type,
        "bindZip": self.bind_zip,
        "centroid": self.centroid,
        "useStWithin": self.use_st_within,
    }

    if self.filters:
        params["filters"] = self.filters

    if self.geometry_type:
        params["geometryType"] = self.geometry_type

    return params

`validate_bind_zip_compatibility(output_type, bind_zip)` `staticmethod` ¶

Validate if the output format is compatible with bindZip=False

Source code in osm_data_client/models.py

@staticmethod
def validate_bind_zip_compatibility(output_type, bind_zip):
    """Validate if the output format is compatible with bindZip=False"""
    streaming_compatible_formats = [
        "geojson",
        "cog",
        "fgb",
    ]  # Cloud Optimized GeoTIFF, FlatGeoBuf

    if not bind_zip and output_type.lower() not in streaming_compatible_formats:
        log.warning(
            f"Format '{output_type}' requires ZIP packaging. "
            f"Automatically setting bindZip=True"
        )
        return True
    return bind_zip

options: show_source: false heading_level: 3

Immutable metadata about a dataset

Source code in osm_data_client/models.py

@dataclass(frozen=True)
class RawDataApiMetadata:
    """Immutable metadata about a dataset"""

    task_id: str
    format_ext: str
    timestamp: str
    size_bytes: int
    file_name: str
    download_url: str
    is_zipped: bool
    bbox: Optional[tuple[float, float, float, float]] = None

    @classmethod
    def from_api_result(
        cls, result: dict[str, Any], params: RequestParams
    ) -> "RawDataApiMetadata":
        """
        Create a RawDataApiMetadata from API result and request parameters.

        Args:
            result: API result dictionary from task status
            params: Request parameters used for the API request

        Returns:
            RawDataApiMetadata instance
        """
        task_result = result.get("result", {})
        task_id = result.get("id", "")
        timestamp = task_result.get("response_time", "")
        size_bytes = task_result.get("zip_file_size_bytes", 0)
        download_url = task_result.get("download_url", "")

        bbox = None
        query_area = task_result.get("queryArea", "")
        if query_area and query_area.startswith("bbox[") and query_area.endswith("]"):
            try:
                coords_str = query_area[5:-1]  # Remove "bbox[" and "]"
                coords = [float(x) for x in coords_str.split(",")]
                if len(coords) == 4:
                    bbox = tuple(coords)
                    log.debug("Extracted bbox: %s", bbox)
            except (ValueError, IndexError):
                log.warning("Could not parse bbox from queryArea: %s", query_area)

        return cls(
            is_zipped=params.bind_zip,
            file_name=params.file_name,
            task_id=task_id,
            format_ext=params.output_type,
            timestamp=timestamp,
            size_bytes=size_bytes,
            download_url=download_url,
            bbox=bbox,
        )

    def __str__(self) -> str:
        """Returns a string representation of RawDataApiMetadata for debugging purposes."""
        bbox_str = f"{self.bbox}" if self.bbox else "None"

        return (
            f"RawDataApiMetadata(\n"
            f"  task_id: {self.task_id}\n"
            f"  format_ext: {self.format_ext}\n"
            f"  timestamp: {self.timestamp}\n"
            f"  size_bytes: {self.size_bytes:,} bytes ({self._format_size()})\n"
            f"  file_name: {self.file_name}\n"
            f"  is_zipped: {self.is_zipped}\n"
            f"  bbox: {bbox_str}\n"
            f")"
        )

    def _format_size(self) -> str:
        """Helper method to format size in human-readable form."""
        size = self.size_bytes
        for unit in ["B", "KB", "MB", "GB", "TB"]:
            if size < 1024 or unit == "TB":
                return f"{size:.2f} {unit}"
            size /= 1024

`str()` ¶

Returns a string representation of RawDataApiMetadata for debugging purposes.

Source code in osm_data_client/models.py

def __str__(self) -> str:
    """Returns a string representation of RawDataApiMetadata for debugging purposes."""
    bbox_str = f"{self.bbox}" if self.bbox else "None"

    return (
        f"RawDataApiMetadata(\n"
        f"  task_id: {self.task_id}\n"
        f"  format_ext: {self.format_ext}\n"
        f"  timestamp: {self.timestamp}\n"
        f"  size_bytes: {self.size_bytes:,} bytes ({self._format_size()})\n"
        f"  file_name: {self.file_name}\n"
        f"  is_zipped: {self.is_zipped}\n"
        f"  bbox: {bbox_str}\n"
        f")"
    )

`from_api_result(result, params)` `classmethod` ¶

Create a RawDataApiMetadata from API result and request parameters.

Parameters:

Name	Type	Description	Default
`result`	`dict[str, Any]`	API result dictionary from task status	required
`params`	`RequestParams`	Request parameters used for the API request	required

Returns:

Type	Description
`RawDataApiMetadata`	RawDataApiMetadata instance

Source code in osm_data_client/models.py

@classmethod
def from_api_result(
    cls, result: dict[str, Any], params: RequestParams
) -> "RawDataApiMetadata":
    """
    Create a RawDataApiMetadata from API result and request parameters.

    Args:
        result: API result dictionary from task status
        params: Request parameters used for the API request

    Returns:
        RawDataApiMetadata instance
    """
    task_result = result.get("result", {})
    task_id = result.get("id", "")
    timestamp = task_result.get("response_time", "")
    size_bytes = task_result.get("zip_file_size_bytes", 0)
    download_url = task_result.get("download_url", "")

    bbox = None
    query_area = task_result.get("queryArea", "")
    if query_area and query_area.startswith("bbox[") and query_area.endswith("]"):
        try:
            coords_str = query_area[5:-1]  # Remove "bbox[" and "]"
            coords = [float(x) for x in coords_str.split(",")]
            if len(coords) == 4:
                bbox = tuple(coords)
                log.debug("Extracted bbox: %s", bbox)
        except (ValueError, IndexError):
            log.warning("Could not parse bbox from queryArea: %s", query_area)

    return cls(
        is_zipped=params.bind_zip,
        file_name=params.file_name,
        task_id=task_id,
        format_ext=params.output_type,
        timestamp=timestamp,
        size_bytes=size_bytes,
        download_url=download_url,
        bbox=bbox,
    )

options: show_source: false heading_level: 3

Bases: Enum

Options for controlling extraction behavior of ZIP archives.

Source code in osm_data_client/models.py

class AutoExtractOption(Enum):
    """Options for controlling extraction behavior of ZIP archives."""

    automatic = auto()  # Decide based on format and size
    force_zip = auto()  # Always keep as zip
    force_extract = auto()  # Always extract regardless of size/format

options: show_source: false heading_level: 3

Configuration for Raw Data API client.

Source code in osm_data_client/models.py

@dataclass
class RawDataClientConfig:
    """Configuration for Raw Data API client."""

    access_token: Optional[str] = None
    memory_threshold_mb: int = 50
    base_api_url: str = "https://api-prod.raw-data.hotosm.org/v1"
    output_directory: Path = Path.cwd()

    @property
    def memory_threshold_bytes(self) -> int:
        """Convert memory threshold to bytes."""
        return self.memory_threshold_mb * 1024 * 1024

    @classmethod
    def default(cls) -> "RawDataClientConfig":
        """Create a default configuration."""
        return cls()

`memory_threshold_bytes: int` `property` ¶

Convert memory threshold to bytes.

`default()` `classmethod` ¶

Create a default configuration.

Source code in osm_data_client/models.py

@classmethod
def default(cls) -> "RawDataClientConfig":
    """Create a default configuration."""
    return cls()

options: show_source: false heading_level: 3

Options for controlling how output data is handled.

Source code in osm_data_client/models.py

@dataclass
class RawDataOutputOptions:
    """Options for controlling how output data is handled."""

    download_file: bool = True
    auto_extract: AutoExtractOption = AutoExtractOption.automatic

    @classmethod
    def default(cls) -> "RawDataOutputOptions":
        """Create default output options."""
        return cls()

`default()` `classmethod` ¶

Create default output options.

Source code in osm_data_client/models.py

@classmethod
def default(cls) -> "RawDataOutputOptions":
    """Create default output options."""
    return cls()

options: show_source: false heading_level: 3

API Docs¶

Raw Data API Python Client¶

Client¶

__init__(config=RawDataClientConfig.default()) ¶

get_osm_data(geometry, output_options=RawDataOutputOptions.default(), **kwargs) async ¶

__init__(config=RawDataClientConfig.default()) ¶

download_to_disk(data, options=RawDataOutputOptions.default()) async ¶

poll_task_status(task_link, polling_interval=2) async ¶

request_plain_geojson_snapshot(geometry, params) async ¶

request_snapshot(geometry, params) async ¶

Results¶

__str__() ¶

exists() ¶

suffix() ¶

Models¶

from_input(geometry) classmethod ¶

to_dict() ¶

from_kwargs(**kwargs) classmethod ¶

to_api_params() ¶

validate_bind_zip_compatibility(output_type, bind_zip) staticmethod ¶

__str__() ¶

from_api_result(result, params) classmethod ¶

memory_threshold_bytes: int property ¶

default() classmethod ¶

default() classmethod ¶

`init(config=RawDataClientConfig.default())` ¶

`get_osm_data(geometry, output_options=RawDataOutputOptions.default(), **kwargs)` `async` ¶

`init(config=RawDataClientConfig.default())` ¶

`download_to_disk(data, options=RawDataOutputOptions.default())` `async` ¶

`poll_task_status(task_link, polling_interval=2)` `async` ¶

`request_plain_geojson_snapshot(geometry, params)` `async` ¶

`request_snapshot(geometry, params)` `async` ¶

`str()` ¶

`exists()` ¶

`suffix()` ¶

`from_input(geometry)` `classmethod` ¶

`to_dict()` ¶

`from_kwargs(**kwargs)` `classmethod` ¶

`to_api_params()` ¶

`validate_bind_zip_compatibility(output_type, bind_zip)` `staticmethod` ¶

`str()` ¶

`from_api_result(result, params)` `classmethod` ¶

`memory_threshold_bytes: int` `property` ¶

`default()` `classmethod` ¶

`default()` `classmethod` ¶