Overview

The state backend remembers when each job last ran and how it finished, so a restart does not lose that history. If you pass nothing, a SQLite backend is created using CRON_SQLITE_DB_PATH.

Backend	Import from	Extra needed	Good for
`SQLiteStateBackend`	`fastapi_crons.state`	—	Single instance, local development. The default.
`RedisStateBackend`	`fastapi_crons.state`	—	Several replicas sharing state.
`SQLAlchemyStateBackend`	`fastapi_crons.state.sqlalchemy`	`sqlalchemy`	Reusing the database your app already has.
`SQLModelStateBackend`	`fastapi_crons.state.sqlalchemy`	`sqlmodel`	The same, in a SQLModel codebase.

The SQL backends are not exported at the top level

Import them from fastapi_crons.state.sqlalchemy, not from fastapi_crons. Importing without the extra installed raises an ImportError that names the extra to install.

SQLite

Zero configuration and the default. Writes to a single file through aiosqlite.

from fastapi_crons import Crons
from fastapi_crons.state import SQLiteStateBackend

state = SQLiteStateBackend(db_path="cron_state.db")
crons = Crons(app, state_backend=state)

One writer only

SQLite does not cope with several processes writing concurrently over a network filesystem. Use Redis or a SQL backend once you run more than one replica.

Redis

Shares state across replicas and keeps writes off your primary database.

import redis.asyncio as redis
from fastapi_crons import Crons
from fastapi_crons.state import RedisStateBackend

client = redis.from_url("redis://localhost:6379")
state = RedisStateBackend(client)
crons = Crons(app, state_backend=state)

SQLAlchemy

Stores state in the database your application already uses. Both sync and async engines are accepted — the backend detects which one it was given.

from sqlalchemy.ext.asyncio import create_async_engine
from fastapi_crons import Crons
from fastapi_crons.state.sqlalchemy import SQLAlchemyStateBackend

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/app")
state = SQLAlchemyStateBackend(engine)
crons = Crons(app, state_backend=state)

Parameters

engine

Engine | AsyncEngine

A SQLAlchemy engine. Async engines are used directly; sync engines are run in a thread.

create_tables

bool

Create the backing tables on first use. Defaults to True.

Three tables are created: job_state for last-run times, job_status for the current status, and job_execution_log for history.

Managing the schema yourself

If you own the schema with Alembic, disable automatic creation:

state = SQLAlchemyStateBackend(engine, create_tables=False)

SQLModel

SQLModelStateBackend is an alias of SQLAlchemyStateBackend. SQLModel engines are SQLAlchemy engines, so no separate implementation is needed — the alias exists to make intent obvious in SQLModel projects.

from sqlmodel import create_engine
from fastapi_crons.state.sqlalchemy import SQLModelStateBackend

# SQLModel engines are SQLAlchemy engines, so the same backend handles both.
engine = create_engine("postgresql://user:pass@localhost/app")
state = SQLModelStateBackend(engine)

Writing your own

Subclass StateBackend from fastapi_crons.state and implement its abstract methods — get_last_run, set_last_run, get_job_status and set_job_status — then pass an instance as state_backend.