Last active
November 6, 2020 13:07
-
-
Save niconoe/3593689d164985a5e3b5f0300c2ab649 to your computer and use it in GitHub Desktop.
Python: Synchronously (block until it's ready) request a GBIF download
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
# Quick and dirty gist to show how to request a GBIF download, block until it's ready then finally download it. | |
# (no need to monitor an email address, nor to install PyGBIF and all its dependencies. Only requests is needed) | |
# TODO: use download id as filename | |
# TODO: improve make_predicate function so it's much more flexible | |
from time import sleep | |
import requests | |
SLEEP_DURATION = 20 | |
def make_predicate_dataset_key( | |
dataset_key: str, username: str, notification_address: str | |
): | |
v = { | |
"creator": username, | |
"notificationAddresses": [notification_address], | |
"predicate": {"type": "equals", "key": "DATASET_KEY", "value": dataset_key}, | |
} | |
return v | |
def test_blocking_download( | |
dataset_key: str, | |
username: str, | |
password: str, | |
notification_email: str, | |
output_path: str, | |
): | |
p = make_predicate_dataset_key(dataset_key, username, notification_email) | |
print(f"Will trigger a download for datasetKey={dataset_key}") | |
download_trigger = requests.post( | |
"https://api.gbif.org/v1/occurrence/download/request", | |
json=p, | |
auth=(username, password), | |
) | |
download_id = download_trigger.text | |
print(f"Download triggered, download_id is {download_id}") | |
while True: | |
print(f"(re)trying to get download {download_id}...") | |
download_get = requests.get( | |
f"https://api.gbif.org/v1/occurrence/download/request/{download_id}" | |
) | |
status = download_get.status_code | |
if status != 404: # 404: not ready yet | |
if status == 200: | |
print(f"Download is ready, getting it") | |
with open(output_path, "wb") as f: | |
f.write(download_get.content) | |
else: | |
print(f"Status is {status}, why?") | |
print(download_get) | |
break | |
sleep(SLEEP_DURATION) | |
if __name__ == "__main__": | |
test_blocking_download( | |
dataset_key="b7ee2a4d-8e10-410f-a951-a7f032678ffe", | |
username="<YOUR_GBIF_USERNAME_HERE>", | |
password="<YOUR_GBIF_PASSWORD_HERE>", | |
notification_email="<YOUR_GBIF_EMAIL_HERE>", | |
output_path="download.zip", | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment