""" Base Processing Accounting Work Flow ===================================== Base Work Flow class that processing accounting 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 import boto3 from botocore.config import Config from garcon import activity import raven import sentry_sdk logger = logging.getLogger('garcon') class FlowBase(object): """Base Garcon Workflow that other Flows can extend """ def __init__(self, flow_name, version, domain=None): """Initializes Flow Initializes the Flow's: domain - 'prod_swf_processing_accounting' or 'dev name - '_processing_accounting' Args: flow_name (str): name of the work flow. 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(domain) self.name = generate_workflow_name(flow_name, domain) self.version = version self.client = boto3.client('swf', config=Config( connect_timeout=60, region_name='us-east-1', read_timeout=180, retries={'max_attempts': 2})) self.create = activity.create( self.client, 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() # client grabs new sentry dsn from SENTRY_DSN_NEW environment variable sentry_dsn_new = os.environ.get('SENTRY_DSN_NEW') if sentry_dsn_new: sentry_sdk.init( sentry_dsn_new, ) sentry_sdk.capture_exception(exception) if isinstance(actor, activity.Activity): actor.logger.error(exception) else: logger.error(exception) def workflow_id(self, initial_context=None): """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): raise NotImplementedError() def generate_workflow_name(workflow_name, domain=None): """Generate workflow name To play nice with non-class based Flows Args: workflow_name (str): name of the work flow. domain (str): SWF domain name Returns: str: SWF name in the format of '_processing_accounting' """ assert workflow_name.islower() and ' ' not in workflow_name, ( 'feed name must be lower case with no spaces') if domain: return workflow_name return '{workflow_name}_processing-accounting'.format( workflow_name=workflow_name) def get_domain(domain=None): """Get workflow domain To play nice with non-class based Flows Args: domain (str): SWF domain name Returns: str: SWF domain the workflow should use """ if domain: return domain return '{}_swf_processing_accounting'.format(os.getenv('Environment'))