Load アクションをオーバーライドする#

import sys,os
from logging import getLogger

from forepaas.core.settings import PARAMS
from forepaas.actions.aggregate.fpaggregate import aggregate

logger = getLogger(__name__)

def override_aggregate(event):
    
    try:
    	# Set filename. could be get through segmentation or parameters
	filename = "sales_2020-01-01.csv"
        
        # Override source filename
        path = "/".join(PARAMS["load_from"][0]["source"].split("/")[:-1]) + "/" + filename
        PARAMS["load_from"][0]["source"] = path

        # override source_date in schema
        source_date = filename.replace("sales_","").replace(".csv","")
        PARAMS["schema"]["source_date"] = {"type":"replace", "value": str(source_date)}
        logger.info(f"{filename} {source_date}")
        
        # executing the aggregate (=load) action, it will use the overriden PARAMS.
        aggregate(event)

    except Exception as err: 
        message="error on file:{}: {} L:{}".format(filename,err,sys.exc_info()[2].tb_lineno)
        raise Exception(message)