Database-Driven Cron Jobs with HexCore DynamicScheduler

DynamicScheduler evaluates a database-driven list of cron jobs on every tick and enqueues due tasks through ITaskEnqueuer. Because the schedule is stored in a database rather than hardcoded in configuration, you can add, modify, enable, or disable cron jobs at runtime — with zero application restarts. An optional ILockProvider (backed by Redis or PostgreSQL) prevents duplicate execution across multiple replicas.

Domain contracts

`CronJobDefinition`

A dataclass that describes a single scheduled task, read from your database by ICronJobRepository:

# hexcore/domain/cqrs/cron.py
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class CronJobDefinition:
    job_id: str
    task_name: str
    cron_expression: str
    payload: dict = field(default_factory=dict)
    queue: str = "default"
    is_active: bool = True
    last_run_at: datetime | None = None

job_id

str

required

Unique identifier for the job. Used as part of the distributed lock key.

task_name

str

required

Fully-qualified function name of the task to enqueue (e.g. "app.tasks.generate_report"). The worker resolves this via importlib at execution time.

cron_expression

str

required

Standard five-field cron expression (e.g. "0 * * * *" for hourly). Evaluated using the croniter library.

payload

dict

default:"{}"

Arbitrary keyword arguments forwarded to the task function when it is invoked.

queue

str

default:"default"

Name of the task queue to route this job to.

is_active

bool

default:"True"

When False, the scheduler skips this job entirely on the next tick.

last_run_at

datetime | None

default:"None"

Tracks the last execution time. The scheduler uses this to avoid re-enqueueing within the same tick. Updated by ICronJobRepository.update_last_run() after successful enqueue.

`ICronJobRepository`

Your application must implement this interface to feed jobs to the scheduler. Any ORM (SQLAlchemy, Beanie, Tortoise) works.

# hexcore/domain/cqrs/cron.py
import abc

class ICronJobRepository(abc.ABC):
    @abc.abstractmethod
    async def get_active_jobs(self) -> list[CronJobDefinition]:
        """Return all currently active job definitions from the database."""
        pass

    @abc.abstractmethod
    async def update_last_run(self, job_id: str, run_time: datetime) -> None:
        """Persist the last execution timestamp for a job."""
        pass

`ILockProvider`

# hexcore/domain/cqrs/cron.py
class ILockProvider(abc.ABC):
    @abc.abstractmethod
    async def acquire_lock(self, lock_key: str, ttl_seconds: int) -> bool:
        """Try to acquire a distributed lock. Returns True if acquired."""
        pass

    @abc.abstractmethod
    async def release_lock(self, lock_key: str) -> None:
        """Release the lock explicitly (TTL releases it automatically)."""
        pass

`DynamicScheduler`

# hexcore/application/cqrs/scheduler.py (simplified)
class DynamicScheduler:
    def __init__(
        self,
        repository: ICronJobRepository,
        enqueuer: ITaskEnqueuer,
        lock_provider: ILockProvider | None = None,
        tick_interval_seconds: int = 30,
    ) -> None: ...

    async def start(self) -> None:
        """Start the infinite scheduler loop (blocking coroutine)."""
        ...

    def stop(self) -> None:
        """Signal the loop to exit cleanly."""
        ...

repository

ICronJobRepository

required

Your implementation that reads active jobs from the database. Called on every tick so the schedule is always fresh.

enqueuer

ITaskEnqueuer

required

Task enqueuer used to dispatch due jobs. The job’s task_name and payload are forwarded directly via enqueuer.enqueue_task().

lock_provider

ILockProvider | None

default:"None"

Optional distributed lock provider. When set, the scheduler acquires a per-job-per-minute lock before enqueueing. If the lock is already held (by another replica), the job is skipped silently for that tick.

tick_interval_seconds

int

default:"30"

How often (in seconds) the scheduler wakes up to evaluate jobs. Set to 60 to match the smallest meaningful cron granularity.

How the tick loop works

On each tick the scheduler:

Reads all active jobs via repository.get_active_jobs() — always fresh from the database.
For each job, checks if croniter.croniter.match(cron_expression, current_time) returns True.
If a lock provider is configured, attempts to acquire hexcore:cron_lock:{job_id}:{minute_iso}. Skips the job if the lock is already held.
Calls enqueuer.enqueue_task(task_name, payload, queue).
Calls repository.update_last_run(job_id, current_time) to record execution.

croniter must be installed separately: pip install croniter. It is listed as an optional dependency of HexCore.

Implementing `ICronJobRepository`

SQLAlchemy
Beanie (MongoDB)

# app/scheduler/repository.py
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from hexcore.domain.cqrs.cron import ICronJobRepository, CronJobDefinition
from app.models import CronJobModel

class SqlCronJobRepository(ICronJobRepository):
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get_active_jobs(self) -> list[CronJobDefinition]:
        result = await self._session.execute(
            select(CronJobModel).where(CronJobModel.is_active == True)
        )
        rows = result.scalars().all()
        return [
            CronJobDefinition(
                job_id=row.id,
                task_name=row.task_name,
                cron_expression=row.cron_expression,
                payload=row.payload or {},
                queue=row.queue,
                is_active=row.is_active,
                last_run_at=row.last_run_at,
            )
            for row in rows
        ]

    async def update_last_run(self, job_id: str, run_time: datetime) -> None:
        await self._session.execute(
            update(CronJobModel)
            .where(CronJobModel.id == job_id)
            .values(last_run_at=run_time)
        )
        await self._session.commit()

# app/scheduler/repository.py
from datetime import datetime
from hexcore.domain.cqrs.cron import ICronJobRepository, CronJobDefinition
from app.documents import CronJobDocument

class BeanieConJobRepository(ICronJobRepository):
    async def get_active_jobs(self) -> list[CronJobDefinition]:
        docs = await CronJobDocument.find(
            CronJobDocument.is_active == True
        ).to_list()
        return [
            CronJobDefinition(
                job_id=str(doc.id),
                task_name=doc.task_name,
                cron_expression=doc.cron_expression,
                payload=doc.payload or {},
                queue=doc.queue,
                is_active=doc.is_active,
                last_run_at=doc.last_run_at,
            )
            for doc in docs
        ]

    async def update_last_run(self, job_id: str, run_time: datetime) -> None:
        doc = await CronJobDocument.get(job_id)
        if doc:
            doc.last_run_at = run_time
            await doc.save()

Distributed lock providers

When you run multiple replicas of your application (e.g. Kubernetes pods), multiple DynamicScheduler instances may wake up at the same second. Without a lock, the same job gets enqueued multiple times. Inject a ILockProvider to prevent this.

RedisLockProvider
PostgresLockProvider

Uses Redis SET NX EX — atomic, TTL-based, no deadlocks.

# hexcore/infrastructure/cqrs/redis_lock.py
from hexcore.infrastructure.cqrs.redis_lock import RedisLockProvider
import redis.asyncio as aioredis

redis_client = aioredis.from_url("redis://localhost:6379")
lock_provider = RedisLockProvider(redis_client=redis_client)

redis_client

redis.asyncio.Redis

required

An active async Redis client instance.

The lock key format is hexcore:cron_lock:{job_id}:{minute_iso}, where minute_iso is the current UTC time truncated to the minute. The TTL is 60 seconds, guaranteeing the lock expires even if the scheduler crashes mid-tick.

Uses an hexcore_cron_locks table with ON CONFLICT DO UPDATE ... WHERE expires_at < NOW() for atomic acquisition — no Redis required.

from hexcore.infrastructure.cqrs.postgres_lock import PostgresLockProvider
import asyncpg

pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")
lock_provider = PostgresLockProvider(pool=pool)

# Create the lock table on startup (idempotent)
await lock_provider.setup()

pool

asyncpg.Pool | asyncpg.Connection

required

An active asyncpg connection pool or single connection.

