__init__.py 390 B

12345678910111213
  1. from celery.utils import get_cls_by_name
  2. ALIASES = {
  3. "processes": "celery.concurrency.processes.TaskPool",
  4. "eventlet": "celery.concurrency.evlet.TaskPool",
  5. "gevent": "celery.concurrency.evg.TaskPool",
  6. "threads": "celery.concurrency.threads.TaskPool",
  7. "solo": "celery.concurrency.solo.TaskPool",
  8. }
  9. def get_implementation(cls):
  10. return get_cls_by_name(cls, ALIASES)