Run with scheduler

You can run Taskiq Dashboard with scheduler. By passing scheduler instance to the TaskiqDashboard constructor, you will get additional page at dashboard with scheduled tasks.

In this example, we will demonstrate how to set up Taskiq Dashboard along with a task scheduler:

import asyncio
import datetime as dt
import random
import typing as tp

from taskiq import TaskiqScheduler
from taskiq.middlewares.taskiq_admin_middleware import TaskiqAdminMiddleware
from taskiq_pg.asyncpg import AsyncpgBroker, AsyncpgResultBackend, AsyncpgScheduleSource

from taskiq_dashboard import TaskiqDashboard


dsn = 'postgres://taskiq-dashboard:look_in_vault@localhost:5432/taskiq-dashboard'
broker = (
    AsyncpgBroker(dsn)
    .with_result_backend(AsyncpgResultBackend(dsn))
    .with_middlewares(
        TaskiqAdminMiddleware(
            url='http://0.0.0.0:8000',  # the url to your taskiq-admin instance
            api_token='supersecret',  # any secret enough string
            taskiq_broker_name='my_worker',
        )
    )
)
scheduler = TaskiqScheduler(
    broker=broker,
    sources=[
        AsyncpgScheduleSource(
            dsn=dsn,
            broker=broker,
        ),
    ],
)


@broker.task(
    task_name='solve_all_problems',
    schedule=[
        {'cron': '*/1 * * * *'},
        {'time': dt.datetime.now(dt.timezone.utc) + dt.timedelta(minutes=2)},
    ],
)
async def best_task_ever(*args, **kwargs) -> dict[str, tp.Any]:
    """Solve all problems in the world."""
    await asyncio.sleep(15)
    print('All problems are solved!')

    error_probability = 0.2
    if random.random() < error_probability:
        raise RuntimeError('An unexpected error occurred while solving problems.')

    return {
        'status': 'success',
        'random_number': random.randint(1, 42),
        'args': args,
        'kwargs': kwargs,
    }


def run_admin_panel() -> None:
    app = TaskiqDashboard(
        api_token='supersecret',
        broker=broker,
        scheduler=scheduler,
        host='0.0.0.0',
        port=8000,
    )
    app.run()


if __name__ == '__main__':
    run_admin_panel()

To run this example, execute the following commands in different terminals:

1. Run the worker:

   uv run taskiq worker docs.examples.example_with_scheduler:broker --workers 1
   

2. Run the admin panel:

   uv run python -m docs.examples.example_with_scheduler
   

3. Run the scheduler (optional, only for sending task):

   uv run taskiq scheduler docs.examples.example_with_scheduler:scheduler
   

After that, open your browser and navigate to http://0.0.0.0:8000/schedules to access the dashboard and see your scheduled task.