""" Base Aggregation Workflow class. Base class that new flows should extend. Standardizes how flow domain, name, and workflow_id are generated and adds default on_exception method. Flows should override the decider method. """ import datetime import logging import os from garcon import activity import raven from analytics_aggregation import base_config PROD_DOMAIN = 'prod_analytics_aggregation' QA_DOMAIN = 'qa_analytics_aggregation' TEST_DOMAIN = 'test' SHADOW_DOMAIN = 'shadow_swf_feed_ingestion' DEV_DOMAIN = os.environ.get('DEV_DOMAIN', 'dev') logger = logging.getLogger('garcon') class FlowBase(object): """Base Garcon Workflow that other Flows can extend.""" def __init__(self, feed_name, version): """Initialize Flow. Initializes the Flow's: domain - 'prod_analytics_aggregation' or 'dev feed_name - name of feed used in the status table name - '_analytics_aggregation' ex. spotify_sos_analytics_aggregation (this is the workflow name as opposed to the feed name) Args: feed_name (str): Name of the feed the workflow is processing, with underscores replacing spaces. ex. 'spotify_sos' version (str): Version of the workflow type to filter on. should be a 'x.x' float (ex. 1.0, 1.1, etc.) """ self.domain = get_domain() self.name = generate_feed_name(feed_name) # name = workflow_name self.feed_name = feed_name # feed name verified in generate_feed_name self.version = version self.create = activity.create( self.domain, self.name, version=self.version, on_exception=self.on_exception) def on_exception(self, actor, exception): """Capture an exception that has occurred in the application. Args: actor (ActivityWorker, DeciderWorker): The actor that has received the exception. exception (Exception): The exception to capture. """ # client grabs sentry dns from SENTRY_DSN environment variable if os.environ.get('SENTRY_DSN'): client = raven.Client() client.captureException() if isinstance(actor, activity.Activity): actor.logger.error(exception, exc_info=True) else: logger.error(exception, exc_info=True) def workflow_id(self, initial_context): """Generate workflow id. Assumes a unique workflow is defined by it's context date and defaults to today's date. If that is not the case this method should be overridden. Args: initial_context (dict): The initial context for the flow. Returns: str: A unique identifier for a workflow being executed In the forms of '-YYYY-MM-DD', where YYYY-MM-DD is the context date or, if none passed the current date. """ if 'context_date' not in initial_context: date = datetime.datetime.today().strftime('%Y-%m-%d') else: date = initial_context['context_date'] return '{flow_name}-{date}'.format(flow_name=self.name, date=date) def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ raise NotImplementedError() def generate_feed_name(feed_name): """Generate workflow name. Args: feed_name (str): Name of the feed the work flow is processing, must be lower case and have spaces replaced with underscores. ex. 'spotify_sos' Returns: str: SWF name in the format of '_analytics_aggregation'. """ assert feed_name.islower() and ' ' not in feed_name, ( 'feed name must be lower case with no spaces') return '{feed_name}_analytics_aggregation'.format(feed_name=feed_name) def get_domain(): """Get workflow domain. Returns: str: SWF domain the workflow should use. """ if base_config.environment == 'prod': return PROD_DOMAIN elif base_config.environment == 'test': return TEST_DOMAIN elif base_config.environment == 'qa': return QA_DOMAIN elif base_config.environment == 'shadow': return SHADOW_DOMAIN else: return DEV_DOMAIN