Demo script for requesting routes from PolarRouteServer API using Python standard library.

This script demonstrates the two-step workflow: 1. Submit a route request to /api/route, which returns a job ID 2. Monitor job status at /api/job/{job_id} until completion 3. When job is complete, retrieve route data from /api/route/{route_id}

make_request(type, url, endpoint, headers, body=None)

Sends HTTP request, prints details and returns response.

Parameters:
  • type (str) –

    HTTP request type, e.g. "GET" or "POST"

  • url (str) –

    base url to send request to

  • endpoint (str) –

    endpoint, e.g. "/api/route/some-id"

  • headers (dict) –

    HTTP headers

  • body (dict, default: None ) –

    HTTP request body. Defaults to None.

Returns:
  • HTTPResponse

    http.client.HTTPResponse

Source code in polarrouteserver/demo.py
def make_request(
    type: str, url: str, endpoint: str, headers: dict, body: dict = None
) -> http.client.HTTPResponse:
    """Sends HTTP request, prints details and returns response.

    Args:
        type (str): HTTP request type, e.g. "GET" or "POST"
        url (str): base url to send request to
        endpoint (str): endpoint, e.g. "/api/route/some-id"
        headers (dict): HTTP headers
        body (dict, optional): HTTP request body. Defaults to None.

    Returns:
        http.client.HTTPResponse
    """
    sending_str = f"Sending {type} request to {url}{endpoint}: \nHeaders: {headers}\n"

    if body:
        sending_str += f"Body: {body}\n"

    print(sending_str)

    # data = parse.urlencode(body).encode("utf-8") if body else None
    req = request.Request(url + endpoint, data=body, headers=headers)
    unverified_context = ssl._create_unverified_context()
    response = request.urlopen(req, context=unverified_context)

    print(f"Response: {response.status} {response.reason}")

    return json.loads(response.read()), response.status

parse_location(location)

Parameters:
  • location (str) –

    a location either as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01

Returns:
  • Location

    a Location object

Source code in polarrouteserver/demo.py
def parse_location(location: str) -> Location:
    """
    Args:
     location (str): a location either as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01

    Returns:
        a Location object
    """
    pattern = r"[+-]?([0-9]*[.])?[0-9]+,[+-]?([0-9]*[.])?[0-9]+"
    if location in STANDARD_LOCATIONS.keys():
        standard_location = STANDARD_LOCATIONS.get(location)
        return standard_location
    elif re.search(pattern, location):
        coords = re.search(pattern, location).group().split(",")
        return Location(float(coords[0]), float(coords[1]))
    else:
        raise ValueError(
            f"Expected input as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01, got {location}"
        )

request_route(url, start, end, status_update_delay=30, num_requests=10, force_new_route=False, mesh_id=None)

Requests a route from polarRouteServer, monitors job status until complete, then retrieves route data.

Parameters:
  • url (str) –

    Base URL to send request to.

  • start (Location) –

    Start location of route

  • end (Location) –

    End location of route

  • status_update_delay (int, default: 30 ) –

    Delay in seconds between each status request. Defaults to 10.

  • num_requests (int, default: 10 ) –

    Max number of status requests before giving up. Defaults to 10.

  • force_new_route (bool, default: False ) –

    Force recalculation of an already existing route. Default: False.

  • mesh_id (int, default: None ) –

    Custom mesh ID to use for route calculation. Default: None.

Raises:
  • Exception

    If no status URL is returned.

Returns:
  • str( str ) –

    JSON response of route data, or None if request failed.

Source code in polarrouteserver/demo.py
def request_route(
    url: str,
    start: Location,
    end: Location,
    status_update_delay: int = 30,
    num_requests: int = 10,
    force_new_route: bool = False,
    mesh_id: int = None,
) -> str:
    """Requests a route from polarRouteServer, monitors job status until complete, then retrieves route data.

    Args:
        url (str): Base URL to send request to.
        start (Location): Start location of route
        end (Location): End location of route
        status_update_delay (int, optional): Delay in seconds between each status request. Defaults to 10.
        num_requests (int, optional): Max number of status requests before giving up. Defaults to 10.
        force_new_route (bool, optional): Force recalculation of an already existing route. Default: False.
        mesh_id (int, optional): Custom mesh ID to use for route calculation. Default: None.

    Raises:
        Exception: If no status URL is returned.

    Returns:
        str: JSON response of route data, or None if request failed.
    """

    # make route request
    response_body, status = make_request(
        "POST",
        url,
        "/api/route",
        {"Content-Type": "application/json"},
        json.dumps(
            {
                "start_lat": start.lat,
                "start_lon": start.lon,
                "end_lat": end.lat,
                "end_lon": end.lon,
                "start_name": start.name,
                "end_name": end.name,
                "force_new_route": force_new_route,
                "mesh_id": mesh_id,
            },
        ).encode("utf-8"),
    )

    print(pprint.pprint(response_body))

    if not str(status).startswith("2"):
        return None

    # if route is returned
    if response_body.get("json") is not None:
        return response_body["json"]

    # if no route returned, request status at status-url
    status_url = response_body.get("status-url")
    if status_url is None:
        print(
            "No status URL returned. Route may have failed or been returned immediately."
        )
        return None
    job_id = response_body.get("id")

    status_request_count = 0
    while status_request_count <= num_requests:
        status_request_count += 1
        print(
            f"\nWaiting for {status_update_delay} seconds before sending status request."
        )
        time.sleep(status_update_delay)

        # make job status request
        print(f"Status request #{status_request_count} of {num_requests}")
        status_response, status_code = make_request(
            "GET",
            url,
            f"/api/job/{job_id}",
            headers={"Content-Type": "application/json"},
        )

        print(f"Route calculation {status_response.get('status')}.")
        print(pprint.pprint(status_response))
        if status_response.get("status") == "PENDING":
            continue
        elif status_response.get("status") == "FAILURE":
            return None
        elif status_response.get("status") == "SUCCESS":
            # Job is complete, now get the actual route data
            route_url = status_response.get("route_url")
            if route_url:
                # Extract route ID from the route_url (e.g., "/api/route/123")
                route_id = route_url.split("/")[-1]
                print(f"Job complete! Fetching route data from route ID: {route_id}")

                route_response, route_status = make_request(
                    "GET",
                    url,
                    f"/api/route/{route_id}",
                    headers={"Content-Type": "application/json"},
                )

                if str(route_status).startswith("2"):
                    return route_response
                else:
                    print(f"Failed to fetch route data: {route_status}")
                    return None
            else:
                print("Job completed but no route_url provided")
                return None
    print(
        f'Max number of requests sent. Quitting.\nTo send more status requests, run: "curl {url}/api/job/{job_id}"'
    )
    return None