in datasets/new_york_taxi_trips/pipelines/_images/run_csv_transform_kub/csv_transform.py [0:0]
def download_file(source_url: str, source_file: pathlib.Path) -> bool:
logging.info(f"Downloading {source_url} into {source_file}")
success = True
r = requests.get(source_url, stream=True)
with open(source_file, "wb") as f:
for chunk in r:
f.write(chunk)
# if the file contains the string "<Code>NoSuchKey</Code>" then the url returned
# that it could not locate the respective file
if open(source_file, "rb").read().find(b"<Code>NoSuchKey</Code>") > -1:
success = False
if success:
logging.info(f"Download {source_url} to {source_file} complete.")
else:
logging.info(
f"Unable to download {source_url} to {source_file} at this time. The URL may not exist."
)
return success