Utilities (emmaa.util)

exception emmaa.util.NotAClassName[source]

Bases: Exception

emmaa.util.does_exist(bucket, prefix, extension=None)[source]

Check if the file with exact key or starting with prefix and/or with extension exist in a bucket.

emmaa.util.find_latest_emails(email_type, time_delta=None, w_dt=False)[source]

Return a list of keys of the latest emails delivered to s3

Parameters:
  • email_type (str) – The email type to look for, e.g. ‘feedback’ if listing bounce and complaint emails sent to the ReturnPath address.
  • time_delta (datetime.timedelta) – The timedelta to look backwards for listing emails.
  • w_dt (bool) – If True, return a list of (key, datetime.datetime) tuples.
Returns:

A list of keys to the emails of the specified type. If w_dt is True, each item is a tuple of (key, datetime.datetime) of the LastModified date.

Return type:

list[Keys]

emmaa.util.find_latest_s3_file(bucket, prefix, extension=None)[source]

Return the key of the file with latest date string on an S3 path

emmaa.util.find_latest_s3_files(number_of_files, bucket, prefix, extension=None)[source]

Return the keys of the specified number of files with latest date strings on an S3 path sorted by date starting with the earliest one.

emmaa.util.find_nth_latest_s3_file(n, bucket, prefix, extension=None)[source]

Return the key of the file with nth (0-indexed) latest date string on an S3 path

emmaa.util.get_s3_client(unsigned=True)[source]

Return a boto3 S3 client with optional unsigned config.

Parameters:unsigned (Optional[bool]) – If True, the client will be using unsigned mode in which public resources can be accessed without credentials. Default: True
Returns:A client object to AWS S3.
Return type:botocore.client.S3
emmaa.util.make_date_str(date=None)[source]

Return a date string in a standardized format.

Parameters:date (Optional[datetime.datetime]) – A date object to get the standardized string for. If not provided, utcnow() is used to construct the date. (Note: using UTC is important because this code may run in multiple contexts).
Returns:The datetime string in a standardized format.
Return type:str
emmaa.util.sort_s3_files_by_date_str(bucket, prefix, extension=None)[source]

Return the list of keys of the files on an S3 path sorted by date starting with the most recent one.

emmaa.util.sort_s3_files_by_last_mod(bucket, prefix, time_delta=None, extension=None, unsigned=True, reverse=False, w_dt=False)[source]

Return a list of s3 object keys sorted by their LastModified date on S3

Parameters:
  • bucket (str) – s3 bucket to look for keys in
  • prefix (str) – The prefix to use for the s3 keys
  • time_delta (Optional[datetime.timedelta]) – If used, should specify how far back the to look for files on s3. Default: None
  • extension (Optional[str]) – If used, limit keys to those with the matching file extension. Default: None.
  • unsigned (bool) – If True, use unsigned s3 client. Default: True.
  • reverse (bool) – Reverse the sort order of the returned s3 files. Default: False.
  • w_dt (bool) – If True, return list with datetime object along with key as tuple (key, datetime.datetime). Default: False.
Returns:

A list of s3 keys. If w_dt is True, each item is a tuple of (key, datetime.datetime) of the LastModified date.

Return type:

list

emmaa.util.strip_out_date(keystring, date_format='datetime')[source]

Strips out datestring of selected date_format from a keystring