有没有一种方法可以直接或通过查找方法(如Django的`get_model())从Celery任务访问可调用的Python对象。我有一个点符号引用，指向可调用对象的声明位置，但在发送/调用任务之前，我需要实际对象访问它的属性。
我搜索了公共和私人方法，但没有一个明显是我想要的：
$ dir(celery_task_instance)
> ['AsyncResult', 'MaxRetriesExceededError', 'OperationalError', 'Strategy',
'__bound__', '__call__', '__class__', '__delattr__', '__dict__', '__dir__',
'__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__',
'__hash__', '__header__', '__init__', '__le__', '__lt__', '__module__',
'__name__', '__ne__', '__new__', '__qualname__', '__reduce__',
'__reduce_ex__', '__repr__', '__self__', '__setattr__', '__sizeof__',
'__str__', '__subclasshook__', '__trace__', '__v2_compat__', '__weakref__',
'__wrapped__', '_app', '_backend', '_decorated', '_default_request',
'_exec_options', '_get_app', '_get_exec_options', '_get_request',
'abstract', 'acks_late', 'add_around', 'add_to_chord', 'add_trail',
'after_return', 'annotate', 'app', 'apply', 'apply_async', 'autoregister',
'backend', 'bind', 'chunks', 'default_retry_delay', 'delay', 'expires',
'from_config', 'ignore_result', 'map', 'max_retries', 'name', 'on_bound',
'on_failure', 'on_retry', 'on_success', 'pop_request', 'push_request',
'rate_limit', 'reject_on_worker_lost', 'replace', 'request',
'request_stack', 'resultrepr_maxsize', 'retry', 'run', 's', 'send_event',
'send_events', 'serializer', 'shadow_name', 'si', 'signature', 'signature_from_request',
'soft_time_limit', 'starmap', 'start_strategy', 'store_errors_even_if_ignored', 'subtask',
'subtask_from_request', 'throws', 'time_limit', 'track_started', 'trail', 'update_state']
背景
我使用django-celery-beat为Django项目中的Celery创建计划任务。
我想在创建计划任务时验证args和kwargs字段的模式，以预测和防止未来的运行时问题。例如，通过kwargs字段定义的API参数。
我可以在Djangoforms.ModelForm.clean()方法中访问已注册任务的列表我已经能够访问已注册的Celery任务。我认为我可以将验证模式存储为目标@shared_task的一个属性，然后在创建任务时访问它。模式被定义为marshmallow.Schema，以防它相关，但它也可以被硬编码。
我找错地方了——importlib让这件事变得微不足道。我在想这是我需要从Celery那里得到的东西。
import importlib
split_path = dot_notation_path.split(".")
module_name, callable_name = ".".join(split_path[:-1]), split_path[-1]
module_object = importlib.import_module(module_name)
callable_object = getattr(somemodule, callable_name)
具有
@app.task(name='some_task')
def some_function_with_task_decorator(params: dict):
do_sth()
CCD_ 9属于CCD_。要获得可调用名称，请使用some_function_with_task_decorator._get_current_object().__name__
some_function_with_task_decorator._get_current_object()从PromiseProxy中取出Task
由于Task是一个装饰器，请使用__name__来获得装饰函数。然后，您可以使用模块的名称(字符串)动态调用模块的函数