Python CLI for creating full custom downloads on the PDOK download API

pdok-download-api-full-custom.py

import requests
import json
import time
import os
import pathlib
import argparse
from urllib.parse import urlparse


def main(api_url, featuretypes, format, bbox_rd=None, geofilter=None):
    if bbox_rd is None and geofilter is None:
        raise ValueError("both bbox and geofilter args are None")
    if bbox_rd is not None:
        geofilter = "POLYGON(({} {},{} {},{} {},{} {},{} {}))".format(
            bbox_rd[0],
            bbox_rd[1],
            bbox_rd[0],
            bbox_rd[3],
            bbox_rd[2],
            bbox_rd[3],
            bbox_rd[2],
            bbox_rd[1],
            bbox_rd[0],
            bbox_rd[1],
        )  # noqa: E501

    # zie api docs: https://api.pdok.nl/lv/bgt/download/v1_0/ui/
    print(f"view api docs: {api_url}/ui/")
    api_url_p = urlparse(api_url)
    base_url = f"{api_url_p.scheme}://{api_url_p.netloc}"
    full_custom_url = f"{api_url}/full/custom"
    headers = {"Content-Type": "application/json", "Accept": "application/json"}

    data = {"featuretypes": featuretypes, "format": format, "geofilter": geofilter}
    r = requests.post(full_custom_url, data=json.dumps(data), headers=headers)

    if r.status_code != 202:
        print("error occured creating custom download")
        print(f"response status: {r.status_code}")
        print(f"response body: {r.text}")
        exit(1)

    response_object = json.loads(r.text)
    status_path = response_object["_links"]["status"]["href"]
    download_request_id = response_object["downloadRequestId"]
    download_url = None
    status_url = f"{base_url}{status_path}"
    while True:
        r = requests.get(status_url)
        status_object = r.json()
        custom_status = status_object["status"]
        print(f"status generating download: {custom_status}")
        if custom_status == "COMPLETED":
            download_path = status_object["_links"]["download"]["href"]
            download_url = f"{status_url}{download_path}"
            break
        elif custom_status == "PENDING":
            print(f"progress generating download:  {status_object['progress']}")
        else:
            break
        time.sleep(2)

    if download_url:
        filename = f"{download_request_id}_{os.path.basename(download_url)}"
        filepath = os.path.join(pathlib.Path().absolute(), filename)
        print(f"downloading file {download_url} to {filepath}")
        r = requests.get(download_url)
        with open(filepath, "wb") as f:
            f.write(r.content)
    else:
        print("error occured generating download")


def list_str(values):
    return values.split(",")

class UltimateHelpFormatter(
    argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFormatter
):
    pass

if __name__ == "__main__":
    program_name = os.path.basename(__file__)
    example_1 = f'{program_name} https://api.pdok.nl/lv/bgt/download/v1_0 "wegdeel,weginrichtingselement,wijk" --bbox-rd "200417.7884,446856.2778,200942.1976,447187.3237" --format citygml'  # noqa: E501
    example_2 = f'{program_name} https://api.pdok.nl/kadaster/kadastralekaart/download/v5_0 "perceel,kadastralegrens" --bbox-rd "200417.7884,446856.2778,200942.1976,447187.3237" --format gml'  # noqa: E501
    parser = argparse.ArgumentParser(
        prog=program_name,
        description="CLI for creating full custom downloads on the PDOK download API",
        epilog=f'exampe usage:\n\n1) {example_1}\n2) {example_2}',  # noqa: E501
        formatter_class=UltimateHelpFormatter,
    )
    parser.add_argument("api_url")
    parser.add_argument("featuretypes", type=list_str)
    parser.add_argument("--bbox-rd")
    parser.add_argument("--geofilter")
    parser.add_argument("--format")
    args = parser.parse_args()
    main(**vars(args))