import os
from datetime import datetime, date
import logging
import boto3
from google.oauth2.service_account import Credentials
from google.cloud import storage
from pymorri.exceptions import MoriException
from pymorri.util import ResponseBuilder
from constants import DATADOG_APP_NAME, APP_NAME, \
DD_CHECK_METRIC_NAME, OCEAN_SSM_KEY
from pymorri.util.application_dependencies import ApplicationDependencies
from pymorri.util.general_utils import build_response_or_throw
from pymorri.util.datadog_utils import post_datadog_event_to_sns
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
[docs]def get_application_dependencies():
"""
Returns application_dependencies for a given config_file and ssm_param_store_name
Args:
Returns:
application_dependencies (class): object containing config and ssm values
Raises:
MoriException: if there is any issues with config and ssm name
"""
application_dependencies = ApplicationDependencies(config_file=f"{APP_NAME}.json",
ssm_param_store_name=OCEAN_SSM_KEY)
return application_dependencies
[docs]def get_s3_client():
"""
Returns s3 client
Args:
Returns:
s3_client (class): s3 client
"""
return boto3.client('s3')
[docs]def get_todays_s3_source_filename(application_configuration):
"""
Returns file name to be read from Kevin's account
Args:
application_configuration (dict): contains config values
Returns:
s3_file_name (str): file name
Raises:
Exception: if there is any issue with file read
"""
s3_bucket = application_configuration.get("S3_BUCKET")
source_file_name_prefix_1 = application_configuration.get("S3_SOURCE_FILE_PREFIX")
source_file_name_prefix_2 = datetime.now().strftime("%Y%m%d")
source_file_prefix = f"{source_file_name_prefix_1}{source_file_name_prefix_2}"
logger.info(f"getting s3 file_name from bucket {s3_bucket} and prefix {source_file_prefix}")
suffix = ".csv"
s3 = get_s3_client()
response = s3.list_objects_v2(Bucket=s3_bucket, Prefix=source_file_prefix)
logger.info(f"response {response}")
if response.get("KeyCount") == 0:
logger.error(f"file with prefix {source_file_prefix} not found in Kevin account S3 bucket")
message = "file not found in Kevin account S3 bucket"
raise MoriException(http_status_code=500, detail=message)
if response.get("KeyCount") > 1:
logger.error(f"Multiple files with prefix {source_file_prefix} found in kevin account")
message = "More than 1 file found in Kevin account S3 bucket"
raise MoriException(http_status_code=500, detail=message)
s3_file_name = None
for file_item in response.get('Contents'):
if file_item.get('Key').endswith(suffix):
print(f"filename is {file_item.get('Key')}")
s3_file_name = file_item.get('Key')
logger.info(f"s3_file_name {s3_file_name}")
if s3_file_name is None:
raise Exception("More than 1 file found in Kevin's account S3 bucket")
return s3_file_name
[docs]def get_file_from_s3(s3_bucket, source_file):
"""
Returns file data read from a file in s3 bucket and it's pending folder
Args:
s3_bucket (str): Name of s3 bucket
source_file (str): Name of s3 source file
Returns:
s3_client (stream): streaming object
Raises:
Exception: if there is any issue with file read
"""
s3_client = get_s3_client()
try:
data = s3_client.get_object(Bucket=s3_bucket, Key=source_file)
except Exception as exc:
logging.error(exc)
return None
serialized_object = data['Body'].read()
return serialized_object.decode('utf-8')
[docs]def put_file_into_s3(s3_bucket, target_file, serialized_data):
"""
Returns response after uploading file to s3 bucket
Args:
s3_bucket (str): Name of s3 bucket
target_file (str): Name of s3 target file
serialized_data (byte): utf-8 encoded bytes
Returns:
put_response (dict): response from s3 put file operation
Raises:
Exception: if there is any issue with file put
"""
s3_client = get_s3_client()
put_response = s3_client.put_object(Bucket=s3_bucket, Key=target_file, Body=serialized_data)
s3 = boto3.resource('s3')
file_object = s3.Bucket(s3_bucket).Object(target_file)
file_object.Acl().put(ACL='bucket-owner-full-control')
return put_response
[docs]def delete_file_from_s3(s3_bucket, source_file):
"""
Returns response after deleteing file from s3 bucket
Args:
s3_bucket (str): Name of s3 bucket
source_file (str): Name of s3 target file
Returns:
delete_response (dict): response from s3 delete file operation
Raises:
Exception: if there is any issue with file put
"""
s3_client = get_s3_client()
return s3_client.delete_object(Bucket=s3_bucket, Key=source_file)
[docs]def get_colleague_data_from_s3(application_configuration, todays_source_file_name):
"""
Returns file contents reaD from s3 bucket
Args:
application_configuration (dict): contains config values
Returns:
customer_info_buffer (str): file contents from s3 file
Raises:
Exception: if there is any issue with file read
"""
s3_bucket = application_configuration.get("S3_BUCKET")
archive_file_name = todays_source_file_name.replace("pending", "archive")
logger.info(f"archive_file_name... {archive_file_name}")
customer_info_buffer = get_file_from_s3(s3_bucket, todays_source_file_name)
if customer_info_buffer is None:
return customer_info_buffer
serialized_data = customer_info_buffer.encode('utf-8')
put_response = put_file_into_s3(s3_bucket, archive_file_name, serialized_data)
if put_response.get('ResponseMetadata').get('HTTPStatusCode') == 200:
delete_response = delete_file_from_s3(s3_bucket, todays_source_file_name)
if delete_response.get('ResponseMetadata').get('HTTPStatusCode') != 204:
raise Exception(f"Error moving source file with prefix {todays_source_file_name}")
else:
raise Exception(f"Error deleting source file with prefix {todays_source_file_name}")
return customer_info_buffer
[docs]def get_ocean_colleague_data(table_key):
"""
Returns all records from colleaaue_discount table
Args:
Returns:
customer_record_list (list): customer rercords from dynamodb table
"""
ocean_colleague_discount_table = os.environ[table_key]
logger.info(f"ocean table is {ocean_colleague_discount_table}")
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(ocean_colleague_discount_table)
response = table.scan()
data = response['Items']
while 'LastEvaluatedKey' in response:
response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
data.extend(response['Items'])
customer_record_list = list()
for customer_record in data:
customer_record_list.append(customer_record)
logger.info(f"get_ocean_colleague_data returns {len(customer_record_list)} records")
return customer_record_list
[docs]def get_storage_client(application_dependencies):
"""
Returns Google Cloud Storage client object
Args:
- application_dependencies (class) - application_dependencies object
Raises:
Exception: MoriException containing details about exception
"""
google_application_credentials = application_dependencies.get_ssm_params().get('GOOGLE_CREDENTIALS')
if google_application_credentials.get("GOOGLE_CREDENTIALS"):
google_application_credentials = google_application_credentials.get("GOOGLE_CREDENTIALS")
if google_application_credentials:
logging.info("GOOGLE_APPLICATION_CREDENTIALS var found in ssm")
try:
credentials = Credentials.from_service_account_info(google_application_credentials)
except Exception as e:
error_msg = f"GOOGLE_APPLICATION_CREDENTIALS is not valid json, " \
"the service account file should be stored as a string with valid json structure." \
"\n Error: {str(e)}"
logging.error(error_msg)
post_datadog_event_to_sns(application_name=DATADOG_APP_NAME, metric_type="check",
metric_name=DD_CHECK_METRIC_NAME, metric_value="2")
raise Exception(error_msg)
# Make sure gcp key is valid
try:
storage_client = storage.Client(google_application_credentials["project_id"], credentials)
except Exception as e:
error_msg = f"GOOGLE_APPLICATION_CREDENTIALS is not valid, " \
"it is missing one or more of the required properties" \
"\n Missing: {str(e)}"
logging.fatal(error_msg)
post_datadog_event_to_sns(application_name=DATADOG_APP_NAME, metric_type="check",
metric_name=DD_CHECK_METRIC_NAME, metric_value="2")
raise Exception(error_msg)
return storage_client
else:
error_msg = "GOOGLE_APPLICATION_CREDENTIALS not found in ssm"
logging.fatal(error_msg)
post_datadog_event_to_sns(application_name=DATADOG_APP_NAME, metric_type="check",
metric_name=DD_CHECK_METRIC_NAME, metric_value="2")
raise Exception(error_msg)
[docs]def upload_file_to_storage(application_dependencies, gcp_storage_file_name, file_content):
"""
Uploads ocean colleague discount data to Google cloud storage
Args:
- application_dependencies (class) - application_dependencies object
Returns:
upload_file_flag (Bool): showing file upload operation status
"""
storage_client = get_storage_client(application_dependencies)
gcp_storage_bucket_name = application_dependencies.get_config().get("GCP_BUCKET")
logger.info(f"Uploading file {gcp_storage_file_name} into bucket {gcp_storage_bucket_name}")
gcp_file_prefix = application_dependencies.get_config().get("GCP_UPLOAD_FILE_NAME_PREFIX")
logger.info(f"gcp_file_prefix {gcp_file_prefix}")
gcp_storage_full_file_name = f"{gcp_file_prefix}{gcp_storage_file_name}"
logger.info(f"gcp_storage_full_file_name {gcp_storage_full_file_name}")
bucket = storage_client.get_bucket(gcp_storage_bucket_name)
blob = bucket.blob(gcp_storage_full_file_name)
blob.upload_from_string(file_content, content_type="text/csv")
logger.info("File upload to GCP complete....... ")
return True
[docs]def process_ocean_friend_and_family(application_dependencies):
"""
This method will used for processing ocean_friend_and_family customer to upload file into ocean
GCP bucket
"""
logger.info("Process start for ocean_friend_and_family_discount")
customer_info_list = list()
table_records = get_ocean_colleague_data('FRIENDS_AND_FAMILY_DISCOUNT_TABLE')
if len(table_records) > 0:
for item in table_records:
customer_id = item.get("customerID")
customer_info = f"{customer_id}"
customer_info_list.append(customer_info)
if len(customer_info_list) > 0:
logger.info(f"Number of records to process for friends and family discount table {len(customer_info_list)}")
today_date = date.today().strftime("%Y%m%d")
now_time = date.today().strftime("%h%M")
gcp_file_name = f"familyandfriends-discounts-{today_date}-{now_time}.csv"
customer_info_from_table = '\n'.join(customer_info_list)
upload_file_to_storage(application_dependencies, gcp_file_name, customer_info_from_table)
else:
logger.info(f"There is no record in friends and family discount table")
logger.info("Process end for ocean_friend_and_family_discount")
[docs]@build_response_or_throw
def lambda_handler(event, context):
"""
Lambda is called on a daily schedule and queries ocean_colleague_discount table and fetches all records,
then prepares customer_info data payload and uploads the same to Google Cloud Storage
Args:
- event (dict) - empty dict, but cannot be null or None
Example:
.. code-block:: python
event = {}
Output:
- resp (dict):
Example:
.. code-block:: python
resp = {
'statusCode': 200,
'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
'body': '{
"status": "success",
"message": "Ocean file uploaded to Ocado successfully",
"body": []
}'
}
"""
# update event in order this function could send STATUS data to DataDog API
# from build_response_or_throw
event.update({
"DATADOG_APP_NAME": DATADOG_APP_NAME,
"DD_CHECK_METRIC_NAME": DD_CHECK_METRIC_NAME,
})
logger.info(f"input event: {event}")
application_dependencies = get_application_dependencies()
process_ocean_friend_and_family(application_dependencies)
logger.info("Process start for ocean_colleague_discount")
customer_info_list = list()
table_records = get_ocean_colleague_data('COLLEAGUE_DISCOUNT_TABLE')
customer_info_from_table = None
if len(table_records) > 0:
for item in table_records:
customer_id = item.get("customerID")
customer_info = f"{customer_id}"
customer_info_list.append(customer_info)
customer_info_from_table = '\n'.join(customer_info_list)
source_file_name = get_todays_s3_source_filename(application_dependencies.get_config())
colleague_data_from_s3 = get_colleague_data_from_s3(application_dependencies.get_config(), source_file_name)
if colleague_data_from_s3 is not None:
logger.info(f"number of colleague_data_from_s3 {len(colleague_data_from_s3)}")
if colleague_data_from_s3 is None and len(table_records) == 0:
event['response_message'] = 'Colleague Discount file not uploaded to Ocean as data not found for today'
logger.warning(event['response_message'])
post_datadog_event_to_sns(application_name=DATADOG_APP_NAME, metric_type="check",
metric_name="colleague_discount_check", metric_value="1")
return event
if colleague_data_from_s3 is not None:
if len(table_records) > 0:
# this is to strip last blank line
only_data_lines_from_s3 = "".join(
[s for s in colleague_data_from_s3.strip().splitlines(True) if s.strip("\r\n").strip()])
customer_info_daily_journal = only_data_lines_from_s3 + '\n' + customer_info_from_table
else:
customer_info_daily_journal = colleague_data_from_s3
else:
customer_info_daily_journal = customer_info_from_table
filename_token_list = source_file_name.split("/")
gcp_file_name = filename_token_list[len(filename_token_list)-1]
upload_file_to_storage(application_dependencies, gcp_file_name, customer_info_daily_journal)
logger.info("Process end for ocean_colleague_discount")
event['response_message'] = 'Ocean file uploaded to Ocado successfully'
return event
if __name__ == "__main__":
event = {}
response = lambda_handler(event, '')
logger.info(f"{response}")