EVECelery.tasks.BaseTasks.TaskBase#

Module Contents#

Classes#

TaskBase

The base Celery task class used by EVECelery.

class EVECelery.tasks.BaseTasks.TaskBase.TaskBase#

Bases: celery.Task

The base Celery task class used by EVECelery.

All tasks to be processed and automatically registered by the EVECelery workers should inherit from this base task.

This BaseTask supports all methods and properties available by Celery’s class-based tasks.

property queue_assignment: str | None#

The queue to register this task with.

Returns the name of the queue to optionally register this task with. If None is provided then this task is registered with the default queue defined by the celery app.

Return type:

str | None

classmethod reflection_get_model(model_name: str)#

Lookup a pydantic model in the current module by its string name.

Resolves a pydantic model existing or imported by the current module by its name. :param model_name: The class name to lookup

Parameters:

model_name (str) –

classmethod to_pydantic(json_data: str | dict) Type[EVECelery.tasks.BaseTasks.Models.ModelsBase.ModelBaseWithModelInfo]#

Deserialize a JSON serialized string to a pydantic model.

This function will deserialize a json string to a pydantic model. It will also create a pydantic model from a dictionary if the data has been deserialized elsewhere.

For this function to correctly identify the pydantic model to generate the input must contain the ‘pydantic_model’ key. This key refers to the name of the pydantic class model within the calling module. The pydantic model to generate must exist within the current module’s namespace and its name must exactly match the ‘pydantic_model’ value.

Parameters:

json_data (Union[str, dict]) – The JSON data to deserialize. If this is a string it is first deserialized using json.loads(). If this is a dictionary it is directly parsed by the pydantic model.

Raises:
  • KeyError – If the input data does not contain the ‘pydantic_model’ key.

  • AttributeError – If the lookup key from ‘pydantic_model’ is unable to find the specified model

  • TypeError – If the lookup key refers to an unexpected type

Returns:

The deserialized JSON data as a pydantic object.

Return type:

Type[EVECelery.tasks.BaseTasks.Models.ModelsBase.ModelBaseWithModelInfo]

get_sync(kwargs_apply_async: dict | None = None, kwargs_get: dict | None = None, **kwargs)#

Call this task and block until the result is available.

Calls this celery task with additional helper functionality specific for EVECelery (deserialization to pydantic models support and result backend resource cleanup). This function is a wrapper around Celery’s task.apply_async() and AsyncResult.get() methods.

Parameters:
  • kwargs_apply_async (Optional[dict]) –

    Dictionary of keyword arguments passed to task.apply_async()

  • kwargs_get (Optional[dict]) –

    Dictionary of keyword arguments passed to AsyncResult.get()

  • **kwargs

    Keyword arguments passed to this task’s run() method.

Returns:

Result of a task

Examples:

>>> TaskBase.get_sync(kwargs_get={'timeout': 5}) # equivalent to BaseTask.apply_async().get(timeout=5)
>>> r = TaskBase.get_sync() # equivalent to BaseTask.apply_async().get()
>>> r = TaskBase.get_sync(lookup_id=5) # equivalent to BaseTask.apply_async(kwargs={'lookup_id': 5}).get()
classmethod get_all_subtasks()#

Yield task subclasses that inherit from this task class.

Yields all subclasses from this base class. The base class itself is not yielded.