EVECelery.tasks.BaseTasks.TaskCached#

Module Contents#

Classes#

TaskCached

A task utilizing Redis for some form of caching and locks where the results are cached.
class EVECelery.tasks.BaseTasks.TaskCached.TaskCached#

Bases: EVECelery.tasks.BaseTasks.TaskBase.TaskBase

A task utilizing Redis for some form of caching and locks where the results are cached.

property redis_locks: redis.Redis#

Get the Redis client for managing the locks

Return type:

redis.Redis

property redis_cache: redis.Redis#

Get the Redis client for managing the cache

Return type:

redis.Redis

default_ttl() → int#

Default TTL to set in Redis for results that don’t have expiry information.

Returns:

The number of seconds to cache a response

Return type:

int

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) –

get_hash(**kwargs) → str#

Returns the SHA256 hash of kwargs

Return type:

str

get_redis_cache_key(**kwargs) → str#

Returns the Redis key for storing a cached response with the provided input parameters.

Parameters:

kwargs – Parameters passed to the task

Returns:

Redis key for storing the value of the ESI response

Return type:

str

get_redis_lock_key(**kwargs) → str#

Returns the Redis key for lock response with the provided input parameters.

Parameters:

kwargs – ESI request parameters

Returns:

Redis key for storing ESI locks

Return type:

str

cache_key_exists(**kwargs) → bool#

Check if cache key exists for given input

Return type:

bool

get_sync(kwargs_apply_async: dict | None = None, kwargs_get: dict | None = None, **kwargs) → EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedResponse#

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:

Return type:

EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedResponse

>>> 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()
run(**kwargs) → dict#

Returns the result from the Redis cache if it exists, otherwise runs and returns _run_get_result() as a dictionary.

This function checks the cache based on the provided input and will return the result from the cache first, skipping a rerun of _run_get_result(). If the result does not exist in the cache it will run _run_get_result() and cache the result for future task runs.

If an exception is raised by _run_get_result() the exception will be raised to the calling client and will not be cached. If _run_get_result() returns a ModelCachedException() then the exception will first be cached and then thrown to the calling client as a CachedException().

Returns:

The response data as a dictionary. This dictionary can be converted to a pydantic model by passing it to this task’s to_pydantic() method for easier inspection and schema documentation (required fields, field meanings, etc.).

Return type:

dict

abstract _run_get_result(**kwargs) → EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedResponse | EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedException#

The task body to execute if the result is not currently cached.

This function is the body of a task that gets executed when a previously returned result is not cached or has expired. This function must return a pydantic model that inherits from ModelCachedResponse() or ModelCachedException().

If ModelCachedException() is returned, then the exception will be cached for the defined TTL and have an exception thrown to the calling client.

Returns:

The response success or cached exception pydantic model.

Return type:

Union[EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedResponse, EVECelery.tasks.BaseTasks.Models.ModelsCached.ModelCachedException]