Привет не могу понять как работают чанки в селери. Расскажите плиз как их использовать.
@celery_app.task(name='high.add')
def add(x:list):
print(x)
add.chunks(iter([1,2,3,4,5]), 3).group().apply_async(queue='high')
TypeError: api.celery_tasks.queues.high.add() argument after * must be an iterable, not int
Я хочу разбить список на мелкие списки по N элементов и запустить параллельное выполнение. Но из того что я сейчас вижу оно пытается взять первые 3 элемента из списка и запустить их параллельно add(1) add(2) add(3)
А я хочу что-то в духе add([1,2,3]) add([4,5])
Окей так и таким образом он тоже не работает, я пробую запускать
add.chunks(iter([[1,2,3,4,5], [1,3,4,5,6]]), 3).group().apply_async(queue='high')
TypeError: add() takes 1 positional argument but 5 were given
Оно парсит подсписок и пытается каждый элемент отправить в функцию я понять не могу.
И вот так я пробовал
@celery_app.task()
def send(id):
return send_message(id)
send.chunks([1,2,3,4,5], 3).group().apply_async(queue='high')
TypeError: api.celery_tasks.queues.high.add() argument after * must be an iterable, not int
Хелпаните плиз, я уже закипаю.
