Advertisements I have a Django project structure like this (not showing all the files/folders): myproj/ ├── __init__.py ├── config/ │ ├── __init__.py <– imports `celery_app` as per docs │ ├── settings/ │ │ ├── __init__.py │ │ └── base.py │ ├── celery.py <– defines `app = Celery("myproj")` │ └── wsgi.py ├── myapp1/ └── myapp2/ I… Read More ModuleNotFoundError when starting celery in Django project

Advertisements I’m trying to test that my Celery task updates Django model. It works fine, but behaves strangely during testing. # someapp/models.py class SomeModel(models.Model): … hidden = models.BooleanField(default=False) # someapp/tasks.py @shared_task() def my_task(model_id): model_instance = someapp.models.SomeModel.objects.get(id=model_id) model_instance.hidden = True model_instance.save() logger.info(f’Uncovered model instance with id {model_id]’) To test this I’ve implemented following workflow: I create… Read More Celery task behave strangely while testing

Advertisements I am trying to auto-reload celery that is part of a Django application when code changes are detected. I am starting celery with the following management command: """ https://avilpage.com/2017/05/how-to-auto-reload-celery-workers-in-development.html """ import shlex import subprocess from django.core.management.base import BaseCommand from django.utils import autoreload def restart_celery_worker(): cmd = "pkill -9 celery" subprocess.call(shlex.split(cmd)) cmd = "celery -A… Read More Python 3.10 subprocess error: FileNotFoundError: [Errno 2] No such file or directory: 'pkill'