def _unpack_fields(cls, session, schedule,
                       args=None, kwargs=None, relative=None, options=None,
                       **entry):
        """

        **entry sample:

            {'task': 'celery.backend_cleanup',
             'schedule': <crontab: 0 4 * * * (m/h/d/dM/MY)>,
             'options': {'expires': 43200}}

        """
        model_schedule, model_field = cls.to_model_schedule(session, schedule)
        entry.update(
            # the model_id which to relationship
            {model_field + '_id': model_schedule.id},
            args=dumps(args or []),
            kwargs=dumps(kwargs or {}),
            **cls._unpack_options(**options or {})
        )
        return entry 

def update_from_dict(self, mapping):
        s = {}
        for name, entry_fields in items(mapping):
            # {'task': 'celery.backend_cleanup',
            #  'schedule': schedules.crontab('0', '4', '*'),
            #  'options': {'expires': 43200}}
            try:
                entry = self.Entry.from_entry(
                    name, Session=self.Session, app=self.app,
                    **entry_fields)
                if entry.model.enabled:
                    s[name] = entry
            except Exception as exc:
                logger.error(ADD_ENTRY_ERROR, name, exc, entry_fields)

        # update self.schedule
        self.schedule.update(s) 

def decode_schedule(obj):
    if obj is None:
        return None

    _type = obj['__type__']
    value = obj['__value__']

    if _type == 'datetime':
        return decode_datetime(value)
    elif _type == 'crontab':
        return crontab(*value.split('\t'))
    elif _type == 'solar':
        return solar(**value)
    elif _type == 'schedule':
        return schedule(**value)
    else:
        raise NotImplementedError(
            'Cannot deserialize schedule %(type)s type' % {
                'type': _type
            }
        ) 

def test_parsing_dict(self):
        task = PeriodicTask(**self.task_dict)
        assert task.name == 'bruteforce_ssh.AlertBruteforceSsh'
        assert task.task == 'bruteforce_ssh.AlertBruteforceSsh'
        assert task.args == []
        assert task.kwargs == {}
        assert task.enabled is True
        assert task.queue is None
        assert task.exchange is None
        assert task.routing_key is None
        assert task.last_run_at is None
        assert task.run_immediately is False
        assert task.total_run_count is 0
        assert task.schedule_type == 'crontab'
        assert task.schedule_str == '0 5 * * *'
        assert isinstance(task.celery_schedule, Crontab) is True
        assert isinstance(task.schedule, celery_crontab) is True 

def update_celerybeat_schedule(celerybeat_schedule_settings, figures_env_tokens):
    """
    Figures pipeline job schedule configuration in CELERYBEAT_SCHEDULE.

    Daily metrics pipeline scheduler is on by default
    Course MAU metrics pipeline scheduler is off by default

    TODO: Language improvement: Change the "IMPORT" to "CAPTURE" or "EXTRACT"
    """
    if figures_env_tokens.get('ENABLE_DAILY_METRICS_IMPORT', True):
        celerybeat_schedule_settings['figures-populate-daily-metrics'] = {
            'task': 'figures.tasks.populate_daily_metrics',
            'schedule': crontab(
                hour=figures_env_tokens.get('DAILY_METRICS_IMPORT_HOUR', 2),
                minute=figures_env_tokens.get('DAILY_METRICS_IMPORT_MINUTE', 0),
                ),
            }

    if figures_env_tokens.get('ENABLE_DAILY_MAU_IMPORT', False):
        celerybeat_schedule_settings['figures-daily-mau'] = {
            'task': 'figures.tasks.populate_all_mau',
            'schedule': crontab(
                hour=figures_env_tokens.get('DAILY_MAU_IMPORT_HOUR', 0),
                minute=figures_env_tokens.get('DAILY_MAU_IMPORT_MINUTE', 0),
                ),
            }

    if figures_env_tokens.get('ENABLE_FIGURES_MONTHLY_METRICS', False):
        celerybeat_schedule_settings['figures-monthly-metrics'] = {
            'task': 'figures.tasks.run_figures_monthly_metrics',
            'schedule': crontab(0, 0, day_of_month=1),
            } 

def __str__(self):
        fmt = '{0.name}: {0.crontab}'
        return fmt.format(self) 

