Created
November 30, 2023 11:53
-
-
Save arbakker/730305c0bbde32070f0d98a0a0144e86 to your computer and use it in GitHub Desktop.
Python CLI for creating full custom downloads on the PDOK download API
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment