FastScheduler（ファストスケジューラ）

FastScheduler は、Python 用の軽量で非同期に優しいタスクスケジューラです。タイムゾーン・cron 式・永続状態をサポートし、FastAPI が提供するリアルタイムダッシュボードと連携します。

主な特徴

async/await

インストール

# 基本インストール
pip install fastscheduler

# FastAPI ダッシュボード付き
pip install fastscheduler[fastapi]

# cron サポート付き
pip install fastscheduler[cron]

# すべてのオプションを有効にする
pip install fastscheduler[all]

クイックスタート

from fastscheduler import FastScheduler

scheduler = FastScheduler(quiet=True)

@scheduler.every(10).seconds
def task():
    print("Task executed")

@scheduler.daily.at("14:30")
async def daily_task():
    print("Daily task at 2:30 PM")

scheduler.start()

スケジューリングオプション

インターバルベース

@scheduler.every(10).seconds
@scheduler.every(5).minutes
@scheduler.every(2).hours
@scheduler.every(1).days

時間ベース

@scheduler.daily.at("09:00")               # 毎朝9時
@scheduler.hourly.at(":30")                # 毎時間30分に
@scheduler.weekly.monday.at("10:00")       # 月曜10時
@scheduler.weekly.weekdays.at("09:00")     # 平日9時
@scheduler.weekly.weekends.at("12:00")     # 週末正午

cron 式（

pip install fastscheduler[cron]

が必要）

@scheduler.cron("0 9 * * MON-FRI")
def market_open():
    ...

@scheduler.cron("*/15 * * * *")
def frequent_check():
    ...

@scheduler.cron("0 0 1 * *")
def monthly_report():
    ...

一度だけ実行

@scheduler.once(60)
def delayed_task():
    ...

@scheduler.at("2024-12-25 00:00:00")
def christmas_task():
    ...

タイムゾーンサポート

# tz パラメータ
@scheduler.daily.at("09:00", tz="America/New_York")
def nyc_morning(): ...

# チェーン可能な .tz()
@scheduler.weekly.monday.tz("Europe/London").at("09:00")
def london_standup(): ...

# cron でも
@scheduler.cron("0 9 * * MON-FRI").tz("Asia/Tokyo")
def tokyo_market(): ...

代表的なタイムゾーンは

UTC

America/New_York

Europe/London

Asia/Tokyo

ジョブ制御

タイムアウト

@scheduler.every(1).minutes.timeout(30)
def quick_task():
    ...

@scheduler.daily.at("02:00").timeout(3600)  # 最大1時間
def nightly_backup():
    ...

再試行

@scheduler.every(5).minutes.retries(5)
def flaky_api_call():
    ...
# 指数関数的バックオフ：2s, 4s, 8s …

カッチアップをスキップ

@scheduler.every(1).hours.no_catch_up()
def hourly_stats():
    ...

一時停止 / 再開 / キャンセル

scheduler.pause_job("job_0")
scheduler.resume_job("job_0")
scheduler.cancel_job("job_0")
scheduler.cancel_job_by_name("my_task")

FastAPI 連携

from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

app = FastAPI()
scheduler = FastScheduler(quiet=True)

app.include_router(create_scheduler_routes(scheduler))

@scheduler.every(30).seconds
def background_task():
    print("Background work")

scheduler.start()

ダッシュボードは

http://localhost:8000/scheduler/

API エンドポイント

/scheduler/

/scheduler/api/status

/scheduler/api/jobs

/scheduler/api/jobs/{job_id}

/scheduler/api/jobs/{job_id}/pause

/scheduler/api/jobs/{job_id}/resume

/scheduler/api/jobs/{job_id}/cancel

/scheduler/api/history

/scheduler/api/dead-letters

/scheduler/api/dead-letters

/scheduler/events

設定

scheduler = FastScheduler(
    state_file="scheduler.json",      # 永続ファイル
    quiet=True,                       # ログ抑制
    auto_start=False,                 # 手動起動
    max_history=5000,                 # 履歴最大件数
    max_workers=20,                   # 同時ジョブ数
    history_retention_days=8,         # 古い履歴削除期間
    max_dead_letters=500,             # デッドレターキューサイズ
)

履歴保持

• max_history

  （件数上限）

• history_retention_days

  （時間上限；0 で無効化）

デッドレターキュー

• *_dead_letters.json

  に保存
• 最新の

  max_dead_letters

  件を詳細付きで保持

モニタリング & プログラム的アクセス

jobs = scheduler.get_jobs()
job = scheduler.get_job("job_0")
history = scheduler.get_history(limit=100)
stats = scheduler.get_statistics()
scheduler.print_status()

コンテキストマネージャーで使用可能：

with FastScheduler(quiet=True) as scheduler:
    @scheduler.every(5).seconds
    def task():
        print("Running")
    time.sleep(30)   # スケジューラは自動実行
# 退出時に停止

状態永続化

• ジョブ定義・スケジュール・履歴・統計・ジョブカウンタをディスクへ保存
• 再起動時：ジョブ復元、欠落した実行を算出し、必要ならキャッチアップジョブを実行（

  no_catch_up()

  で除外可）

完全な例

import asyncio, time
from fastscheduler import FastScheduler

scheduler = FastScheduler(quiet=True)

@scheduler.every(10).seconds
def heartbeat():
    print(f"[{time.strftime('%H:%M:%S')}] ❤️ Heartbeat")

@scheduler.daily.at("09:00", tz="America/New_York").timeout(60)
async def morning_report():
    print("Generating report...")
    await asyncio.sleep(5)
    print("Report sent!")

@scheduler.cron("*/5 * * * *").retries(3)
def check_api():
    print("Checking API health")

@scheduler.weekly.monday.at("10:00")
def weekly_standup():
    print("Time for standup!")

scheduler.start()

try:
    while True:
        time.sleep(60)
        scheduler.print_status()
except KeyboardInterrupt:
    scheduler.stop()

FastAPI とライフサイクル

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

scheduler = FastScheduler(quiet=True)

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    yield
    scheduler.stop(wait=True)

app = FastAPI(lifespan=lifespan)
app.include_router(create_scheduler_routes(scheduler))

@scheduler.every(30).seconds
def background_job():
    print("Working...")

API 参照

start()

stop(wait=True, timeout=30)

get_jobs()

get_job(job_id)

get_history(func_name=None, limit=50)

get_statistics()

get_dead_letters(limit=100)

clear_dead_letters()

pause_job(job_id)

resume_job(job_id)

cancel_job(job_id)

cancel_job_by_name(func_name)

print_status()

ライセンス

MIT © 2024

貢献歓迎！GitHub で issue や PR を作成してください。