可从 Celery 任务实例调用的访问目标 Python
Access target Python callable from Celery task instance
有没有办法直接或通过查找方法（如 Django 的 `get_model()）从 Celery 任务访问可调用的 Python 对象。我有一个指向可调用对象声明位置的点符号引用，但我需要实际对象才能在发送/调用任务之前访问它的属性。
我已经搜索了 public 和私有方法，但其中 none 显然是我要找的东西：
$ 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 字段的架构，以预测和防止未来的运行时问题。例如，API 参数定义通过 kwargs 字段提供。
我可以在我的 Django forms.ModelForm.clean() 方法中访问已注册任务的列表我已经能够访问已注册的 Celery 任务。我认为我可以将验证模式存储为目标 @shared_task 的属性，然后在任务创建期间访问它。该模式被定义为 marshmallow.Schema 以防它是相关的，但这同样可以被硬编码。
我找错地方了 - importlib 让这变得相当微不足道。我在想这是我需要从芹菜那里得到的东西。
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)
有没有办法直接或通过查找方法（如 Django 的 `get_model()）从 Celery 任务访问可调用的 Python 对象。我有一个指向可调用对象声明位置的点符号引用，但我需要实际对象才能在发送/调用任务之前访问它的属性。
我已经搜索了 public 和私有方法，但其中 none 显然是我要找的东西：
$ 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 字段的架构，以预测和防止未来的运行时问题。例如，API 参数定义通过 kwargs 字段提供。
我可以在我的 Django forms.ModelForm.clean() 方法中访问已注册任务的列表我已经能够访问已注册的 Celery 任务。我认为我可以将验证模式存储为目标 @shared_task 的属性，然后在任务创建期间访问它。该模式被定义为 marshmallow.Schema 以防它是相关的，但这同样可以被硬编码。
我找错地方了 - importlib 让这变得相当微不足道。我在想这是我需要从芹菜那里得到的东西。
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)