def schedule(self):
        if self.crontab:
            return self.crontab.schedule
        if self.interval:
            return self.interval.schedule 

def install_default_entries(self, data):
        entries = {}
        if self.app.conf.CELERY_TASK_RESULT_EXPIRES:
            entries.setdefault(
                'celery.backend_cleanup', {
                    'task': 'celery.backend_cleanup',
                    'schedule': schedules.crontab('*/5', '*', '*'),
                    'options': {'expires': 12 * 3600},
                },
            )
        self.update_from_dict(entries) 

def create_app(debug=False):
    """Create an application context with blueprints."""
    app = Flask(__name__, static_folder='./resources')
    app.config['SECRET_KEY'] = 'RYVl4Fg3n1JLDaxWyr1m'
    app.config['MONGO_DBNAME'] = 'chirp'
    app.config['USERS_COLLECTION'] = 'accounts'
    app.config['MONITORS_COLLECTION'] = 'monitors'
    app.config['ARTICLES_COLLECTION'] = 'articles'
    app.config['GLOBAL_COLLECTION'] = 'global'
    login_manager.init_app(app)
    mongo.init_app(app)
    app.config.update(
        BROKER_TRANSPORT_OPTIONS={'visibility_timeout': 3600},
        CELERY_BROKER_URL='redis://localhost:6379',
        CELERY_RESULT_BACKEND='redis://localhost:6379',
        CELERYBEAT_SCHEDULE={
            # 'heartbeat': {
            #     'task': 'heartbeat',
            #     'schedule': crontab(minute='*')
            # },
            'process_all_rss': {
                'task': 'process_all_rss',
                'schedule': crontab(minute='*/15')
            }
        }
    )
    celery.conf.update(app.config)

    from .core import core as core_blueprint
    app.register_blueprint(core_blueprint)
    app.register_error_handler(404, page_not_found)
    app.register_error_handler(500, server_error)

    return app 

def schedule(self):
        return schedules.crontab(minute=self.minute,
                                 hour=self.hour,
                                 day_of_week=self.day_of_week,
                                 day_of_month=self.day_of_month,
                                 month_of_year=self.month_of_year) 

def schedule(self):
        if self.interval:
            return self.interval.schedule
        if self.crontab:
            return self.crontab.schedule 

def from_entry(cls, name, Session, app=None, **entry):
        """

        **entry sample:

            {'task': 'celery.backend_cleanup',
             'schedule': schedules.crontab('0', '4', '*'),
             'options': {'expires': 43200}}

        """
        session = Session()
        with session_cleanup(session):
            periodic_task = session.query(
                PeriodicTask).filter_by(name=name).first()
            if not periodic_task:
                periodic_task = PeriodicTask(name=name)
            temp = cls._unpack_fields(session, **entry)
            periodic_task.update(**temp)
            session.add(periodic_task)
            try:
                session.commit()
            except sqlalchemy.exc.IntegrityError as exc:
                logger.error(exc)
                session.rollback()
            except Exception as exc:
                logger.error(exc)
                session.rollback()
            res = cls(periodic_task, app=app, Session=Session, session=session)
            return res 

def install_default_entries(self, data):
        entries = {}
        if self.app.conf.result_expires:
            entries.setdefault(
                'celery.backend_cleanup', {
                    'task': 'celery.backend_cleanup',
                    'schedule': schedules.crontab('0', '4', '*'),
                    'options': {'expires': 12 * 3600},
                },
            )
        self.update_from_dict(entries) 

def __repr__(self):
        return """<crontab: {0._orig_minute} {0._orig_hour} \
{0._orig_day_of_week} {0._orig_day_of_month} \
{0._orig_month_of_year} (m/h/d/dM/MY), {0.tz}>""".format(self) 

def run_django_management_command(self, command, *args, **kwargs):
    management.call_command(command, *args, **kwargs)


# Configure periodic tasks here
# This task is not needed anymore (number of ground truth annotations per taxonomy node is real time:
# updated at each generation of ground truth)
# However this can serve as an example for possibly future periodic tasks.
# @app.on_after_configure.connect
# def setup_periodic_tasks(sender, **kwargs):
#     sender.add_periodic_task(
#         crontab(hour="*", minute="*/15"),
#         run_django_management_command.s('compute_priority_score_taxonomy_node'),
#         name='compute priority score taxonomy node'
#     ) 

