def from_entry(cls, name, session, skip_fields=('relative', 'options'), **entry):
        """
        PeriodicTask
        :param session:
        :param name:
        :param skip_fields:
        :param entry:
        :return:
        """
        fields = dict(entry)
        for skip_field in skip_fields:
            fields.pop(skip_field, None)
        schedule = fields.pop('schedule')
        model_schedule, model_field = cls.to_model_schedule(schedule, session)
        fields[model_field] = model_schedule
        fields['args'] = json.dumps(fields.get('args') or [])
        fields['kwargs'] = json.dumps(fields.get('kwargs') or {})
        model, _ = PeriodicTask.update_or_create(session, name=name, defaults=fields)
        cls.save_model(session, model)
        return cls(model) 

def from_schedule(cls, dbsession, schedule):
        spec = {'minute': schedule._orig_minute,
                'hour': schedule._orig_hour,
                'day_of_week': schedule._orig_day_of_week,
                'day_of_month': schedule._orig_day_of_month,
                'month_of_year': schedule._orig_month_of_year}
        try:
            query = dbsession.query(CrontabSchedule)
            query = query.filter_by(**spec)
            existing = query.one()
            return existing
        except NoResultFound:
            return cls(**spec)
        except MultipleResultsFound:
            query = dbsession.query(CrontabSchedule)
            query = query.filter_by(**spec)
            query.delete()
            dbsession.commit()
            return cls(**spec) 

def from_entry(cls, name, skip_fields=('relative', 'options'), **entry):
        options = entry.get('options') or {}
        fields = dict(entry)
        for skip_field in skip_fields:
            fields.pop(skip_field, None)
        schedule = fields.pop('schedule')
        model_schedule, model_field = cls.to_model_schedule(schedule)
        fields[model_field] = model_schedule
        fields['args'] = fields.get('args') or []
        fields['kwargs'] = fields.get('kwargs') or {}
        fields['queue'] = options.get('queue')
        fields['exchange'] = options.get('exchange')
        fields['routing_key'] = options.get('routing_key')

        query = dbsession.query(DatabaseSchedulerEntry)
        query = query.filter_by(name=name)
        db_entry = query.first()
        if db_entry is None:
            new_entry = DatabaseSchedulerEntry(**fields)
            new_entry.name = name
            dbsession.add(new_entry)
            dbsession.commit()
            db_entry = new_entry
        return cls(db_entry) 

def from_schedule(cls, session, schedule):
        spec = {
            'minute': schedule._orig_minute,
            'hour': schedule._orig_hour,
            'day_of_week': schedule._orig_day_of_week,
            'day_of_month': schedule._orig_day_of_month,
            'month_of_year': schedule._orig_month_of_year,
        }
        if schedule.tz:
            spec.update({
                'timezone': schedule.tz.zone
            })
        model = session.query(CrontabSchedule).filter_by(**spec).first()
        if not model:
            model = cls(**spec)
            session.add(model)
            session.commit()
        return model 

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 all_as_schedule(self):
        # TODO:
        session = self.Session()
        with session_cleanup(session):
            logger.debug('DatabaseScheduler: Fetching database schedule')
            # get all enabled PeriodicTask
            models = session.query(self.Model).filter_by(enabled=True).all()
            s = {}
            for model in models:
                try:
                    s[model.name] = self.Entry(model,
                                               app=self.app,
                                               Session=self.Session,
                                               session=session)
                except ValueError:
                    pass
            return s 

def sync(self):
        """override"""
        logger.info('Writing entries...')
        _tried = set()
        _failed = set()
        try:
            while self._dirty:
                name = self._dirty.pop()
                try:
                    self.schedule[name].save()  # save to database
                    logger.debug(
                        '{name} save to database'.format(name=name))
                    _tried.add(name)
                except (KeyError) as exc:
                    logger.error(exc)
                    _failed.add(name)
        except sqlalchemy.exc.IntegrityError as exc:
            logger.exception('Database error while sync: %r', exc)
        except Exception as exc:
            logger.exception(exc)
        finally:
            # retry later, only for the failed ones
            self._dirty |= _failed 

def schedule(self):
        initial = update = False
        if self._initial_read:
            logger.debug('DatabaseScheduler: initial read')
            initial = update = True
            self._initial_read = False
        elif self.schedule_changed():
            # when you updated the `PeriodicTasks` model's `last_update` field
            logger.info('DatabaseScheduler: Schedule changed.')
            update = True

        if update:
            self.sync()
            self._schedule = self.all_as_schedule()
            # the schedule changed, invalidate the heap in Scheduler.tick
            if not initial:
                self._heap = []
                self._heap_invalidated = True
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug('Current schedule:\n%s', '\n'.join(
                    repr(entry) for entry in values(self._schedule)),
                )
        # logger.debug(self._schedule)
        return self._schedule 

def test_serialize_entry_json_serializable(dt):
    entry = ScheduleEntry(
        name='entry-1',
        task='entry-1-task',
        schedule=dt,
        args=('arg1', 'arg2'),
        kwargs={'key1': 'val1', 'key2': 'val2'},
        last_run_at=datetime.now(),
        total_run_count=1,
        options={},
    )

    obj = serialize_entry(entry)

    try:
        json.dumps(obj)
    except Exception as e:
        pytest.fail(e) 

def serialize_entry(entry):
    """
    Serialize ScheduleEntry to json-valid dictionary.

    Helps serialize entry to json, yml and any other formats.

    :param entry: ScheduleEntry
    :return: json-valid dictionary
    """
    return {
        'name': entry.name,
        'task': entry.task,
        'schedule': encode_schedule(entry.schedule),
        'args': entry.args,
        'kwargs': entry.kwargs,
        'last_run_at': encode_datetime(entry.last_run_at),
        'total_run_count': entry.total_run_count,
        'options': entry.options
    } 

