z4j-taskiqscheduler

The native TaskIQ scheduler adapter for z4j, with dashboard management of label-source schedules. Dashboard management needs the z4j server, z4j-core, z4j-bare and this adapter from the same release, plus the producer setup below.

Compatibility

TaskIQ 0.13+; Python 3.11+.

Live Redis qualification uses TaskIQ 0.13.0 with taskiq-redis 1.2.4 and redis 8.1.0 in a separate environment from arq (which requires redis below 6); keep the two dependency sets apart.

Installation and producer setup

Install z4j-taskiqscheduler and z4j-bare from the same release wave, plus the native scheduler and broker/store dependencies used by your application. The example assumes your task code, writable application-state directory and agent credentials already exist. Use the project UUID from agent registration.

import asyncio
import os
from datetime import timedelta
from myapp.tasks import broker  # Import all task definitions first.
from taskiq.cli.scheduler.run import SchedulerLoop
from taskiq.schedule_sources import LabelScheduleSource
from taskiq.scheduler.scheduler import TaskiqScheduler
from z4j_bare import NativeScheduleControl, install_agent
from z4j_taskiqscheduler import TaskiqSchedulerAdapter

async def main():
    await broker.startup()
    source = LabelScheduleSource(broker)
    await source.startup()
    adapter = TaskiqSchedulerAdapter(
        source=source, source_loop=asyncio.get_running_loop(),
        control=NativeScheduleControl("/var/lib/myapp/taskiq-schedules.sqlite3"),
    )
    agent = install_agent(
        engines=[], schedulers=[adapter],
        brain_url=os.environ["Z4J_BRAIN_URL"], token=os.environ["Z4J_TOKEN"],
        hmac_secret=os.environ["Z4J_HMAC_SECRET"], project_id=os.environ["Z4J_PROJECT_ID"],
    )
    try:
        await SchedulerLoop(TaskiqScheduler(broker, [source])).run(
            update_interval=timedelta(seconds=1),
        )
    finally:
        agent.stop()
        await broker.shutdown()

asyncio.run(main())

Start the normal TaskIQ worker separately against the same broker. Control must wrap the actual LabelScheduleSource in the scheduler process. Custom schedule sources retain their inventory and explicit deletion extension; the durable managed write path supports the stock label source only. Without control, the label source remains read-only for definitions; manual runs require its owning event loop.

Dashboard controls

The connected producer supports inventory, create, edit, pause, resume, delete and Run now. A write is complete only after z4j accepts a snapshot confirming the native source. The dashboard uses the current connection's advertised capabilities; an unavailable producer cannot accept new edits. Repeating an operation ID does not submit it twice. A timeout or uncertain outcome requires inspection before submitting a new request.

A pause prevents future scheduling. Work already queued or running can finish. Run now submits an additional execution without resuming a paused schedule. Native scheduling keeps the upstream scheduler's timing and delivery semantics; it does not acquire z4j-scheduler's catch-up, overlap or exactly-once guarantees. Queue task history requires the corresponding engine adapter in the workers.

Code definitions and durable overrides

NativeScheduleControl stores dashboard changes in a private SQLite file in the producer's application state directory. Keep that file on a durable local volume, back it up with the application, and reuse the same path after restart. Use a separate file for each application and scheduler owner. Do not put it on an ephemeral container filesystem or a network filesystem.

Code supplies the initial definitions. Each saved change overrides the fields it submits: pause and resume submit only the enabled flag, while the dashboard edit form submits every editable field (task, schedule type, expression, timezone, queue, arguments and enabled flag), so later code changes to those fields stay hidden. Removing a code definition removes its projection; reintroducing the same source identity restores any saved override. Deletion is a durable tombstone, so restarting the application does not recreate it. Dashboard-created schedules survive with the same file. No Python files are rewritten. New schedules can invoke only tasks already registered in the host application; deploying new task code remains an application operation.

The producer reads committed controls locally and keeps scheduling during a z4j server outage. If that local state cannot be read or validated, periodic dispatch is withheld and the error is logged. Normal task processing continues. Install control before starting periodic dispatch. Keep one periodic producer per application, as required by its native scheduler. There is currently no separate dashboard action to remove an override and restore the code default.

Supported definitions and limits

• Supports five-field cron, positive integer interval seconds and offset-aware one-shot datetimes. Native cron weekday names and fixed offsets are preserved. Each managed definition must select exactly one trigger type; combined native cron/interval/time entries are rejected before installing control.
• Prefer explicit unique schedule_id values in code labels. Without one, a deterministic definition hash supplies identity; changing that code definition creates a new identity and its old overrides do not transfer.
• Two identical code label entries need distinct explicit IDs.
• The registered task's labels determine routing; a dashboard queue override is unavailable.
• A cached schedule is rechecked before dispatch, so paused, removed or edited definitions do not run with obsolete cached settings.

Links

• Documentation
• Changelog
• Security policy; report vulnerabilities to security@z4j.com
• Apache-2.0 license