Check object Exist
ef check_obj_exist(s3, bucketname, region, key):
try:
s3.head_object(Bucket=bucketname, Key=key)
return True
except ClientError as e:
code = e.response.get("Error", {}).get("Code")
if code in ("404", "NoSuchKey", "NotFound"):
return False
raise
Put/Update S3 objects
- Put (create) and Update are the same call in S3:
put_object
- if Bucket is in a different region than your client, need region
def put_obj(s3, bucketname, region, key, body):
s3.put_object(
Bucket=bucketname,
Key=key,
Body=body
)
print(f"--- PUT {key} size: {len(body)} bytes ---")
Delete S3 objects
def delete_obj(s3, bucketname, region, key):
if not check_obj_exist(s3, bucketname, region, key):
print(f"--- Not exist: {key} ---")
return
s3.delete_object(Bucket=bucketname, Key=key)
print(f"--- Delete {key} ---")