This document describes the current stable version of Celery (4.4). For development docs, go here.

Source code for celery.concurrency.solo

# -*- coding: utf-8 -*-
"""Single-threaded execution pool."""
from __future__ import absolute_import, unicode_literals

import os

from celery import signals

from .base import BasePool, apply_target

__all__ = ('TaskPool',)

[docs]class TaskPool(BasePool): """Solo task pool (blocking, inline, fast).""" body_can_be_buffer = True def __init__(self, *args, **kwargs): super(TaskPool, self).__init__(*args, **kwargs) self.on_apply = apply_target self.limit = 1 signals.worker_process_init.send(sender=None) def _get_info(self): return { 'max-concurrency': 1, 'processes': [os.getpid()], 'max-tasks-per-child': None, 'put-guarded-by-semaphore': True, 'timeouts': (), }