"""Populate PASSTHRU_PROCESSED table. To run type: MYSQL_HOST= MYSQL_USER= \ MYSQL_PASSWORD= AF_MYSQL_DATABASE= python populate_passthru_processed.py \ --offset=0 --batch_size=10000 --period_id=235 """ import argparse import decimal import logging import sys import time from ows_accounting import config from ows_accounting.connectors import accountingflat from ows_accounting.models import passthru_processed as ptp_model from ows_accounting.models.sql import populate_passthru_processed as ptp_sql logger = logging.getLogger(__name__) EXCHANGE_RATE_BY_UPC = {} UPCS = set() def select_passthru_sales_detail(period_id, row_count, offset): """Execute select query for getting passthru_sales_details entries. Args: period_id (int): ID of period. row_count (int): Row count. offset (int): Offset. Returns: list: List of result values in tuples. """ with accountingflat.db_session() as session: result = session.execute( ptp_sql.SELECT_PT_TEMP_DATA_SQL.format( period_id=period_id, row_count=row_count, offset=offset)) all_rows = list(result.fetchall()) return all_rows def select_vendors(upcs): """Execute select query for getting vendors ids by upc. Args: upcs (list): List of upcs. Returns: list: List of tuples with vendor id and upc. """ upcs_string = ', '.join([str(u) for u in upcs]) with accountingflat.db_session() as session: result = session.execute( ptp_sql.SELECT_VENDORS_SQL.format(upcs=upcs_string)) all_rows = list(result.fetchall()) return all_rows def select_exchange_rate(period_id, vendors_ids): """Execute select query for getting exchange_rate by period and vendor ids. Args: period_id (int): ID of period. vendors_ids (list): List of vendors ids. Returns: list: List of tuples with exchange rate values and vendor id. """ vendors_ids_string = ', '.join([str(v) for v in vendors_ids]) with accountingflat.db_session() as session: result = session.execute( ptp_sql.SELECT_VENDORS_EXCHANGE_RATE_SQL.format( vendors_ids=vendors_ids_string, period_id=period_id)) all_rows = list(result.fetchall()) return all_rows def calculate_amount(exchange_rate, original_price, qty): """Calculate final amount by exchange_rate, original_price, qty. Args: exchange_rate (int): Exchange rate to expected currency. original_price (list): Price for transaction in USD. qty (int): Transactions quantity. Returns: double: Amount. """ return round(exchange_rate * original_price * qty, 6) def insert_passthru_processed_entries(entries): """Insert prepared entries to passthru_processed table. Args: entries (list): List of prepared passthru_processed entries. """ with accountingflat.db_session() as session: session.execute( ptp_model.PassthruProcessed.__table__.insert(), entries) def upload_exchange_rate_to_cache(period_id, upcs): """Upload exchange rate to cache for given vendors by upcs. Args: period_id (int): ID of period. upcs (list): List of upcs. """ UPCS.update(upcs) vendors = select_vendors(upcs) vendor_id_to_upcs = {} for vid, upc in vendors: if upc not in EXCHANGE_RATE_BY_UPC: EXCHANGE_RATE_BY_UPC[upc] = (vid, 1, decimal.Decimal(1)) vendor_id_to_upcs.setdefault(vid, []).append(upc) exchange_rates = select_exchange_rate(period_id, vendor_id_to_upcs.keys()) for ex in exchange_rates: for upc in vendor_id_to_upcs[ex[0]]: EXCHANGE_RATE_BY_UPC[upc] = ex def populate_passthru_processed_table(period_id, passthru_sales_details): """Generate entries and populate passthru_processed table. Args: period_id (int): ID of period. passthru_sales_details (list): List of passthru_sales_detail entries. """ passthru_processed = [] nonexistent_vendors_for_provided_upc = [] for psd in passthru_sales_details: try: entry = { 'statement_detail_id': psd[0], 'vendor_id': EXCHANGE_RATE_BY_UPC[psd[2]][0], 'period_id': period_id, 'dms_customer_id': None, 'date': psd[1], 'upc': psd[2], 'cd': psd[3], 'track_id': psd[4], 'qty': psd[5], 'amount': calculate_amount( EXCHANGE_RATE_BY_UPC[psd[2]][2], psd[7], psd[5]), 'trans_type': psd[6], 'payout_currency_id': EXCHANGE_RATE_BY_UPC[psd[2]][1], 'fx_adjusted_exchange_rate': EXCHANGE_RATE_BY_UPC[psd[2]][2], 'original_price': psd[7] } passthru_processed.append(entry) except KeyError as e: if e.args[0] == psd[2]: nonexistent_vendors_for_provided_upc.append(psd) else: logger.error('Unexpected error:', exc_info=True) sys.exit(1) if len(nonexistent_vendors_for_provided_upc) > 0: logger.error( 'There are {} non processed entries because of not found' ' vendor'.format(len(nonexistent_vendors_for_provided_upc)), exc_info=True) insert_passthru_processed_entries(passthru_processed) def execute_batch_update(period_id, offset, row_count): """Execute batch update loop. Args: period_id (int): ID of period. row_count (int): Row count. offset (int): Offset. """ entry_counter = row_count start_iteration_time = time.time() passthru_sales_details = select_passthru_sales_detail( period_id, row_count, offset) while len(passthru_sales_details) > 0: upcs = {v[2] for v in passthru_sales_details} if not UPCS.issuperset(upcs): upload_exchange_rate_to_cache(period_id, list(upcs-UPCS)) populate_passthru_processed_table(period_id, passthru_sales_details) offset += row_count passthru_sales_details = select_passthru_sales_detail( period_id, row_count, offset) logger.info('Execution time of the one iteration: %s seconds' % ( time.time() - start_iteration_time)) logger.info('Inserted {} entries'.format(entry_counter)) entry_counter += row_count start_iteration_time = time.time() def setup_parser(): """Set args parser.""" new_parser = argparse.ArgumentParser() new_parser.add_argument( '--offset', required=False, default=0, type=int, dest='offset', help='Initial offset') new_parser.add_argument( '--batch_size', required=False, default=200000, type=int, dest='row_count', help='Size of the batch') new_parser.add_argument( '--period_id', required=False, default=235, type=int, help='An id of the period for processing particular transactions') return new_parser if __name__ == '__main__': # parse cli arguments parser = setup_parser() args = parser.parse_args() # logger setup logger.setLevel(logging.INFO) handler = logging.FileHandler(config.LOG_FILENAME) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) try: execute_batch_update( period_id=args.period_id, offset=args.offset, row_count=args.row_count ) except Exception: logger.error("Unexpected error: ", exc_info=True) sys.exit(1)