This script will allow you grab the latest date for the bucket upload. The BUCKETS variable will maintain a list of the AWS S3 buckets you wish to receive updates for. This example include 4 different data sources that customers are uploading smtp, bloomberg email, bloomberg chat, and thomas routers chat files.
BUCKETS = ['04-prod-bb-chat-2868b1d6c06b', '04-prod-bb-email-b3b3503572be', '04-prod-smtp-archive-02269a7c9af5', '04-prod-trchat-ef96c2c80017']
This section dictates the login method to the AWS Console. You will need to change the key ID and access key as needed for your own environment. It then retrieves the bucket_name and last_object which will verify whether or not customers have uploaded their latest electronic communication files to our S3 bucket.
#login
s3 = boto3.resource(
service_name='s3',
region_name='eu-west-3',
aws_access_key_id='ASIAQON5W62NMFAGDRX5',
aws_secret_access_key='BeuQX+jbpJav/wAbYKJ2h3Ribfbx+O+SLLiY6SaD'
)
def get_last_modified_file(bucket_name):
client = boto3.client('s3', region_name=REGION_NAME)
response = client.list_objects_v2(Bucket=bucket_name)
last_object = sorted(response['Contents'], key=lambda item: item['LastModified'])[-1]
return last_object
def is_uploaded_today(last_object):
last_modified_datetime = last_object['LastModified']
current_date = datetime.now(timezone.utc)
ts = (current_date - last_modified_datetime).total_seconds()
if ts < SECS_IN_DAY:
return True
else:
return False
def get_date_from_key(name):
pass
The last part of my code provides a table in the CLI output. It will include the Bucket Name, File Name, Last Modified Date, and a Pass column. This is the only part of the code that still needs any real work but it gets the job done for now.
def main():
print(" ========== S3 Bucket Dashboard ======")
data_table = []
for bucket in BUCKETS:
try:
last_file = get_last_modified_file(bucket)
last_file_name = last_file['Key']
last_file_datetime = last_file['LastModified'].strftime("%m/%d/%Y: %H:%M:%S")
was_last_file_today = str(is_uploaded_today(last_file))
except KeyError:
continue
data_table.append([bucket, last_file_name, last_file_datetime, was_last_file_today])
print(tabulate(data_table, headers=["Bucket", "File", "Last Modified", "Pass"]))
if __name__ == '__main__':
main()
Full Code
import boto3
from datetime import datetime, timezone
#import tabulate to make it print nicer
from tabulate import tabulate
REGION_NAME = 'eu-west-3'
SECS_IN_DAY = 86400
BUCKETS = ['04-prod-bb-chat-2868b1d6c06b', '04-prod-bb-email-b3b3503572be', '04-prod-smtp-archive-02269a7c9af5', '04-prod-trchat-ef96c2c80017']
#login
s3 = boto3.resource(
service_name='s3',
region_name='eu-west-3',
aws_access_key_id='ASIAQON5W62NMFAGDRX5',
aws_secret_access_key='BeuQX+jbpJav/wAbYKJ2h3Ribfbx+O+SLLiY6SaD'
)
def get_last_modified_file(bucket_name):
client = boto3.client('s3', region_name=REGION_NAME)
response = client.list_objects_v2(Bucket=bucket_name)
last_object = sorted(response['Contents'], key=lambda item: item['LastModified'])[-1]
return last_object
def is_uploaded_today(last_object):
last_modified_datetime = last_object['LastModified']
current_date = datetime.now(timezone.utc)
ts = (current_date - last_modified_datetime).total_seconds()
if ts < SECS_IN_DAY:
return True
else:
return False
def get_date_from_key(name):
pass
def main():
print(" ========== S3 Bucket Dashboard ======")
data_table = []
for bucket in BUCKETS:
try:
last_file = get_last_modified_file(bucket)
last_file_name = last_file['Key']
last_file_datetime = last_file['LastModified'].strftime("%m/%d/%Y: %H:%M:%S")
was_last_file_today = str(is_uploaded_today(last_file))
except KeyError:
continue
data_table.append([bucket, last_file_name, last_file_datetime, was_last_file_today])
print(tabulate(data_table, headers=["Bucket", "File", "Last Modified", "Pass"]))
if __name__ == '__main__':
main()