def start_stacking_scheduler():
    logger.info('Entered entrypoint to celery beat scheduling')
    runtime_context = parse_args(settings)
    for site, entry in runtime_context.SCHEDULE_STACKING_CRON_ENTRIES.items():
        app.add_periodic_task(crontab(minute=entry['minute'], hour=entry['hour']),
                              schedule_calibration_stacking.s(site=site, runtime_context=vars(runtime_context)))

    beat = celery.bin.beat.beat(app=app)
    logger.info('Starting celery beat')
    beat.run() 

def get_schedule(self) -> Any:
        if self.type == "CRONTAB":
            return crontab(**self.crontab_kwargs)
        return float(self.schedule) 

def schedule(self):
        return schedules.crontab(minute=self.minute,
                                 hour=self.hour,
                                 day_of_week=self.day_of_week,
                                 day_of_month=self.day_of_month,
                                 month_of_year=self.month_of_year) 

def test_panoptes_celery_plugin_scheduler(self):

        celery_config = PanoptesCeleryConfig(u'test')
        panoptes_context = PanoptesContext(self.panoptes_test_conf_file)

        celery_instance = PanoptesCeleryInstance(panoptes_context, celery_config)
        celery_plugin_scheduler = PanoptesCeleryPluginScheduler(app=celery_instance.celery)

        new_schedule = dict()
        new_schedule[u'celery.backend_cleanup'] = {
            u'task': u'celery.backend_cleanup',
            u'schedule': crontab(u'0', u'4', u'*'),
            u'options': {u'expires': 12 * 3600}}
        new_schedule[u'test_task'] = {
            u'task': const.POLLING_PLUGIN_AGENT_MODULE_NAME,
            u'schedule': timedelta(seconds=60),
            u'args': (u"test_plugin", u"test"),
            u'last_run_at': datetime.utcfromtimestamp(DUMMY_TIME - 61),
            u'options': {
                u'expires': 60,
                u'time_limit': 120
            }
        }
        new_schedule[u'test_task_2'] = {
            u'task': const.POLLING_PLUGIN_AGENT_MODULE_NAME,
            u'schedule': timedelta(seconds=60),
            u'args': (u"test_plugin", u"test_2"),
            u'last_run_at': datetime.utcfromtimestamp(DUMMY_TIME - 1),
            u'options': {
                u'expires': 60,
                u'time_limit': 120
            }
        }

        celery_plugin_scheduler.update(celery_plugin_scheduler.logger, new_schedule)
        self.assertEqual(len(celery_plugin_scheduler.schedule), len(new_schedule))

        mock_producer = Mock()

        with patch(u'yahoo_panoptes.framework.celery_manager.PanoptesCeleryPluginScheduler.apply_entry',
                   return_value=None):
            with patch(u'yahoo_panoptes.framework.celery_manager.PanoptesCeleryPluginScheduler.producer',
                       mock_producer):
                self.assertIsNone(celery_plugin_scheduler._heap)
                self.assertEqual(celery_plugin_scheduler.tick(), 0)
                self.assertEqual(celery_plugin_scheduler.tick(), 0)
                assert celery_plugin_scheduler.tick() > 0 

def encode_schedule(value):
    if value is None:
        return None
    elif isinstance(value, datetime):
        return {
            '__type__': 'datetime',
            '__value__': encode_datetime(value)
        }
    elif isinstance(value, crontab):
        return {
            '__type__': 'crontab',
            '__value__': '%(minute)s\t%(hour)s\t%(day_of_week)s\t'
                         '%(day_of_month)s\t%(month_of_year)s' % {
                             'minute': value._orig_minute,
                             'hour': value._orig_hour,
                             'day_of_week': value._orig_day_of_week,
                             'day_of_month': value._orig_day_of_month,
                             'month_of_year': value._orig_month_of_year,
                         }
        }
    elif isinstance(value, solar):
        return {
            '__type__': 'solar',
            '__value__': {
                'event': value.event,
                'lat': value.lat,
                'lon': value.lon
            }
        }
    elif isinstance(value, schedule):
        return {
            '__type__': 'schedule',
            '__value__': {
                'run_every': value.run_every.total_seconds(),
                'relative': bool(value.relative),
            }
        }
    else:
        raise NotImplementedError(
            'Cannot serialize schedule %(type)s type' % {
                'type': type(value).__name__
            }
        ) 

