Improvement: Adds field PeriodicTask.origin_key to sync tasks defined in source code with settings.beat_schedule

[pztrick]

This adds a field PeriodicTask.origin_key which indicates if a task database row was populated dynamically or from source code (e.g. settings.CELERYBEAT_SCHEDULE or settings.beat_schedule in 4.1+).

If an entry is removed from settings.CELERYBEAT_SCHEDULE it will also delete any previously populated database row for PeriodicTask.

E.g.

# before
CELERYBEAT_SCHEDULE = {
    'test-sms': {
        'task': 'app.tasks.test_sms',
        'schedule': datetime.timedelta(seconds=60),
        'args': None
    },
}

# after
CELERYBEAT_SCHEDULE = {
    # 'test-sms': {
    #     'task': 'app.tasks.test_sms',
    #     'schedule': datetime.timedelta(seconds=60),
    #     'args': None
    # },
}

After commenting out the above, celery beat would purge the task from database and also emit this logger statement:

[2018-02-09 13:51:48,345: WARNING/Beat] Purged periodic tasks [origin_key=beat_schedule]: test-sms

---

I also updated the scheduler to only ingest one and only one schedule when reading settings.CELERYBEAT_SCHEDULE. The old behavior was that if you re-wrote an interval schedule to a cron schedule, it would remain associated with the old (now deleted) interval schedule and be called for both schedules. E.g.:

# old
CELERYBEAT_SCHEDULE = {
    'test-sms': {
        'task': 'app.tasks.test_sms',
        'schedule': datetime.timedelta(seconds=60),
        'args': None
    },
}

# new
CELERYBEAT_SCHEDULE = {
    'test-sms': {
        'task': 'app.tasks.test_sms',
        'schedule': crontab(minute='22', hour='14', day_of_week='*'),
        'args': None
    },
}

[auvipy]

A new pr has merged upstream, pulling that and regenerating the
migrations by deleting the one here and a test case to verify the change is needed.

[liquidpele]

So this will run two queries instead of one now? It would be nicer to keep it as one.

[liquidpele]

Could replace a lot of logic here with just this I think:

tasks_to_purge = PeriodicTask.objects.filter(
    origin_key=origin_key).exclude(name__in=mapping.keys())
if tasks_to_purge:
    tasks_to_purge.delete()
    log.warn(...)

[liquidpele]

Other migrations has been added, you'll need to rebase this to be the next one in the list based on what's in the master branch.