"""Lambda util shared module.""" import csv from sosmodels import consts as sos_consts import consts # noqa DB_FIELD_PATTERN = 'transactiontype_{tid}__{metric}' csv.register_dialect( 'escaped', delimiter=',', quotechar='"', escapechar='\\', quoting=csv.QUOTE_ALL) def totals_as_dict(item_data, transactiontype_id): """Generate total metrics dict by transaction type. Args: item_data (dict): Representation transactiontype_id (int): Transaction type id value. Returns: dict: values for all total metrics. """ totals = { metric: int(item_data[DB_FIELD_PATTERN.format( tid=transactiontype_id, metric=metric)]) for metric in sos_consts.TOTAL_METRICS} return totals def extract_triggered_key(lambda_event): """Get S3 bucket and key from lambda event. Args: lambda_event (dict): AWS Lambda event meta data object. Returns: (str, str): Tuple of bucket and key names. """ record = lambda_event['Records'][0]['s3'] return record['bucket']['name'], record['object']['key'] def object_is_file(s3_object): """Check if passed object is file (and not folder). Args: s3_object (str): S3 object suffix (file or folder). Returns: bool: True if object is file, otherwise False. """ return s3_object[-1] != '/' def identify_retailer(s3_key_url): """Identify store by key path. Args: s3_key_url (str): S3 key path. """ for store in consts.AVAILABLE_STORES: if store in s3_key_url: return store def identify_totals_type(s3_key_url): """Identify totals_type by key path. Args: s3_key_url (str): S3 key path. """ for totals_type in consts.AVAILABLE_TOTALS_TYPES: if totals_type in s3_key_url: return totals_type