def create_app(debug=False):
    """Create an application context with blueprints."""
    state = housekeeping()
    if not state:
        sys.exit(1)
    app = Flask(__name__, static_folder='./resources')
    app.config['SECRET_KEY'] = 'tRSn3mh2bY3@1$W2T9aQ'
    app.config['MONGO_DBNAME'] = 'netinfo'
    app.config['MONGO_HOST'] = 'localhost'
    app.config['ASNDB'] = None
    app.config['GEOIPDB'] = None
    app.config['DEBUG'] = debug
    muri = "mongodb://%s:27017/%s" % (app.config['MONGO_HOST'],
                                      app.config['MONGO_DBNAME'])
    app.config['MONGO_URI'] = muri
    mongo.init_app(app)
    app.config.update(
        CELERY_BROKER_URL='redis://localhost:6379',
        CELERY_RESULT_BACKEND='redis://localhost:6379',
        CELERYBEAT_SCHEDULE={
            'fetch-rib': {
                'task': 'fetch-rib',
                'schedule': crontab(minute='*/5')
            },
            'fetch-as-name': {
                'task': 'fetch-as-names',
                'schedule': crontab(hour="*/12")
            },
            'fetch-geo': {
                'task': 'fetch_geoip',
                'schedule': crontab(hour=7, minute=30, day_of_week=1)
            }
        }
    )
    celery.conf.update(app.config)

    config_file = '%s/resources/config.json' % APP_BASE
    if not os.path.exists(config_file):
        config = {'asn': {'last_rib_file': None, 'last_update': None},
                  'geoip': {'last_update': None}}
        json.dump(config, open(config_file, 'w'), indent=4)

    from .core import core as core_blueprint
    app.register_blueprint(core_blueprint)
    app.register_error_handler(404, page_not_found)
    app.register_error_handler(500, server_error)

    return app 

def load_and_register_alerts(self):
        existing_alert_schedules = self.fetch_schedule_dict()
        alert_schedules = {}
        for alert_name, params in ALERTS.items():
            # Register alerts in celery
            try:
                alert_tokens = alert_name.split(".")
                alert_module_name = alert_tokens[0]
                alert_classname = alert_tokens[-1]
                alert_module = import_module(alert_module_name)
                alert_class = getattr(alert_module, alert_classname)
                current_app.register_task(alert_class())
            except ImportError as e:
                logger.exception("Error importing {0}: {1}".format(alert_name, e))
                pass
            except Exception as e:
                logger.exception("Generic error registering {0}: {1}".format(alert_name, e))
                pass
            alert_schedule = {
                "name": alert_name,
                "task": alert_name,
                "enabled": True,
            }
            if 'args' in params:
                alert_schedule['args'] = params['args']
            if 'kwargs' in params:
                alert_schedule['kwargs'] = params['kwargs']

            if isinstance(params['schedule'], timedelta):
                alert_schedule['schedule_type'] = 'interval'
                alert_schedule['celery_schedule'] = {
                    "every": params['schedule'].total_seconds(),
                    "period": "seconds"
                }
            elif isinstance(params['schedule'], crontab):
                alert_schedule['schedule_type'] = 'crontab'
                alert_schedule['celery_schedule'] = {
                    "minute": params['schedule']._orig_minute,
                    "hour": params['schedule']._orig_hour,
                    "day_of_week": params['schedule']._orig_day_of_week,
                    "day_of_month": params['schedule']._orig_day_of_month,
                    "month_of_year": params['schedule']._orig_month_of_year,
                }

            if alert_name not in existing_alert_schedules:
                logger.debug("Inserting schedule for {0} into mongodb".format(alert_name))
                updated_alert_schedule = alert_schedule
            else:
                existing_schedule = existing_alert_schedules[alert_name]
                logger.debug("Updating existing schedule ({0}) with new information into mongodb".format(alert_name))
                existing_schedule['schedule_type'] = alert_schedule['schedule_type']
                existing_schedule['celery_schedule'] = alert_schedule['celery_schedule']
                updated_alert_schedule = existing_schedule

            alert_schedules[alert_name] = PeriodicTask(**updated_alert_schedule).to_dict()
        self.update_schedules(alert_schedules)