"""A mixin that implements generic methods.""" import datetime class FlowBaseMixin: """A mixin that adds generic methods to a flow class.""" def workflow_id(self, initial_context): """Generate workflow id. 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) @staticmethod 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. Returns: str: SWF name in the format of 'swf_'. """ assert feed_name.islower() and ' ' not in feed_name, ( 'feed name must be lower case with no spaces') return 'swf_{feed_name}'.format(feed_name=feed_name)