Celery.beat.SchedulingError: Couldn't apply scheduled task broadcast_notification_8: 'int' object is not iterable.... спасибо всем
celery.beat.SchedulingError: Couldn't apply scheduled task broadcast_notification_8: 'int' object is not iterable.... спасибо всем
Traceback (most recent call last):
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
401, in apply_async
entry_args = _evaluate_entry_args(entry.args)
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
211, in _evaluate_entry_args
for v in entry_args
TypeError: 'int' object is not iterable
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
287, in apply_entry
result = self.apply_async(entry, producer=producer, advance=False)
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
414, in apply_async
entry, exc=exc)), sys.exc_info()[2])
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\exceptions.py",
line 108, in reraise
raise value.with_traceback(tb)
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
401, in apply_async
entry_args = _evaluate_entry_args(entry.args)
File "C:\Users\USER\AppData\Roaming\Python\Python37\site-packages\celery\beat.py", line
211, in _evaluate_entry_args
for v in entry_args
celery.beat.SchedulingError: Couldn't apply scheduled task broadcast_notification_8:
int' object is not iterable
celery.beat.SchedulingError: Couldn't apply scheduled task broadcast_notification_8: 'int' object is not iterable.... спасибо всем. Мой tasks.py
@shared_task(bind=True)
def broadcast_notification(self, data):
print(data)
try:
notification = BroadcastNotification.objects.filter(id=int(data))
if len(notification) > 0:
notification = notification.first()
channel_layer = get_channel_layer()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(channel_layer.group_send(
"notification_broadcast",
{
'type': 'send_notification',
'message': json.dumps(notification.message),
}))
notification.sent = True
notification.save()
return 'Done'
else:
self.update_state(
state = 'FAILURE',
meta = {'exe': 'Not Found'}
)
raise Ignore()
except:
self.update_state(
state = 'FAILURE',
meta = {
'exe': 'Failed',
# 'exc_type': type(ex).__name__,
# 'exc_message': traceback.format_exc().split('\n')
# 'custom': '...'
}
)
models.py
class BroadcastNotification(models.Model):
message = models.TextField()
broadcast_on = models.DateTimeField()
sent = models.BooleanField(default=False)
class Meta:
ordering = ['-broadcast_on']
@receiver(post_save, sender=BroadcastNotification)
def notification_handler(sender, instance, created, **kwargs):
if created:
schedule, created = CrontabSchedule.objects.get_or_create(hour =
instance.broadcast_on.hour, minute=instance.broadcast_on.minute,
day_of_month=instance.broadcast_on.day, month_of_year=instance.broadcast_on.month)
task = PeriodicTask.objects.create(crontab=schedule,
name="broadcast_notification_"+str(instance.id),
task="notification_app.tasks.broadcast_notification", args=json.dumps((instance.id)))
celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'emis.settings')
app = Celery('emis')
app.conf.enable_utc = False
app.conf.update(timezone = 'Africa/Lagos')
app.config_from_object(settings, namespace='CELERY')
app.conf.beat_schedule = {}
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')
run
celery -A emis.celery worker --pool=solo -l info
celery -A emis beat -l info