giambattista.bloisi
/
lot1-kickoff
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial stage

This commit is contained in:
Giambattista Bloisi 2024-07-02 11:45:36 +02:00
parent 833ea1538a
commit 0339a92de5
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
airflow/dags/openaire_to_S3.py
View File

@ -46,6 +46,7 @@ def download_uri(session: requests.Session, url: str, s3_client, bucket, key, ma
response = s3_client.create_multipart_upload(Bucket=bucket, response = s3_client.create_multipart_upload(Bucket=bucket,
Key=key) Key=key)
upload_id = response['UploadId']
tries = 0 tries = 0
while tries < max_retries: while tries < max_retries:
try: try:
@ -57,13 +58,12 @@ def download_uri(session: requests.Session, url: str, s3_client, bucket, key, ma
chunk_size = max(total_size // (10000 - 1), 15 * 1024 * 1024) chunk_size = max(total_size // (10000 - 1), 15 * 1024 * 1024)
for chunk in r.iter_content(chunk_size=chunk_size): for chunk in r.iter_content(chunk_size=chunk_size):
if chunk: if chunk:
# Upload part by part to S3
response = s3_client.upload_part( response = s3_client.upload_part(
Body=chunk, Body=chunk,
Bucket=bucket, Bucket=bucket,
Key=key, Key=key,
PartNumber=part_number, PartNumber=part_number,
UploadId=response['UploadId'] UploadId=upload_id
) )
parts.append({'PartNumber': part_number, 'ETag': response['ETag']}) parts.append({'PartNumber': part_number, 'ETag': response['ETag']})
current_size += len(chunk) current_size += len(chunk)
@ -83,14 +83,14 @@ def download_uri(session: requests.Session, url: str, s3_client, bucket, key, ma
s3_client.abort_multipart_upload( s3_client.abort_multipart_upload(
Bucket=bucket, Bucket=bucket,
Key=key, Key=key,
UploadId=response['UploadId'] UploadId=upload_id
) )
raise raise
s3_client.complete_multipart_upload( s3_client.complete_multipart_upload(
Bucket=bucket, Bucket=bucket,
Key=key, Key=key,
UploadId=response['UploadId'], UploadId=upload_id,
MultipartUpload={'Parts': parts} MultipartUpload={'Parts': parts}
) )