diff --git a/opl/s3_tools.py b/opl/s3_tools.py index f7df042..919b1a9 100644 --- a/opl/s3_tools.py +++ b/opl/s3_tools.py @@ -1,11 +1,17 @@ import logging - import boto3 - import botocore.client +from typing import List, Dict -def connect(s3_conf): +def connect(s3_conf: Dict) -> boto3.resource: + """ + Connect to s3 + Args: + s3_conf: all aws access creadentials like access_key, secret_key, region, etc. + Returns: + s3 connection object + """ logging.debug("Going to connect") s3_resource = boto3.resource( "s3", @@ -42,3 +48,32 @@ def get_presigned_url(s3_resource, bucket, remote_name): ) logging.info(f"For {remote_name} obtained signed url {download_url}") return download_url + + +def delete_files( + s3_resource: boto3.resource, bucket_name: str, file_paths: str | List +) -> Dict | bool: + """ + Delete s3-objets from s3 bucket + Args: + s3_resource: create s3 resource object from using connect() and pass + bucket_name: s3 bucket name + file_paths: filename or list of file names you want to delete + Returns: + Delete response or False + """ + # Create a list of objects to delete + if isinstance(file_paths, list): + objects_to_delete = [{"Key": path} for path in file_paths] + elif isinstance(file_paths, str): + objects_to_delete = [{"Key": file_paths}] + else: + print(f"file_paths attribute must be string or list not {type(file_paths)}") + return False + logging.debug("Going to delete") + if objects_to_delete: + response = s3_resource.meta.client.delete_objects( + Bucket=bucket_name, Delete={"Objects": objects_to_delete} + ) + logging.debug("object/objects deleted") + return response