quart_tasks package#

Submodules#

Module contents#

class quart_tasks.MemoryStore#

Bases: TaskStoreABC

An in memory store of execution times.

async get(key: str, default: datetime) → datetime#

Get the last execution time for the task for the given key if present or the default if not.

Parameters:

  • key – The key to indentify the task.

  • default – If no last execution for the key is available, return this default.

Returns:

The last execution time.

async set(key: str, executed: datetime) → None#

Set the execution time for the given key.

Parameters:

  • key – The key to indentify the task.

  • executed – The time it was executed.

async shutdown() → None#

A coroutine within which any cleanup can be done.

async startup() → None#

A coroutine within which any setup can be done.

class quart_tasks.QuartTasks(app: Quart | None = None, *, store: TaskStoreABC | None = None, tzinfo: tzinfo | None = None)#

Bases: object

add_cron_task(task: Callable, cron_expression: str | None = None, *, seconds: str | None = None, minutes: str | None = None, hours: str | None = None, day_of_month: str | None = None, month: str | None = None, day_of_week: str | None = None, name: str | None = None) → None#

Add a cron task.

If the task is a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned).

The cron definition can either be given in the cron format, or as individual arguments. See for example https://crontab.guru/. An example usage,

async def infrequent_task():
    ...

tasks.add_cron_task(infrequent_task, "*/5 * * * *)
Parameters:

  • cron_expression – The cron defintion.

  • task – The callable to execute as the task.

  • seconds – The seconds part of the cron definition.

  • minutes – The minutes part of the cron definition.

  • hours – The hours part of the cron definition.

  • day_of_month – The day of month part of the cron definition.

  • month – The month part of the cron definition.

  • day_of_week – The day of week part of the cron definition.

  • name – Name of the task, defaults to the function name.

add_periodic_task(task: Callable, period: timedelta, *, name: str | None = None) → None#

Add a periodic task.

If the task is a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned).

async def frequent_task():
    ...

tasks.add_periodic_task(timedelta(seconds=5), frequent_task)
Parameters:

  • period – The period between task invocations.

  • task – The callable to execute as the task.

  • name – Name of the task, defaults to the function name.

after_task(func: Callable[[], Awaitable[None]]) → Callable[[], Awaitable[None]]#

Add a after task function.

This will allow the function provided to be called once after any invocation of a task. This runs within the same app context as the task invocation.

This is designed to be used as a decorator, if used to decorate a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned). An example usage,

@app.after_task
async def func():
    ...
Parameters:

func – The function itself.

before_task(func: Callable[[], Awaitable[None]]) → Callable[[], Awaitable[None]]#

Add a before task function.

This will allow the function provided to be called once before any invocation of a task. This runs within the same app context as the task invocation.

This is designed to be used as a decorator, if used to decorate a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned). An example usage,

@app.before_task
async def func():
    ...
Parameters:

func – The function itself.

cron(cron_expression: str | None = None, /, *, seconds: str | None = None, minutes: str | None = None, hours: str | None = None, day_of_month: str | None = None, month: str | None = None, day_of_week: str | None = None, name: str | None = None) → Callable[[Callable[[P], T]], Callable[[P], T]]#

Add a cron task.

This is designed to be used as a decorator, if used to decorate a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned).

The cron definition can either be given in the cron format, or as individual arguments. See for example https://crontab.guru/. An example usage,

@tasks.cron("*/5 * * * *)
async def infrequent_task():
    ...
Parameters:

  • cron_expression – The cron defintion.

  • seconds – The seconds part of the cron definition.

  • minutes – The minutes part of the cron definition.

  • hours – The hours part of the cron definition.

  • day_of_month – The day of month part of the cron definition.

  • month – The month part of the cron definition.

  • day_of_week – The day of week part of the cron definition.

  • name – Name of the task, defaults to the function name.

init_app(app: Quart) → None#
periodic(period: timedelta, *, name: str | None = None) → Callable[[Callable[[P], T]], Callable[[P], T]]#

Add a periodic task.

This is designed to be used as a decorator, if used to decorate a synchronous function, the function will be wrapped in run_sync() and run in a thread executor (with the wrapped function returned).

@tasks.periodic(timedelta(seconds=5))
async def frequent_task():
    ...
Parameters:

  • period – The period between task invocations.

  • name – Name of the task, defaults to the function name.

async run(task_name: str | None = None) → None#
task_context() → AsyncGenerator[None, None]#
async test_run(task_name: str) → None#
class quart_tasks.TaskStoreABC#

Bases: object

abstractmethod async get(key: str, default: datetime) → datetime#

Get the last execution time for the task for the given key if present or the default if not.

Parameters:

  • key – The key to indentify the task.

  • default – If no last execution for the key is available, return this default.

Returns:

The last execution time.

abstractmethod async set(key: str, executed: datetime) → None#

Set the execution time for the given key.

Parameters:

  • key – The key to indentify the task.

  • executed – The time it was executed.

abstractmethod async shutdown() → None#

A coroutine within which any cleanup can be done.

abstractmethod async startup() → None#

A coroutine within which any setup can be done.