celery once test

start-beat.sh

#!/bin/bash
python tasks.py beat

start-worker.sh

#!/bin/bash
python tasks.py worker

tasks.py

#!/usr/bin/env python

from datetime import timedelta
from celery import Celery
from celery_once import QueueOnce
from config import Config

app = Celery(broker='redis://localhost')
app.conf.update(
    CELERYBEAT_SCHEDULE = {
        'say_ok': {
            'task': 'tasks.say_ok',
            'schedule': timedelta(seconds=10),
        }
    },
    CELERY_IMPORTS = ['tasks'],
)
app.conf.ONCE_REDIS_URL = 'redis://localhost:6379/0'

@app.task(base=QueueOnce)
def say_ok(*args, **kwargs):
    print('OK')


if __name__ == '__main__':
    app.start()