table_name

str

default:"hexcore_cron_locks"

Name of the lock table. Override if you have naming conventions or schema prefixes.

Full wiring example with FastAPI lifespan

Define a scheduled task

# app/tasks.py
from hexcore.domain.cqrs.decorators import background_task

@background_task(queue="reports")
async def generate_daily_summary(region: str) -> None:
    # This function runs inside the Celery worker
    await build_and_email_summary(region)

Implement the repository and configure the scheduler

# app/scheduler/setup.py
from hexcore.application.cqrs.scheduler import DynamicScheduler
from hexcore.infrastructure.cqrs.redis_lock import RedisLockProvider
from app.scheduler.repository import SqlCronJobRepository
from app.celery_app import celery_enqueuer
import redis.asyncio as aioredis
from sqlalchemy.ext.asyncio import AsyncSession

def create_scheduler(session: AsyncSession) -> DynamicScheduler:
    redis_client = aioredis.from_url("redis://localhost:6379")

    return DynamicScheduler(
        repository=SqlCronJobRepository(session),
        enqueuer=celery_enqueuer,
        lock_provider=RedisLockProvider(redis_client),
        tick_interval_seconds=60,
    )

Start and stop with FastAPI lifespan

# app/main.py
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db import get_session
from app.scheduler.setup import create_scheduler

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start scheduler in background task
    async with get_session() as session:
        scheduler = create_scheduler(session)
        scheduler_task = asyncio.create_task(scheduler.start())

    yield  # Application is running

    # Graceful shutdown
    scheduler.stop()
    scheduler_task.cancel()
    try:
        await scheduler_task
    except asyncio.CancelledError:
        pass

app = FastAPI(lifespan=lifespan)

Seed job definitions in the database

INSERT INTO cron_jobs (id, task_name, cron_expression, payload, queue, is_active)
VALUES
  ('daily-summary-eu',  'app.tasks.generate_daily_summary', '0 6 * * *', '{"region": "EU"}', 'reports', true),
  ('daily-summary-us',  'app.tasks.generate_daily_summary', '0 14 * * *', '{"region": "US"}', 'reports', true);

Now you can update the cron_expression or payload columns at runtime and the scheduler picks up the change on the very next tick — no restart required.

Updating schedules at runtime via API

# app/routers/scheduler.py
from fastapi import APIRouter
from app.db import get_session
from app.models import CronJobModel
from sqlalchemy import update

router = APIRouter(prefix="/scheduler")

@router.patch("/jobs/{job_id}")
async def update_job(job_id: str, cron_expression: str):
    async with get_session() as session:
        await session.execute(
            update(CronJobModel)
            .where(CronJobModel.id == job_id)
            .values(cron_expression=cron_expression)
        )
        await session.commit()
    return {"updated": job_id, "cron": cron_expression}

Always validate cron expressions before persisting them. An invalid expression will cause croniter to raise an exception on every scheduler tick until the bad record is corrected. Consider adding a croniter.croniter.is_valid(expr) check in your update endpoint.

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Database-Driven Cron Jobs with HexCore DynamicScheduler

Domain contracts

`CronJobDefinition`

`ICronJobRepository`

`ILockProvider`

`DynamicScheduler`

How the tick loop works

Implementing `ICronJobRepository`

Distributed lock providers

Full wiring example with FastAPI lifespan

Updating schedules at runtime via API

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Documentation Index

​Domain contracts

​CronJobDefinition

​ICronJobRepository

​ILockProvider

​DynamicScheduler

​How the tick loop works

​Implementing ICronJobRepository

​Distributed lock providers

​Full wiring example with FastAPI lifespan

​Updating schedules at runtime via API

Build docs developers (and LLMs) love

Domain contracts

`CronJobDefinition`

`ICronJobRepository`

`ILockProvider`

`DynamicScheduler`

How the tick loop works

Implementing `ICronJobRepository`

Distributed lock providers

Full wiring example with FastAPI lifespan

Updating schedules at runtime via API