How to use django orm in Celery periodic tasks?

Question

Django

I have a django application that needs to generate a report once a week from information in a database. I set up celery, it says in the logs that the task is running, but nothing changes in the database.
I tried to make a test problem and check whether at least something will change at all.

# файл app/tasks.py
from vkmt_api.celery import app
from authentication.models import User

@app.task
def test_task():
    user = User.objects.get(username="test")
    user.first_name = user.first_name + "1"
    user.save()

# файл mainapp/celery.py

import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_project.settings')
app = Celery('my_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

app.conf.beat_schedule = {
    "test-task": {
        "task": "reports.tasks.test_task",
        "schedule": crontab(minute="*/1")
    }
}

# настройки в файле settings.py

REDIS_HOST = '0.0.0.0'
REDIS_PORT = '6379'

CELERY_BROKER_URL = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0'
CELERY_BROKER_TRANSPORT_OPTION = {'visibility_timeout': 3600}
CELERY_RESULT_BACKEND = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

Based on the above code, the most simple logic follows, once a minute a function should be executed and a unit should be added to the test_task()user 's name. testThat is, 5 minutes after the application is launched, the user testname will be "name11111".
Tell me what am I doing wrong?

Answer 1

In general, with this approach, you need to run it not through beat, but through worker. Which, in fact, is strange, considering that for periodic scheduled tasks, the official documentation says to use beat.
https://docs.celeryproject.org/en/stable/django/fi...