quart_tasks.extension module#

class quart_tasks.extension.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#