Add function to delete objects from s3

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