If using a Task subclass without @app.task what is the correct way to defer registration until app finalize time?

[jeking3]

Summary

I have a couple of classes derived from Task which are not accessed through an @app.task wrapper. Instead the run method is defined in the subclass, and existing code calls it by creating an instance of the task and then calling delay() on it.

The code in @app.task ends up accessing some private stuff inside celery._state to put the task into a set for finalization. There is code in the Application class called register_task which does work for this purpose, but the documentation in the method says it should not be necessary after celery 1.0. I found a way to do it but it required using something in celery._state so it's not ideal (see root cause, below).

Symptom

In a python:3.9.7-slim-buster container, using celery worker 5.1.2 starts fine, but when I upgrade to celery 5.2.0 with no other changes, this happens on startup:

DEBUG 2021-11-10 13:39:24,690 /usr/local/lib/python3.9/site-packages/celery/bootsteps.py _debug 259 | Worker: Starting Pool
[2021-11-10 13:39:24,783: ERROR/ForkPoolWorker-1] Process ForkPoolWorker-1
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/billiard/process.py", line 327, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.9/site-packages/billiard/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.9/site-packages/billiard/pool.py", line 289, in __call__
    self.after_fork()
  File "/usr/local/lib/python3.9/site-packages/billiard/pool.py", line 421, in after_fork
    self.initializer(*self.initargs)
  File "/usr/local/lib/python3.9/site-packages/celery/concurrency/prefork.py", line 75, in process_initializer
    task.__trace__ = build_tracer(name, task, app.loader, hostname,
  File "/usr/local/lib/python3.9/site-packages/celery/app/trace.py", line 361, in build_tracer
    push_request = request_stack.push
AttributeError: 'NoneType' object has no attribute 'push'

It appears the task's request_stack is None.

On the tasks that are showing this issue, they are implemented in a class derived from Task. I must have done something incorrect in my implementation as I do not use the @task wrapper for these.

Root Cause

The root cause for this issue was that since I am not using the @app.task wrapper, I need to properly register the tasks myself. I was incorrectly finalizing the app too early by calling this in the module with the task definition:

celery_app.tasks.register(PullTaskRunner)

Accessing the register property auto-finalizes the app, and this was done too early. The task never bound to the app this way, and thus had an empty request_stack when it was invoked at startup (from tasks pending in the database).

The only fix I could find that worked reasonably was to use the following code in my tasks.py which is pulled in at autodiscover time:

from celery._state import connect_on_app_finalize  # type: ignore
connect_on_app_finalize(celery_app.register_task(PullTaskRunner))  # type: ignore
connect_on_app_finalize(celery_app.register_task(PushTaskRunner))  # type: ignore

I don't like accessing a private module like state, but the code in @app.task essentially does this.
Further, the comments in register_task suggest it has been unnecessary since celery 1.0.

So, for a pure Task class derivation where @app.task is not being used, how is one supposed to properly defer registration of the task to finalization time, when the code that adds to the finalize set in _state is all considered to be private? It looks like @app.task is the only way to do this, but since I already have a Task subclass with a run() and I invoke it directly with delay(), I have tasks in the database for which I would not want to make a structural change that might invalidate them by switching to an @app.task.

[Zlopez]

I found this way of registering the custom task in Celery documentation.