python - Celery 動的タスク / Celery 実装をインターフェースの背後に隠す

Question

Celery の実装に結び付けずに、非同期ジョブを Celery で実装する方法を見つけようとしています。

callable (または callable をラップするオブジェクト) など、スケジュールするオブジェクトを受け入れるインターフェイスがある場合:

ITaskManager(Interface):
    def schedule(task):
        #eventually run task

そして、私はそれを threading モジュールで実装するかもしれません:

ThreadingTaskManager(object)
    def schedule(task):
        Thread(task).start() # or similar

でもセロリでは無理みたいですね。

Answer 1

おそらく、非常に醜いですが、解決策は、引数として渡されるタスクオブジェクトを動的にロードする1つのセロリタスクを定義することかもしれません。

@celery.task
def taskrunner(taskname):
    taskModule = __import__(taskname)
    taskModule.run()

CeleryTaskManager(object)
    def schedule(task):
        taskrunner.delay(task.__file__)


from mytask import run

CeleryTaskManager().schedule(run)