""" This module contains classes for our main databases (reporting DB, Delphi) and AWS S3. To use this module in your script, import the module like so. ``` from djagitit import db ``` """ import os import time import boto3 from io import BytesIO import pandas as pd from sqlalchemy.exc import ProgrammingError, OperationalError, InternalError from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from snowflake.sqlalchemy import URL from abc import ABC, abstractmethod from dotenv import load_dotenv from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives.asymmetric import dsa from cryptography.hazmat.primitives import serialization load_dotenv() def safely_getenv(var): """Checks if environment variable is defined before returning it""" if not os.getenv(var): raise EnvironmentError(f'Environment variable {var} not defined!') return os.getenv(var) class Dbase(ABC): """Only to be used as an abstract base class for database classes such as ReportingDB and DelphiDB The class is non-instantiable, and any inherited class must ha a create_engine -method defined.""" def __new__(cls, *args, **kwargs): '''Makes the class non-instantiable''' if cls is Dbase: raise TypeError(f"only children of '{cls.__name__}' may be instantiated") return object.__new__(cls) # The create engine -method is required for any child classes @abstractmethod def create_engine(self): pass @staticmethod def handle_fatal_exceptions(e): if isinstance(e, ProgrammingError): print('Could not query, bad SQL. Closing connection.') print(f'Error caused by: {e.args[0]}') else: print(f'Unknown exception: {e}, Closing connection.') def query(self, query): """ Executes a data-fetching query. Args: query (str): SQL query string. Returns: df (pd.DataFrame): a dataframe with the fetched data if successful. False (bool): False if unsuccessful after multiple retries. Example: ``` q = "select * from common.dim_products where isrc_cd = 'USRC19401295'" data = rdb.query(q) ``` ???+ warning "Working with wildcards (%)" Note that when you have wildcards in your query, such as: ``` select count(*) from common.dim_products where primary_artist_name ilike 'mobb deep%' ``` The psycopg2 library will not escape the %-character, and the query will result in an error. To overcome this issue, please use a double wildcard in stead: ``` select count(*) from common.dim_products where primary_artist_name ilike 'mobb deep%%' ``` """ for _ in range(5): try: data = pd.read_sql(query, con=self.engine) return data except OperationalError: print('Could not query, connection issue. Refreshing connection and retrying.') self.refresh() except InternalError: print('Could not query, internal issue (likely timed out). Refreshing connection and retrying.') self.refresh() except Exception as e: self.handle_fatal_exceptions(e) break self.engine.dispose() return False def refresh(self): """Disposes the existing DB engine and creates a new one""" self.engine.dispose() time.sleep(2) self.engine = self.create_engine() class ReportingDB(Dbase): """ Used for working with the Reporting DB. Inherits from abstract base class Dbase. ???+ warning "Pre-requisites" To use this class, the following variables must be defined in your .env file: `hostReportingDB`, `portReportingDB`, `dbnameReportingDB` By default, the user credentials are also read from your env file: `usernameReportingDB`, `passwordReportingDB` However, you can also pass `user` and `password` arguments while initializing if you wish to use alternative credentials. See the [.env file configuration][configure-your-env-file] section if needed ???+ warning "Isolation level" This class uses a connection with autocommit, and every operation is executed separately. If you wish to work transactionally, use the ReportingDBSession -class! Before anything, you should initialize an instance of the ReportingDB class: ``` rdb = db.ReportingDB() ``` """ def __init__(self, user=None, password=None): self.user = safely_getenv('usernameReportingDB') if not user else user self.password = safely_getenv('passwordReportingDB') if not password else password self.host = safely_getenv('hostReportingDB') self.port = safely_getenv('portReportingDB') self.database = safely_getenv('dbnameReportingDB') self.engine = self.create_engine() def create_engine(self): return create_engine( f'redshift+psycopg2://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}', connect_args={'sslmode': 'verify-ca'} ) def execute(self, query): """ Execute a SQL query without expecting data as a result Args: query (str): SQL query string. Returns: status (bool): True if successful, False otherwise. Example: ``` table_name_example = 'sandbox.im_tired_of_getting_dropped_and_created' q = f''' drop table if exists {table_name_example}; create table if not exists {table_name_example} ( fixed_value varchar(20) encode zstd ); ''' status = rdb.execute(q) ``` ???+ warning "Working with wildcards (%)" Note that the wildcard-escaping issues in the `query`-method also apply to the execute-method """ for _ in range(5): try: self.engine.execute(query) return True except OperationalError: print('Could not Execute, connection issue. Refreshing connection and retrying.') self.refresh() except Exception as e: self.handle_fatal_exceptions(e) break self.engine.dispose() return False def write(self, data, table_name, chunksize=50000): """ You can obviously also write into the reporting DB! Simply pass your data as pandas dataframe, and specify the table in the DB. The table doesn't need to exist beforehand. If it does, data will be appended, not overwritten! Args: data (pd.DataFrame): Data to be written to the database. table_name (str): Name of the table in the database, in the format: schema.table chunksize (int, optional): Number of rows to write at a time (default is 50000). Returns: status (bool): True if successful, False otherwise. Example: ``` import pandas as pd data = pd.DataFrame({'fixed_value': 'djagitit'}) status = rdb.write(data, table_name=table_name_example) ``` In most cases, the above will suffice. However, if you have a massive data set, you might want to use the write_df_to_reportingDB() method from the S3 class for writing into the database. """ schema = table_name.split('.')[0] tname = table_name.split('.')[1] if not isinstance(data, pd.DataFrame): raise TypeError(f'Bad data type {type(data)}, must be pandas DataFrame') try: data.to_sql( tname, con=self.engine, schema=schema, if_exists='append', method='multi', index=False, chunksize=chunksize ) return True except: return False class RDBSError(Exception): """Custom exception for when a part of a RDB transaction fails.""" def __init__(self): self.message = "An operation failed within a transaction." super().__init__(self.message) class ReportingDBSession(ReportingDB): """ Used for working transactionally with the Reporting DB. Inherits from class ReportingDB. ???+ warning "Pre-requisites" To use this class, the following variables must be defined in your .env file: `hostReportingDB`, `portReportingDB`, `dbnameReportingDB` By default, the user credentials are also read from your env file: `usernameReportingDB`, `passwordReportingDB` However, you can also pass `user` and `password` arguments while initializing if you wish to use alternative credentials. See the [.env file configuration][configure-your-env-file] section if needed ???+ warning "Isolation level" This class uses a connection **WITHOUT** autocommit, and requires an explicit commit. If you wish to work with autocommit, use the ReportingDB -class! Before anything, you should initialize an instance of the ReportingDBSession class: ``` rdbs = db.ReportingDBSession() ``` """ def __init__(self, user=None, password=None): u = safely_getenv('usernameReportingDB') if not user else user p = safely_getenv('passwordReportingDB') if not password else password super().__init__(user=u, password=p) Session = sessionmaker(bind=self.engine) self.session = Session() def execute(self, query): """ Execute a SQL query without committing. If the execution fails, a rollback is invoked automatically. Args: query (str): SQL query string. Raises: RDBSError: If unsuccessful, an exception of type RDBSError is raised. Example: ``` table_name_example = 'sandbox.im_tired_of_getting_dropped_and_created' q = f''' drop table if exists {table_name_example}; create table if not exists {table_name_example} ( fixed_value varchar(20) encode zstd ); ''' rdbs.execute(q) # Remember to also commit rdbs.commit() ``` ???+ warning "Working with wildcards (%)" Note that the wildcard-escaping issues in the `query`-method also apply to this execute-method """ try: self.session.execute(query) except Exception as e: print(f'Execute failed: {e}') self.session.rollback() raise RDBSError def commit(self): """ Commits the transaction. If the commit fails, a rollback is invoked automatically and an RDBSError is raised. Raises: RDBSError: If the commit fails, an exception of type RDBSError is raised. Example: ``` rdbs.commit() ``` """ try: self.session.commit() print("Transaction committed.") except Exception as e: print(f"Commit failed: {e}") self.session.rollback() raise RDBSError def close(self): """ Closes the session. Call at the end of your script after committing. Example: ``` rdbs.close() ``` """ self.session.close() def refresh(self): """ Overrides the parent class method as obsolete. You shouldn't refresh the connection when running transactions. """ raise NotImplementedError("The connection attribute of a ReportingDbSession instance should not be refreshed.") def query(self, query): """ Overrides the parent class method as obsolete. You should use the ReportingDB -class for querying and expecting data as a result. """ raise NotImplementedError("Use an instance of ReportingDB to query data.") def write(self, *args, **kwargs): """ Overrides the parent class method as obsolete. You should use the ReportingDB -class for writing. """ raise NotImplementedError("Use an instance of ReportingDB to write.") class DelphiDB(Dbase): """ Used for working with the Delphi DB. Inherits from abstract base class Dbase. ???+ warning "Pre-requisites" To use this class, the following variables must be defined in your .env file: `accountDelphi`, `dbnameDelphi`, `warehouseDelphi`, `usernameDelphi` and either `privateKeyDelphi`, `passPhraseDelphi` or `passwordDelphi` See the [.env file configuration][configure-your-env-file] section if needed Before anything, you should initialize an instance of the DelphiDB class and specify the schema: ``` dlp = db.DelphiDB('AD_DATA') ``` """ def __init__(self, schema, environment='prod'): if schema not in ['EXP', 'CHARTMETRIC', 'AD_DATA', 'ADS_DATA_CONSOLIDATED']: raise ValueError(f"Unknown schema {schema}: must be one of EXP, CHARTMETRIC, AD_DATA, ADS_DATA_CONSOLIDATED") if environment not in ['prod', 'dev', 'stage']: raise ValueError(f"Unknown environment {environment}: must be one of prod, stage, dev (default value: prod)") inject = '' if environment == 'prod' else environment.captalize() self.password = os.getenv(f'passwordDelphi{inject}') self.privateKey = os.getenv(f'privateKeyDelphi{inject}') self.account = safely_getenv(f'accountDelphi{inject}') self.database = safely_getenv(f'dbnameDelphi{inject}') self.warehouse = safely_getenv('warehouseDelphi') self.user = safely_getenv('usernameDelphi') self.schema = schema self.engine = self.create_engine() def create_engine(self): if not self.privateKey: if not self.password: raise EnvironmentError('Environment variable EITHER privateKeyDelphi OR passwordDelphi not defined!') #print('pwd') engine = create_engine( f'''snowflake://{self.user}:{self.password}@{self.account}/{self.database}/{self.schema}?warehouse={self.warehouse}''' ) else: print('keypair') passPhrase = safely_getenv('passPhraseDelphi') with open(self.privateKey, "rb") as key: p_key = serialization.load_pem_private_key( key.read(), password=passPhrase.encode(), backend=default_backend() ) pkb = p_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() ) engine = create_engine( URL( account=self.account, user=self.user, database=self.database, schema=self.schema, warehouse=self.warehouse ), connect_args={ 'private_key': pkb, }, ) return engine class S3: """ > For this, you'll need to be set up for AWS s3. Contact Julien or Matti if you need this! Used for writing/pulling data from S3 and manage your S3 bucket ???+ warning "Pre-requisites" To use this class, the following variables must be defined in your .env file: `awsAccessKeyId`, `awsSecretAccessKey`, `awsS3Bucket` See the [.env file configuration][configure-your-env-file] section if needed Before anything, you should initialize an instance of the S3 class: ``` s3 = db.S3() ``` """ def __init__(self): self.key = safely_getenv('awsAccessKeyId') self.secret = safely_getenv('awsSecretAccessKey') self.bucket = safely_getenv('awsS3Bucket') self.connection = self.__connect() def __connect(self): session = boto3.Session(aws_access_key_id=self.key, aws_secret_access_key=self.secret) return session.resource('s3') def __get_copy_query(self, tablename, s3_path): """ Generates a query to copy data from S3 into a table. Args: tablename (str): Name of the table. s3_path (str): S3 path. Returns: str: Query string to copy data from S3 into a table. """ s3_query = f''' copy {tablename} from 's3://{self.bucket}/{s3_path}' CREDENTIALS 'aws_access_key_id={self.key};aws_secret_access_key={self.secret}' format as parquet; COMMIT; ''' return s3_query def __get_unload_query(self, tablename, s3_path, format=None, csv_delimiter='|', overwrite=False): """ Generates a query to unload data from a table into S3. Args: tablename (str): Name of the table. s3_path (str): S3 path. format (str, optional): Format of the file. Defaults to None. csv_delimiter (str, optional): Delimiter for csv format. Defaults to '|'. overwrite (bool, optional): If true, allows overwriting of existing files. Defaults to False. Returns: str: Query string to unload data from a table into S3. """ # Define the dynamic clauses header_clause = 'HEADER' if format=='csv' else '' delimiter_clause = f'''DELIMITER '{csv_delimiter}' ''' if format=='csv' else '' overwrite_clause = 'ALLOWOVERWRITE' if overwrite else '' format_clause = 'csv' if format=='csv' else 'parquet' # Generate the query s3_query = f''' UNLOAD ('select * from {tablename}') TO 's3://{self.bucket}/{s3_path}' CREDENTIALS 'aws_access_key_id={self.key};aws_secret_access_key={self.secret}' PARALLEL OFF {header_clause} FORMAT {format_clause} EXTENSION '{format_clause}' {delimiter_clause} {overwrite_clause}; ''' return s3_query def write_df_to_s3(self, df, s3_path, format='parquet', csv_delimiter='|'): """ Writes a DataFrame to an S3 path. Args: df (pd.DataFrame): The DataFrame to write. s3_path (str): The S3 path to write the DataFrame to. format (str, optional): The format to write the DataFrame in. If used, valid options are 'csv' or 'parquet'. Defaults to 'parquet'. csv_delimiter (str, optional): The delimiter to use if writing in csv format. Defaults to '|'. Example: ``` df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]}) # This will write to s3://[YOUR_BUCKET_NAME]/mobbdeep/infamous000.parquet s3.write_df_to_s3(df, 'mobbdeep/infamous') # This will write to s3://[YOUR_BUCKET_NAME]/mobbdeep/infamous.csv s3.write_df_to_s3(df, 'mobbdeep/infamous', format='csv', csv_delimiter=';') ``` """ if not isinstance(df, pd.DataFrame): raise TypeError(f'Bad data type for df {type(df)}, must be pandas DataFrame') if not isinstance(s3_path, str): raise TypeError(f'Bad data type for s3_path {type(s3_path)}, must be a string') if format not in ['csv', 'parquet']: raise ValueError(f''' Wrong format {format}: format should be either 'csv' or 'parquet' (default to parquet)''') # Create the file buffer depending on the format file_buffer = BytesIO() if format=='csv': df.to_csv(file_buffer, index=False, sep=f'{csv_delimiter}' ) else: df.to_parquet(file_buffer, index=False) # Upload the file buffer to S3 self.connection.Object(self.bucket, s3_path+'.'+format).put(Body=file_buffer.getvalue()) return True def write_s3_to_reportingDB(self, s3_path, tablename): """ Writes data from an S3 path to a table in the reporting database. **Note that when writing this way, the table needs to exist beforehand!** Args: s3_path (str): The S3 path where the data is stored. tablename (str): The name of the table in the reporting database (in the form: schemaname.tablename). Example: ``` s3.write_s3_to_reportingDB('mobbdeep/infamous.csv', 'prod_eu_analytics.my_awesome_table') ``` """ if not isinstance(s3_path, str): raise TypeError(f'Bad data type for s3_path {type(s3_path)}, must be a string') if not isinstance(tablename, str): raise TypeError(f'Bad data type for tablename {type(tablename)}, must be a string') if len(tablename.split('.')) != 2: raise ValueError(f'''Wrong syntax for table name {tablename} : must be schemaname.tablename''') # Generate the query s3_query = self.__get_copy_query(tablename, s3_path) # Instantiate the database and execute the query db = ReportingDB() db.execute(s3_query) db.engine.dispose() return True def write_reportingDB_to_s3(self, tablename, s3_path, format='parquet', csv_delimiter='|', overwrite=False): """ Writes data from a table in the reporting database to an S3 path. Args: tablename (str): The name of the table in the reporting database (in the form: schemaname.tablename). s3_path (str): The S3 path where the data will be stored. format (str, optional): The format to write the DataFrame in. If used, valid options are 'csv' or 'parquet'. Defaults to 'parquet'. csv_delimiter (str, optional): The delimiter to use if writing in csv format. Defaults to '|'. overwrite (bool, optional): Whether to overwrite the data if it already exists. Defaults to False. Example: ``` # This will write to s3://[YOUR_BUCKET_NAME]/mobbdeep/infamous000.parquet s3.write_reportingDB_to_s3('prod_france.my_awesome_table', 'mobbdeep/infamous') # This will write to s3://[YOUR_BUCKET_NAME]/mobbdeep/infamous.csv s3.write_reportingDB_to_s3('prod_france.my_awesome_table', 'mobbdeep/infamous', format='csv', csv_delimiter=';') ``` """ if not isinstance(s3_path, str): raise TypeError(f'Bad data type for s3_path {type(s3_path)}, must be a string') if not isinstance(tablename, str): raise TypeError(f'Bad data type for tablename {type(tablename)}, must be a string') if len(tablename.split('.')) != 2: raise ValueError(f'''Wrong syntax for table name {tablename} : must be schemaname.tablename''') if format not in ['csv', 'parquet', None]: raise ValueError(f''' Wrong format {format}: format should be either 'csv' or 'parquet' (default to parquet)''') # Generate the query s3_query = self.__get_unload_query( tablename, s3_path, format, csv_delimiter, overwrite) # Instantiate the database and execute the query db = ReportingDB() db.execute(s3_query) db.engine.dispose() return True def write_df_to_reportingDB(self, df, tablename): """ Writes data from a pandas DataFrame to a table in the reporting database. **Note that when writing this way, the table needs to exist beforehand!** Args: df (pandas.DataFrame): The DataFrame containing the data to be written. tablename (str): The name of the table in the reporting database (in the form: schemaname.tablename). Example: ``` df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]}) s3.write_df_to_reportingDB(df, 'prod_eu_analytics.my_awesome_table') ``` """ if not isinstance(df, pd.DataFrame): raise TypeError(f'Bad data type for df {type(df)}, must be pandas DataFrame') if not isinstance(tablename, str): raise TypeError(f'Bad data type for tablename {type(tablename)}, must be a string') if len(tablename.split('.')) != 2: raise ValueError(f'''Wrong syntax for table name {tablename} : must be schemaname.tablename''') # Define the S3 path where the data will be temporarily stored s3_path = f'ingestion/{tablename}' # Write the data from the DataFrame to S3 self.write_df_to_s3(df, s3_path) # Write the data from S3 to the table in the reporting database self.write_s3_to_reportingDB(s3_path, tablename) # Delete the temporary file from S3 self.connection.Object(self.bucket, s3_path).delete() return True # THIS METHOD SEEMS OVERCOMPLICATED TO USE INSTEAD OF THE SPECIFIC METHODS, DESACTIVED FOR NOW # def write(self, *, from_df=None, from_s3_path=None, from_tablename=None, to_tablename=None, to_s3_path=None, format=None, csv_delimiter=None, overwrite=False): # """ # Writes data from various sources to various destinations. # # Usage: # - All args should be specified by their respective names. # - Only one of from_df, from_s3_path, from_tablename should be specified. # - Only one of to_s3_path, to_tablename should be specified. # - format and csv_delimiter (optionnal) would only be used if to_s3_path is specified. # - overwrite (optionnal) would only be used if both from_tablename and to_s3_path are specified. # # Args: # from_df (pandas.DataFrame, optional): The DataFrame containing the data to be written. # from_s3_path (str, optional): The S3 path where the data is stored. # from_tablename (str, optional): The name of the table in the reporting database from where the data is to be written. # to_tablename (str, optional): The name of the table in the reporting database where the data will be written. # to_s3_path (str, optional): The S3 path where the data will be written. # format (str, optional): The format to write the data in. Defaults to None. # csv_delimiter (str, optional): The delimiter to use if writing in csv format. Defaults to None. # overwrite (bool, optional): Whether to overwrite the data if it already exists. Defaults to False. # """ # # Write data from a pandas DataFrame to a table in the reporting database # if (from_df is not None and to_tablename is not None ) and (from_s3_path or from_tablename or to_s3_path) is None: # self.__write_df_to_reportingDB(df=from_df, tablename=to_tablename) # return True # # Write data from a pandas DataFrame to S3 # elif (from_df is not None) and (to_s3_path is not None) and (from_s3_path or from_tablename or to_tablename) is None: # self.__write_df_to_s3(df=from_df, s3_path=to_s3_path, format=format, csv_delimiter=csv_delimiter) # return True # # Write data from S3 to a table in the reporting database # elif (from_s3_path is not None and to_tablename is not None ) and (from_df or from_tablename or to_s3_path) is None: # self.__write_s3_to_reportingDB(s3_path=from_s3_path, tablename=to_tablename) # return True # # Write data from a table in the reporting database to S3 # elif (from_tablename is not None and to_s3_path is not None ) and (from_df or from_s3_path or to_tablename) is None: # self.__write_reportingDB_to_s3(tablename=from_tablename, s3_path=to_s3_path, format=format, csv_delimiter=csv_delimiter, overwrite=overwrite) # return True # # All Other cases should raise an error # else: # print('''Wrong call - Please conform to the following : # - All args should be specified by their respective names. # - Only one of from_df, from_s3_path, from_tablename should be specified. # - Only one of to_s3_path, to_tablename should be specified. # - format and csv_delimiter (optionnal) would only be used if to_s3_path is specified. # - overwrite (optionnal) would only be used if both from_tablename and to_s3_path are specified. # ''') # return False def upload_file(self, local_path, s3_path=None): """ Uploads a file to S3. Args: local_path (str): The local path of the file to be uploaded. s3_path (str, optional): The S3 path where the file will be uploaded. If not specified, the file will be uploaded to 'ingestion/' directory with the same filename as the local file. Returns: status (bool): The status of the upload. True if successful, False otherwise. Example: ``` s3.upload_file(local_path='my_awesome_directory/infamous.csv', s3_path='mobbdeep/infamous.csv') ``` """ # Default s3_path if not specified if s3_path is None: s3_path = f'ingestion/{os.path.basename(local_path)}' # Upload the file status = self.connection.Object(self.bucket, s3_path).upload_file(local_path) return status def file_list(self, s3_path=''): """ Lists all files in a given S3 path. Args: s3_path (str, optional): The S3 path to list files from. Defaults to ''. Returns: fileList (list): A list of file keys in the given S3 path. Example: ``` # List all files in your bucket s3.file_list() # List all files in 'mobbdeep/' s3.file_list('mobbdeep/') ``` """ # Look for all files in the S3 path bucket = self.connection.Bucket(self.bucket) objs = bucket.objects.filter(Prefix=s3_path) # Build the list of files fileList = [] for obj in objs: print(obj.key) fileList.append(obj.key) # Handle the case where no files are found if fileList==[]: print(f'No File for s3_path {s3_path}') return fileList def download_files(self, s3_path='', local_directory='.'): """ Downloads all files from a given S3 path to a local directory. Args: s3_path (str, optional): The S3 path to download files from. Defaults to ''. local_directory (str, optional): The local directory where the files will be downloaded. Defaults to '.'. Returns: fileList (list): A list of local file paths where the files have been downloaded. Example: ``` # Download all files in your bucket s3.download_files() # Download all files from 'mobbdeep/' to the current directory s3.download_files('mobbdeep/') # Download all files from 'mobbdeep/' to 'my_awesome_directory' s3.download_files('mobbdeep/', 'my_awesome_directory') ``` """ # Look for all files in the S3 path bucket = self.connection.Bucket(self.bucket) objs = bucket.objects.filter(Prefix=s3_path) fileList = [] # For each file, download it and add it to the list of files for obj in objs: print(obj.key) # Define the local filename download_dir = local_directory if local_directory[-1]=='/' else local_directory+'/' filename = download_dir+obj.key.replace('/', '_') #Download the file self.connection.meta.client.download_file(self.bucket, obj.key, filename) print(f''' downloaded as {filename}''') fileList.append(filename) return fileList def __delete(self, s3_path): """ Deletes all files from a given S3 path. Args: s3_path (str): The S3 path from where the files will be deleted. Returns: fileList (list): A list of S3 file paths that have been deleted. """ # Look for all files in the S3 path bucket = self.connection.Bucket(self.bucket) objs = bucket.objects.filter(Prefix=s3_path) # Delete all the files r = objs.delete() fileList = [] # If any files have been deleted, add them to the list if r!=[]: for file in r[0].get('Deleted'): print(f'''File deleted : {file.get('Key')}''') fileList.append(file.get('Key')) # If no files have been deleted, print a message if fileList==[]: print(f'No File deleted for s3_path {s3_path}') return fileList def delete_files(self, s3_path): """ Deletes files from a given S3 path. You can't use it to delete the whole bucket, use the empty_bucket method instead Args: s3_path (str): The S3 path from where the files will be deleted. Returns: (list): A list of S3 file paths that have been deleted. Example: ``` # Delete a single file s3.delete_files('mobbdeep/infamous.csv') # Delete all files in 'mobbdeep/' s3.delete_files('mobbdeep/') ``` """ if s3_path == '' or s3_path == '/' or s3_path == '*': raise ValueError('You cannot use an empty s3_path, if you need to empty the whole bucket use the empty_bucket method instead') return self.__delete(s3_path) def empty_bucket(self, sure='no'): """ Empties the entire S3 bucket. Note: **Be careful as this will erase all files within your bucket and this action cannot be undone!** Args: sure (str, optional): A confirmation that you actually want to empty the bucket. Defaults to 'no'. Returns: (list): A list of S3 file paths that have been deleted. Example: ``` s3.empty_bucket(sure='yes') ``` """ if sure != 'yes': raise ValueError("This operation will empty your whole Bucket, if you are sure call the method with sure='yes'") return self.__delete('')