def deserialize_entry(entry):
    """
    Deserialize ScheduleEntry from dictionary.

    Helps deserialize entry from json, yml and any other formats.

    :param entry:
    :return:
    """
    return ScheduleEntry(
        name=entry['name'],
        task=entry['task'],
        schedule=decode_schedule(entry['schedule']),
        args=entry['args'],
        kwargs=entry['kwargs'],
        last_run_at=decode_datetime(entry['last_run_at']),
        total_run_count=entry['total_run_count'],
        options=entry['options'],
    ) 

def test_parsing_dict(self):
        task = PeriodicTask(**self.task_dict)
        assert task.name == 'bruteforce_ssh.AlertBruteforceSsh'
        assert task.task == 'bruteforce_ssh.AlertBruteforceSsh'
        assert isinstance(task._id, str)
        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 == 'interval'
        assert task.schedule_str == '1.0 seconds'
        assert isinstance(task.celery_schedule, Interval) is True
        assert isinstance(task.schedule, celery_sched) is True 

def schedule(self):
        return schedules.schedule(datetime.timedelta(**{self.period.code: self.every})) 

def from_schedule(cls, session, schedule, period='seconds'):
        every = max(schedule.run_every.total_seconds(), 0)
        obj = cls.filter_by(session, every=every, period=period).first()
        if obj is None:
            return cls(every=every, period=period)
        else:
            return obj 

def from_schedule(cls, session, schedule):
        spec = {'minute': schedule._orig_minute,
                'hour': schedule._orig_hour,
                'day_of_week': schedule._orig_day_of_week,
                'day_of_month': schedule._orig_day_of_month,
                'month_of_year': schedule._orig_month_of_year}
        obj = cls.filter_by(session, **spec).first()
        if obj is None:
            return cls(**spec)
        else:
            return obj 

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

def is_due(self):
        if not self.model.enabled:
            return False, 5.0   # 5 second delay for re-enable.
        return self.schedule.is_due(self.last_run_at) 

def to_model_schedule(cls, schedule, session):
        """
        :param session:
        :param schedule:
        :return:
        """
        for schedule_type, model_type, model_field in cls.model_schedules:
            debug(cls.model_schedules)
            schedule = schedules.maybe_schedule(schedule)
            if isinstance(schedule, schedule_type):
                model_schedule = model_type.from_schedule(session, schedule)
                cls.save_model(session, model_schedule)
                return model_schedule, model_field
        raise ValueError('Cannot convert schedule type {0!r} to model'.format(schedule)) 

def __repr__(self):
        return '<ModelEntry: {0} {1}(*{2}, **{3}) {{4}}>'.format(
            safe_str(self.name), self.task, self.args, self.kwargs, self.schedule,
        ) 

def setup_schedule(self):
        self.install_default_entries(self.schedule)
        self.update_from_dict(self.app.conf.CELERYBEAT_SCHEDULE) 

def sync(self):
        info('Writing entries...')
        _tried = set()
        while self._dirty:
            try:
                name = self._dirty.pop()
                _tried.add(name)
                self.schedule[name].save()
            except KeyError:
                pass 

def update_from_dict(self, dict_):
        s = {}
        for name, entry in dict_.items():
            try:

                s[name] = self.Entry.from_entry(name, self.session, **entry)
            except Exception as exc:
                error(ADD_ENTRY_ERROR, name, exc, entry)
        self.schedule.update(s) 

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 schedule(self):
        update = False
        if not self._initial_read:
            debug('DatabaseScheduler: intial read')
            update = True
            self._initial_read = True
        elif self.schedule_changed():
            info('DatabaseScheduler: Schedule changed.')
            update = True

        if update:
            self.sync()
            self._schedule = self.all_as_schedule()
            debug('Current schedule:\n%s', '\n'.join(repr(entry) for entry in self._schedule.values()))
        return self._schedule 

def create(self):
        entry = RedBeatSchedulerEntry(
            name=self.name,
            task=self.task,
            schedule=schedule(
                run_every=self.run_every,
                # setting "now" to the job start datetime
                nowfun=lambda: self.run_at,
                app=celery,
            ),
            args=(self.spark_job.pk,),
            kwargs={},
            app=celery,
        )
        return entry 

def add(self):
        """
        Create and save an entry to the scheduler
        """
        logger.info(
            "Adding running %s to schedule. "
            "Interval: %s. "
            "Starts at: %s." % (self.name, self.run_every, self.run_at)
        )
        entry = self.create()
        entry.save()
        return entry 

def delete(self):
        """
        If found, delete the entry from the scheduler
        """
        entry = self.get()
        if entry is None:
            return False
        else:
            logger.info(
                "Deleting running %s from schedule. "
                "Interval: %s. "
                "Starts at: %s." % (self.name, self.run_every, self.run_at)
            )
            entry.delete()
            return True 

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):
        return schedules.schedule(datetime.timedelta(**{self.period: self.every})) 

def from_schedule(cls, dbsession, schedule, period='seconds'):
        every = max(schedule.run_every.total_seconds(), 0)
        try:
            query = dbsession.query(IntervalSchedule)
            query = query.filter_by(every=every, period=period)
            existing = query.one()
            return existing
        except NoResultFound:
            return cls(every=every, period=period)
        except MultipleResultsFound:
            query = dbsession.query(IntervalSchedule)
            query = query.filter_by(every=every, period=period)
            query.delete()
            dbsession.commit()
            return cls(every=every, period=period)