"""ActivityDetectorFeed class and related status, finder methods.""" from feed_status import util from feed_status.models.orm import feed_activity_detector_status ACTIVITY_DETECTOR_STATUS_PROCESSED = 'processed' ACTIVITY_DETECTOR_STATUS_PROCESSED_NOTIF_SENT = 'processed_notif_sent' ACTIVITY_DETECTOR_TYPE_DAILY = 'daily' class ActivityDetectorFeed: """ActivityDetectorFeed class to store configuration about each feed.""" def __init__( self, feed_id, feed_name, status_source_db, update_freq, feed_type=None, status=None, metrics=None, ): """Initialize class with attributes.""" self.feed_id = feed_id self.feed_name = feed_name self.status_source_db = status_source_db self.update_freq = update_freq # todo: move feed_type to feed monitor class self.feed_type = feed_type self.status = status self.metrics = metrics def match(self, **kwargs): """Whether instance of AD Feed contains provided attributes. Args: kwargs: keys/values to match on. Returns: boolean: True if contains all provided attributes, False otherwise. """ return all( getattr(self, key, None) == val for (key, val) in kwargs.items()) active_feeds = [ ActivityDetectorFeed( feed_id='chartmetric_spike_detector', feed_name='Chartmetric Spike Detector', feed_type=ACTIVITY_DETECTOR_TYPE_DAILY, status_source_db='feed_status', update_freq=1, ), ] def get_active_feeds(): """Retrieve active feeds. Returns: list: array of ActivityDetectorFeed configuration instances. """ return active_feeds def get_activity_detector_feed_history(feed, from_date, to_date): """Retrieve outgoing history for feed within a date range. Args: feed (OutgoingFeed): OutgoingFeed instance. from_date (date): start date of interest. to_date (date): end date of history. Returns: dict: map of dates to feed status. """ status = util.generate_default_date_dict_range(from_date, to_date) if feed.status_source_db == 'feed_status': feed_history_list = ( feed_activity_detector_status.get_activity_detector_feed_history( feed.feed_id, from_date.isoformat(), to_date.isoformat(), ) ) for feed_history in feed_history_list: status[feed_history['date']] = feed_history['status'].lower() return status