Summary

AsyncpgScheduleSource.startup() calls _update_schedules_on_startup(), which runs a full DELETE FROM <table> and then re-inserts only the schedules extracted from task labels (extract_scheduled_tasks_from_broker()).

This means any schedule added at runtime via add_schedule() (e.g. a one-shot time= task) is permanently deleted on the next startup() of any process that holds this source — including the taskiq scheduler CLI process on every restart/deploy.

As a result, it seems impossible to use a single AsyncpgScheduleSource for both:

• static cron tasks defined in code via @broker.task(schedule=[...]), and
• dynamic, persistent, runtime-added schedules via add_schedule().

The whole point of a DB-backed source (vs LabelScheduleSource) is persistence of dynamic schedules — but startup() wipes them. I'd like to confirm whether this is intended, and what the recommended pattern is.

Environment

• taskiq-postgres (module taskiq_pg) == 0.7.0
• taskiq == 0.12.1
• asyncpg == 0.30.0
• Python 3.13

Relevant source

In taskiq_pg/asyncpg/schedule_source.py:

async def _update_schedules_on_startup(self, schedules: list[ScheduledTask]) -> None:
    async with self._database_pool.acquire() as connection, connection.transaction():
        await connection.execute(DELETE_ALL_SCHEDULES_QUERY.format(self._table_name))  # <-- wipes EVERYTHING
        for schedule in schedules:                                                     # <-- only label-derived
            await self._database_pool.execute(INSERT_SCHEDULE_QUERY.format(...), ...)

async def startup(self) -> None:
    self._database_pool = await asyncpg.create_pool(dsn=self.dsn, **self._connect_kwargs)
    await self._database_pool.execute(CREATE_SCHEDULES_TABLE_QUERY.format(self._table_name))
    scheduled_tasks_for_creation = self.extract_scheduled_tasks_from_broker()
    await self._update_schedules_on_startup(scheduled_tasks_for_creation)

DELETE_ALL_SCHEDULES_QUERY = "DELETE FROM {};" (full truncate).

Minimal reproduction

import asyncio
from datetime import datetime, timezone, timedelta

from taskiq import ScheduledTask
from taskiq_aio_pika import AioPikaBroker
from taskiq_pg.asyncpg import AsyncpgScheduleSource

DSN = "postgresql://postgres:postgres@localhost:5432/postgres"
broker = AioPikaBroker("amqp://guest:guest@localhost:5672//")
source = AsyncpgScheduleSource(broker=broker, dsn=DSN, table_name="taskiq_schedules")


@broker.task(task_name="my_cron", schedule=[{"cron": "0 2 * * *"}])
async def my_cron() -> None: ...


@broker.task(task_name="my_dynamic")
async def my_dynamic(order_id: str) -> None: ...


async def main():
    # --- first boot: register a dynamic one-shot schedule ---
    await source.startup()
    await source.add_schedule(
        ScheduledTask(
            schedule_id="dyn-1",
            task_name="my_dynamic",
            time=datetime.now(timezone.utc) + timedelta(hours=1),
            kwargs={"order_id": "123"}, args=[], labels={},
        )
    )
    print("after add:", [s.task_name for s in await source.get_schedules()])
    # -> ['my_cron', 'my_dynamic']   (cron came from label, dynamic was added)
    await source.shutdown()

    # --- simulate a process restart / redeploy ---
    await source.startup()
    print("after restart:", [s.task_name for s in await source.get_schedules()])
    # -> ['my_cron']                 # <-- 'my_dynamic' is GONE (table was truncated)

    await source.shutdown()


asyncio.run(main())

Actual: after restart the dynamic schedule (my_dynamic) is gone — startup() truncated the table and re-inserted only the label cron.

Expected (or: desired): dynamically added schedules persist across restarts, while label crons are still synced. i.e. the DB source should not destroy rows it didn't create.

My real-world use case

A FastAPI app where:

• The API process calls source.add_schedule(ScheduledTask(time=order_expires_at, task_name="cancel_order_if_expired", ...)) whenever a pending order is created, so the order is auto-cancelled at its expiry time.
• A separate taskiq scheduler process runs the schedules and also serves a couple of static @broker.task(schedule=[...]) crons.

Every deploy of the scheduler (or the API) calls startup() → truncates the table → all pending order-cancellation schedules are lost.

Questions

1. Is the truncate-and-resync-from-labels behavior in startup() intended?
2. What is the recommended way to use AsyncpgScheduleSource for persistent dynamic schedules alongside label-defined crons, without losing the dynamic ones on restart?
3. Would you accept a change (or an option/flag) to make startup() non-destructive — e.g. upsert label schedules by a stable schedule_id instead of DELETE FROM, or a sync_labels=False switch? I'm happy to open a PR.

Currently my only workaround is subclassing and overriding startup() to create the pool + table without the truncate/label-sync, which works but bypasses the intended label-sync entirely.