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_credentials(key, profile_name=None, cred_type='oauth1_0a')[source]

Get twitter credentials from AWS SSM

Parameters

  • key (str) – The initial key to the credentials in SSM. The full key will be /twitter/{key}/{par} where par is determined by the type of credentials.

  • profile_name (str) – The name of the AWS profile to use. If None (default), the default profile will be used.

  • cred_type (str) – The type of credentials to get. Choices are “oauth1_0a” and “bearer”. Default: “oauth1_0a”. Bearer uses OAuth 2.0.

Returns

A dictionary with the requested credentials.

Return type

dict

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.s3_put(bucket, key, body, unsigned_client, intelligent_tiering=True, **s3_options)[source]

A wrapper around boto3’s put_object method.

Parameters

  • bucket (str) – The S3 bucket to put the object in.

  • key (str) – The key to put the object in.

  • body (bytes) – The bytestring representation of an object to put in a body as a bytestring.

  • unsigned_client (bool) – Whether to use an unsigned client.

  • intelligent_tiering (bool) – Whether to use intelligent tiering. Default is True. If the object is smaller than 128 KB, it will be stored in the default storage class regardless of value of intelligent_tiering.

  • s3_options – Additional options to pass to the put_object method. If there are any duplicate values between the function parameters and the keys of s3_options, the values set in the parameters take precedence to the values set in the s3_options dict.

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