APScheduler и Dramatiq

19.06.2026 10:44 · edited 19.06.2026 11:07

Введение

В экосистеме Python существует множество решений для обработки фоновых задач и планирования. Два популярных инструмента — APScheduler (Advanced Python Scheduler) и Dramatiq — часто сравниваются, хотя на самом деле решают разные классы задач. APScheduler — это in-process планировщик задач, тогда как Dramatiq — распределённая очередь задач (task queue) с собственными воркерами.

Обзор и стек технологий

APScheduler: In-Process планировщик

APScheduler (Advanced Python Scheduler) — это лёгкий планировщик задач, работающий внутри вашего процесса. Он не требует внешних брокеров сообщений и может использоваться как самостоятельное решение для периодических задач.

Ключевые характеристики:
- Работает в том же процессе, что и ваше приложение
- Не требует внешних зависимостей (брокеров, БД)
- Поддерживает синхронный и асинхронный (asyncio/Trio) режимы
- Лицензия: MIT

Стек:
- Язык: Python 3.8+
- Хранилища задач (Job Stores): in-memory, SQLAlchemy (любая БД), Redis, MongoDB, ZooKeeper
- Исполнители (Executors): ThreadPool, ProcessPool, асинхронные (asyncio)

Вес библиотеки: APScheduler — это лёгкая библиотека без внешних зависимостей. Установочный пакет минимален, и в отличие от решений на основе брокеров, не требует развёртывания дополнительной инфраструктуры.

Dramatiq: Распределённая очередь задач

Dramatiq — это библиотека для фоновой обработки задач с фокусом на простоту, надёжность и производительность. Она вдохновлена Sidekiq (Ruby) и создавалась как более лёгкая и интуитивная альтернатива Celery.

Ключевые характеристики:
- Работает в отдельном процессе (воркере), отдельно от веб-приложения
- Требует внешний брокер сообщений (Redis или RabbitMQ)
- Поддерживает распределённую обработку (несколько воркеров на разных машинах)
- Лицензия: LGPL

Стек:
- Язык: Python 3.10+
- Брокеры: Redis или RabbitMQ
- Сериализация: по умолчанию JSON, возможны кастомные кодеки

Вес библиотеки: Dramatiq легче Celery — при обработке 200K задач пиковое потребление памяти составило ~73 МБ против ~197 МБ у Celery. Однако требуется Redis или RabbitMQ, что добавляет инфраструктурный вес.

Возможности и API

APScheduler: API и компоненты

APScheduler построен на четырёх основных компонентах:

Триггеры (Triggers) — определяют логику расписания:
DateTrigger — однократный запуск в указанное время
IntervalTrigger — периодический запуск с фиксированным интервалом
CronTrigger — гибкое расписание в стиле cron
Хранилища задач (Job Stores) — сохраняют состояние задач:
MemoryJobStore — по умолчанию, данные в RAM
SQLAlchemyJobStore — в реляционной БД
RedisJobStore, MongoDBJobStore и др.
Исполнители (Executors) — определяют как выполнять задачи:
ThreadPoolExecutor — для I/O-связанных задач
ProcessPoolExecutor — для CPU-связанных задач
Асинхронные исполнители для asyncio
Планировщики (Schedulers) — связывают всё воедино:
BlockingScheduler — блокирующий, для автономных скриптов
BackgroundScheduler — фоновый, для интеграции с приложениями
AsyncIOScheduler — для asyncio-приложений
GeventScheduler, TornadoScheduler, TwistedScheduler, QtScheduler

Пример использования APScheduler:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime

def my_task():
    print(f"Task executed at {datetime.now()}")

scheduler = BackgroundScheduler()
scheduler.add_job(my_task, 'interval', seconds=10)
scheduler.add_job(my_task, CronTrigger(hour=9, minute=30))
scheduler.start()

API Reference включает классы Task, TaskDefaults, Schedule для тонкой настройки каждого задания:
- max_running_jobs — ограничение одновременных экземпляров задачи
- misfire_grace_time — допустимое опоздание выполнения
- coalesce — политика объединения пропущенных запусков
- job_executor — выбор конкретного исполнителя для задачи

Dramatiq: API и компоненты

Dramatiq построен вокруг концепции акторов (actors) и сообщений (messages).

Основные элементы:

Актор (Actor) — декорированная функция, которая может обрабатывать сообщения:

import dramatiq

@dramatiq.actor
def count_words(url):
    response = requests.get(url)
    return len(response.text.split())

Брокер (Broker) — посредник для сообщений (Redis или RabbitMQ):

# Настройка брокера
from dramatiq.brokers.redis import RedisBroker
broker = RedisBroker(host="localhost", port=6379)
dramatiq.set_broker(broker)

Воркер (Worker) — процесс, выполняющий задачи. Запускается через CLI:

dramatiq my_module

Middleware — система плагинов для расширения функциональности:
Retries — автоматические повторные попытки с экспоненциальной задержкой
Rate limiting — ограничение скорости выполнения
Prometheus — метрики и мониторинг
Sentry — интеграция с Sentry для отлова ошибок

Ключевые методы API:
- actor.send(*args, **kwargs) — отправить задачу в очередь
- actor.send_with_options(options) — с опциями задержки, приоритета и т.д.
- dramatiq.get_broker().flush_all() — очистка очередей

Пример сложного сценария:

@dramatiq.actor(max_retries=3, min_backoff=1000, max_backoff=60000)
def process_order(order_id):
    # Обработка с автоматическими повторами при ошибках
    pass

# Отправка с задержкой
process_order.send_with_options(
    args=(order_id,),
    delay=60000  # мс
)

Нюансы и подводные камни

APScheduler

1. Работа в распределённой среде
APScheduler не предназначен для распределённых систем по умолчанию. Если запустить несколько экземпляров приложения с APScheduler, задачи будут выполняться несколько раз. Решение — использование внешнего хранилища с блокировками (например, Redis с распределёнными блокировками) или запуск планировщика только на одной ноде.

2. Пропущенные выполнения (Misfires)
Если планировщик был остановлен или перегружен, задачи могут пропустить своё окно запуска. APScheduler предоставляет механизм misfire_grace_time и политику coalesce для управления этим поведением.

3. Сериализация
При использовании постоянных хранилищ (SQLAlchemy, Redis) задачи сериализуются. Важно, чтобы все аргументы функции были сериализуемы (pickle-совместимы).

4. Асинхронность
APScheduler 4.x (в альфе) добавляет полноценную поддержку asyncio. В версии 3.x асинхронная работа ограничена.

5. Производительность при большом количестве задач
При одновременной постановке тысяч задач рекомендуется использовать пакетную обработку. Выбор исполнителя критичен: для CPU-задач — ProcessPoolExecutor, для I/O — ThreadPoolExecutor.

Dramatiq

1. Требования к брокеру
Dramatiq поддерживает только Redis и RabbitMQ. В отличие от Celery, поддержка SQS и других брокеров отсутствует из коробки.

2. Модель доставки
Dramatiq использует модель "at-least-once" (как минимум один раз), что означает возможные дублирующиеся выполнения. Для критичных операций требуется реализация идемпотентности.

3. Отсутствие встроенного планировщика
В отличие от Celery Beat, Dramatiq не имеет встроенного планировщика для периодических задач. Для cron-подобных задач можно использовать APScheduler в связке с Dramatiq.

4. Обработка ошибок
По умолчанию Dramatiq не перезапускает бесконечно упавшие задачи — это контролируется через middleware и настройки retry.

5. Мониторинг
Dramatiq не предоставляет встроенного веб-интерфейса для мониторинга (в отличие от Flower для Celery). Возможна интеграция с Prometheus или использование сторонних решений.

Производительность и нагрузка

Сравнительное бенчмарки

Согласно бенчмарку 2024–2025 годов (обработка 20,000 задач, 10 воркеров, Redis как брокер):

Библиотека	Время (сек)	Тип воркеров
Taskiq	2.03	—
Huey	3.62–4.15	Threads/Processes
Dramatiq	4.12–4.35	Threads/Processes
Celery	11.68–17.60	Threads/Processes
ARQ	35.37	—
RQ	51.05	—

Dramatiq показал результат, близкий к Huey, и почти в 3 раза быстрее Celery (в режиме потоков).

Потребление памяти

Бенчмарк с 200,000 задач через 10 очередей на Redis 7.4 показал:

Библиотека	Пиковая память (МБ)
Taskiq	33
Sidekiq	34
Dramatiq	73
RQ	100
Laravel	106
Celery	197

Dramatiq занимает промежуточное положение — значительно легче Celery, но тяжелее Taskiq и Sidekiq.

Оптимизация

Dramatiq поддерживает сжатие полезной нагрузки (например, через lz4), что может значительно снизить потребление памяти — в одном из примеров сообщение 30 МБ было сжато до 6 МБ.

Сравнение с аналогами

Общая классификация

Инструмент	Тип	Брокер	Планирование	Распределённость
APScheduler	Планировщик	Не требуется	Встроенное	❌ (одна нода)
Dramatiq	Очередь задач	Redis/RabbitMQ	Внешнее (APScheduler)	✅
Celery	Очередь задач	Redis/RabbitMQ/SQS/др.	Celery Beat	✅
RQ	Очередь задач	Redis	❌	✅
Huey	Очередь задач	Redis/SQLite	Встроенное	✅
Taskiq	Очередь задач	Redis/RabbitMQ/Kafka	Встроенное	✅
Schedule	Планировщик	Не требуется	Встроенное	❌

APScheduler vs аналоги

Характеристика	APScheduler	Schedule	Celery Beat
Сложность	Низкая	Очень низкая	Высокая
Cron-поддержка	✅	✅ (ограниченно)	✅
Постоянное хранение	✅ (SQLAlchemy, Redis, etc)	❌	✅
Асинхронность	✅ (asyncio)	❌	❌
Распределённость	❌ (с оговорками)	❌	✅
Вес	Лёгкий	Минимальный	Тяжёлый

APScheduler — это выбор, когда вам нужен надёжный планировщик внутри приложения, но вы не хотите разворачивать внешние системы. Для простых сценариев (запуск раз в день) можно использовать библиотеку schedule, но она не предоставляет постоянного хранения и устойчивости к сбоям.

Dramatiq vs аналоги

Характеристика	Dramatiq	Celery	RQ	Huey
Простота настройки	Высокая	Низкая	Высокая	Средняя
Брокеры	Redis, RabbitMQ	Множество	Redis	Redis, SQLite
Производительность	Высокая	Средняя	Низкая	Высокая
Потребление памяти	Среднее (73 МБ)	Высокое (197 МБ)	Среднее (100 МБ)	—
Встроенный мониторинг	❌	✅ (Flower)	❌	❌
Планирование	Внешнее	✅ (Beat)	❌	✅
Retry-механизмы	✅	✅	Ограниченно	✅

Dramatiq создавался как более лёгкая и интуитивная альтернатива Celery. Он выигрывает в простоте использования и производительности, но уступает в количестве поддерживаемых брокеров и наличии встроенного планировщика.

RQ ещё проще, но значительно медленнее Dramatiq.

Huey — ближайший конкурент Dramatiq по производительности, но имеет более ограниченную экосистему.

Когда что выбирать

Выбираем APScheduler, если:

✅ Вам нужно периодическое выполнение задач (cron-подобное расписание) внутри одного приложения

✅ Вы не хотите разворачивать Redis или RabbitMQ

✅ Ваше приложение работает на одной ноде (или вы готовы реализовать распределённые блокировки)

✅ Вы используете asyncio и хотите встроенный планировщик в асинхронное приложение

❌ Не выбирайте, если вам нужна распределённая обработка тысяч задач в секунду или горизонтальное масштабирование.

Выбираем Dramatiq, если:

✅ Вам нужно асинхронное выполнение тяжёлых задач вне HTTP-цикла

✅ У вас уже есть Redis или RabbitMQ в инфраструктуре

✅ Вы хотите распределённую обработку с несколькими воркерами

✅ Вы ищете более простую альтернативу Celery

✅ Вам важна производительность — Dramatiq в разы быстрее RQ и Celery

❌ Не выбирайте, если вам нужен только периодический запуск задач без брокера — используйте APScheduler.

Комбинированное использование

Оба инструмента могут работать в связке: APScheduler генерирует периодические задачи и отправляет их в Dramatiq для асинхронного выполнения. Это даёт лучшее из двух миров: гибкое расписание APScheduler + распределённую обработку Dramatiq.

Практические примеры использования

APScheduler в действии

Базовый пример с BackgroundScheduler

Наиболее распространённый сценарий — запуск планировщика в фоновом режиме внутри уже работающего приложения:

from apscheduler.schedulers.background import BackgroundScheduler
import time

def hello():
    print("Hello, world!")

scheduler = BackgroundScheduler()
scheduler.add_job(hello, "interval", seconds=5)
scheduler.start()

try:
    while True:
        time.sleep(2)
except (KeyboardInterrupt, SystemExit):
    scheduler.shutdown()

Этот скрипт будет выводить "Hello, world!" каждые пять секунд.

Cron-подобное расписание

APScheduler поддерживает полноценные cron-выражения без необходимости настройки системного crontab:

from apscheduler.schedulers.blocking import BlockingScheduler

def nightly_job():
    print("Running nightly cleanup...")

scheduler = BlockingScheduler()
scheduler.add_job(nightly_job, "cron", hour=2, minute=30)  # ежедневно в 2:30
scheduler.start()

Передача параметров в задачи

Реальные задачи часто требуют контекста — идентификаторов пользователей, имён файлов или токенов API:

def send_email(user_email):
    print(f"Sending email to {user_email}")

scheduler.add_job(send_email, "interval", minutes=10, args=["team@company.com"])

Обработка ошибок

В продакшене сбои неизбежны. APScheduler позволяет настроить логирование, чтобы задачи не исчезали бесследно:

import logging
logging.basicConfig()
logging.getLogger("apscheduler").setLevel(logging.DEBUG)

def risky_job():
    raise Exception("Something went wrong!")

scheduler.add_job(risky_job, "interval", seconds=30)

Логирование будет фиксировать каждый сбой, что упрощает отладку. В одном из проектов разработчики даже настроили отправку уведомлений об ошибках в Slack непосредственно из логов.

APScheduler 4.0: асинхронный подход

APScheduler 4.0 (на данный момент в стадии альфа) полностью переработан для нативной поддержки асинхронности. Ключевые изменения включают:

Полная совместимость с SQLAlchemy AsyncEngine — в 3.x использование асинхронного движка приводило к ошибкам 'AsyncEngine' object has no attribute '_run_ddl_visitor'
Новая событийная модель — введено понятие "event broker" для более гибкого управления триггерами
AsyncScheduler — центральный класс для асинхронного планирования

Интеграция с FastAPI через lifespan

Правильная интеграция APScheduler 4.0 с FastAPI требует использования механизма lifespan для управления жизненным циклом планировщика:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from apscheduler import AsyncScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.datastores.memory import MemoryDataStore
from apscheduler.eventbrokers.local import LocalEventBroker

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncScheduler(MemoryDataStore(), LocalEventBroker()) as scheduler:
        await scheduler.start_in_background()
        await scheduler.add_schedule(
            tick, 
            IntervalTrigger(seconds=1), 
            id="tick"
        )
        yield
        await scheduler.stop()
        await scheduler.wait_until_stopped()

app = FastAPI(lifespan=lifespan)

Важное предостережение: при запуске с несколькими воркерами (например, через uvicorn) каждый воркер запускает собственный экземпляр планировщика. APScheduler 4.x поддерживает многопоточную работу, но для сценариев, требующих единственного экземпляра, планировщик следует выносить в отдельный процесс.

Производственные рекомендации для APScheduler 4.0:
- Для хранения задач предпочтительнее использовать PostgreSQL — эта БД прошла наиболее полное тестирование
- Оцените требования к стабильности перед внедрением альфа-версии в production
- Для высокой доступности запускайте APScheduler под управлением процессного менеджера (supervisord или systemd)

Dramatiq в действии

Базовый пример

import dramatiq

@dramatiq.actor
def count_words(url):
    import requests
    response = requests.get(url)
    return len(response.text.split())

# Отправка задачи в очередь
count_words.send("https://example.com")

Отправка с опциями (задержка, приоритет)

# Отложенный запуск через 60 секунд
count_words.send_with_options(
    args=("https://example.com",),
    delay=60000  # миллисекунды
)

Автоматические повторные попытки

@dramatiq.actor(max_retries=3, min_backoff=1000, max_backoff=60000)
def process_order(order_id):
    # Обработка с автоматическими повторами при сбоях
    # Экспоненциальная задержка: 1с, 2с, 4с, 8с...
    pass

Интеграция с веб-фреймворками

APScheduler + FastAPI (корректная работа с жизненным циклом)

Выше уже был показан пример с AsyncScheduler. Ключевые принципы:

Инициализация планировщика в lifespan — запуск при старте приложения, остановка при завершении
Единая точка управления задачами — все задачи добавляются централизованно
Корректная очистка ресурсов — использование wait_until_stopped() для гарантированного завершения

APScheduler + Flask

Для Flask-приложений используется BackgroundScheduler, запускаемый вместе с приложением:

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)
scheduler = BackgroundScheduler()

@app.before_first_request
def init_scheduler():
    scheduler.add_job(my_task, "interval", seconds=10)
    scheduler.start()

@app.teardown_appcontext
def shutdown_scheduler(exception=None):
    scheduler.shutdown()

Dramatiq + Django

Официальное расширение django_dramatiq обеспечивает бесшовную интеграцию:

# settings.py
INSTALLED_APPS = [
    ...
    'django_dramatiq',
]

DRAMATIQ_BROKER = {
    "BROKER": "dramatiq.brokers.redis.RedisBroker",
    "OPTIONS": {
        "url": "redis://localhost:6379/0",
    },
}

# tasks.py
import dramatiq

@dramatiq.actor
def send_welcome_email(user_id):
    user = User.objects.get(id=user_id)
    # отправка email

Запуск воркера:

python manage.py rundramatiq

Dramatiq + Flask

Существует расширение Flask-Melodramatiq, предоставляющее тонкие обёртки вокруг брокеров Dramatiq:

from flask import Flask
from flask_melodramatiq import Dramatiq

app = Flask(__name__)
dramatiq = Dramatiq(app)

@dramatiq.actor
def send_notification(email):
    # отправка уведомления
    pass

Проблема Actor в Dramatiq

При интеграции с веб-фреймворками возникает типичная проблема: Actor, определённые на уровне модуля, регистрируются в момент импорта, когда брокер ещё может быть не настроен.

Решение: использование lazy_loader для отложенной загрузки модулей с задачами:

# actions/__init__.py
import lazy_loader as lazy
__getattr__, __dir__, __all__ = lazy.attach(
    __name__,
    submodules=['tasks']  # задачи загрузятся только при первом обращении
)

Это гарантирует, что все Actor будут зарегистрированы только после того, как брокер полностью сконфигурирован.

Мониторинг и наблюдаемость

Dramatiq + Prometheus

Dramatiq имеет встроенный middleware для экспорта метрик в формате Prometheus:

from dramatiq.middleware.prometheus import Prometheus

broker = RedisBroker()
broker.add_middleware(Prometheus(port=9191))
dramatiq.set_broker(broker)

После запуска воркера метрики доступны по адресу localhost:9191. Prometheus middleware больше не зависит от file locking — сервер экспозиции запускается в отдельном процессе.

Доступные метрики включают:
- Количество выполненных задач
- Время выполнения
- Количество ошибок
- Размер очередей

Существует также готовый дашборд для Grafana.

Кастомные middleware для мониторинга

Разработчики могут создавать собственные middleware для сбора специфических метрик — например, гистограммы времени постановки в очередь или счётчики успешных/неудачных постановок.

APScheduler: логирование

APScheduler не имеет встроенной Prometheus-интеграции, но предоставляет детальное логирование через стандартный модуль logging. Для production-систем рекомендуется:

Настроить сбор логов APScheduler в централизованную систему (ELK, Loki)
Использовать кастомные обработчики для отправки метрик в Prometheus
Мониторить состояние через проверку выполнения критических задач

Развёртывание и эксплуатация

APScheduler в production

Ключевые рекомендации:

Используйте постоянное хранилище — MemoryJobStore теряет все задачи при перезапуске. Для production обязательно применяйте SQLAlchemyJobStore или RedisJobStore.
Обеспечьте идемпотентность задач — при сбоях и повторных запусках задача не должна создавать побочные эффекты.
Настройте max_instances — ограничьте количество одновременно выполняющихся экземпляров одной задачи, чтобы избежать перегрузки.
Используйте coalesce — при пропуске нескольких запусков объединяйте их в один, чтобы избежать накопления отложенных выполнений.
Запускайте под supervisord/systemd — для автоматического перезапуска при сбоях.
Управляйте временными зонами — всегда явно указывайте часовой пояс при использовании cron-триггеров.

Dramatiq в production

Ключевые рекомендации:

Ограничивайте параметры Actor — Dramatiq сериализует аргументы в JSON, поэтому передавайте только bool, int, float, bytes, string, list, dict.
Настройка повторных попыток — используйте max_retries, min_backoff, max_backoff для управления поведением при сбоях.
Мониторинг через Prometheus — встроенный middleware предоставляет все необходимые метрики.
Graceful shutdown — корректно завершайте воркеры, чтобы не терять выполняющиеся задачи.
Масштабирование — Dramatiq поддерживает горизонтальное масштабирование: просто запускайте дополнительные воркеры на разных машинах, подключённых к одному брокеру.

Миграция с Celery на Dramatiq

Celery долгое время был стандартом де-факто для распределённых задач в Python, но его сложность и "тяжеловесность" побуждают многие команды искать альтернативы. Dramatiq предлагает практически тот же набор функций, но значительно проще в настройке и быстрее.

Сравнение синтаксиса

Celery:

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def add(x, y):
    return x + y

add.delay(4, 4)

Dramatiq:

import dramatiq

@dramatiq.actor
def add(x, y):
    return x + y

add.send(4, 4)

Ключевые отличия при миграции

Аспект	Celery	Dramatiq
Конфигурация	Сложная, множество опций	Минималистичная
Поддерживаемые брокеры	Redis, RabbitMQ, SQS, и др.	Redis, RabbitMQ
Встроенное планирование	✅ (Celery Beat)	❌ (требуется APScheduler)
Мониторинг	✅ (Flower)	✅ (Prometheus)
Сложные workflows	✅ (canvas: chain, group, chord)	⚠️ (через `dramatiq-workflow`)
Время миграции	—	~1 неделя для большинства задач

Когда миграция оправдана

Ваша текущая инфраструктура на Celery кажется излишне сложной
Вы используете только Redis или RabbitMQ (SQS не требуется)
Вам нужна более высокая производительность при меньшем потреблении памяти
Вы хотите сократить время настройки и поддержки

Что следует учитывать: если вы активно используете Celery Canvas (chain, group, chord) для сложных пайплайнов, вам потребуется dramatiq-workflow — сторонняя библиотека, реализующая аналогичную функциональность.

Связка APScheduler + Dramatiq: лучшее из двух миров

Оба инструмента отлично дополняют друг друга. APScheduler берёт на себя планирование, а Dramatiq — асинхронное выполнение в распределённой среде.

Архитектура

┌─────────────────────────────────────────────────────────┐
│                    Веб-приложение                        │
│  ┌─────────────────┐          ┌─────────────────────┐  │
│  │   APScheduler   │ ──────▶ │   Dramatiq Actor    │  │
│  │  (планировщик)  │          │  (отправка задачи)  │  │
│  └─────────────────┘          └─────────────────────┘  │
└─────────────────────────────────────────────────────────┘
                           │
                           ▼
                  ┌───────────────┐
                  │  Redis/RMQ    │
                  │  (брокер)     │
                  └───────────────┘
                           │
         ┌─────────────────┼─────────────────┐
         ▼                 ▼                 ▼
   ┌──────────┐     ┌──────────┐     ┌──────────┐
   │ Воркер 1 │     │ Воркер 2 │     │ Воркер 3 │
   └──────────┘     └──────────┘     └──────────┘

Пример реализации

# tasks.py — определение Dramatiq-задач
import dramatiq

@dramatiq.actor
def send_report(email, report_data):
    # тяжёлая операция — генерация и отправка отчёта
    pass

# scheduler.py — APScheduler с отправкой задач в Dramatiq
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from tasks import send_report

def schedule_daily_report():
    scheduler = BackgroundScheduler()

    # Каждый день в 9:00 отправляем задачу в Dramatiq
    scheduler.add_job(
        lambda: send_report.send("admin@company.com", {"date": "today"}),
        CronTrigger(hour=9, minute=0)
    )

    scheduler.start()
    return scheduler

Существует также готовый пакет dramatiq-apscheduler, упрощающий эту интеграцию.

Преимущества такого подхода

Надёжное планирование — APScheduler с постоянным хранилищем гарантирует, что задачи не будут потеряны
Масштабируемое выполнение — Dramatiq распределяет нагрузку между воркерами
Разделение ответственности — планирование и выполнение изолированы друг от друга
Горизонтальное масштабирование — можно добавлять воркеры без изменения планировщика

Итог

APScheduler: идеальный выбор, когда...

Критерий	Оценка
Сложность настройки	★☆☆☆☆ (очень просто)
Внешние зависимости	Нет (опционально — БД/Redis)
Распределённость	❌ (требует доработок)
Асинхронность	✅ (4.x)
Персистентность	✅ (SQLAlchemy, Redis, MongoDB)
Production-готовность 4.x	⚠️ (альфа)

Рекомендуется для: cron-подобных задач внутри одного приложения, микросервисов на одной ноде, асинхронных FastAPI-приложений.

Dramatiq: идеальный выбор, когда...

Критерий	Оценка
Сложность настройки	★★☆☆☆ (просто)
Внешние зависимости	Redis или RabbitMQ
Распределённость	✅ (из коробки)
Асинхронность	❌ (синхронные воркеры)
Производительность	Высокая (4.12–4.35 сек на 20K задач)
Потребление памяти	Среднее (~73 МБ)

Рекомендуется для: выноса тяжёлых операций из веб-приложений, распределённой обработки, замены Celery.

Комбинированное использование

Рекомендуется для: сложных систем, где требуется и гибкое планирование, и распределённое выполнение. APScheduler генерирует задачи по расписанию, Dramatiq обрабатывает их асинхронно.

Заключение

APScheduler и Dramatiq решают принципиально разные задачи, и их сравнение корректно лишь в контексте конкретного Use Case:

Сценарий	Инструмент
"Мне нужно запускать отчёт каждый день в 9 утра"	APScheduler
"Мне нужно обрабатывать тысячи загрузок файлов асинхронно"	Dramatiq
"Мне нужно и то, и другое в одном проекте"	APScheduler + Dramatiq

Обе библиотеки активно развиваются: APScheduler движется к стабильному релизу 4.0 с полноценной асинхронной поддержкой, Dramatiq выпустил версию 2.0 с улучшенной типизацией и опциональным Prometheus.
Выбор между ними — это не вопрос "что лучше", а вопрос "что подходит для моей архитектуры". И в этом смысле у Python-разработчика сегодня есть все необходимые инструменты для построения надёжных и производительных систем фоновой обработки.

Шпаргалка по API

APScheduler. Шпаргалка

Основные структуры данных (APScheduler 4.x)

Важно: В APScheduler 4.x (альфа) изменена модель данных. Вместо Job теперь используются Task (что выполнять) и Schedule (когда выполнять).

`Task` — описание задачи

from apscheduler import Task

# Что именно будет выполняться
Task(
    id="send_emails",                     # уникальный идентификатор
    func=my_function,                     # вызываемый объект
    job_executor="default",               # имя исполнителя
    max_running_jobs=1,                   # макс. параллельных экземпляров
    misfire_grace_time=timedelta(seconds=30), # допустимое опоздание
    metadata={"team": "backend"}          # произвольные JSON-данные
)

`TaskDefaults` — значения по умолчанию

TaskDefaults(
    job_executor="default",               # исполнитель по умолчанию
    max_running_jobs=1,                   # макс. экземпляров по умолчанию
    misfire_grace_time=None,              # None = запускать при любом опоздании
)

`Schedule` — расписание задачи

Schedule(
    id="daily_report",                    # уникальный ID расписания
    task_id="send_emails",                # ID задачи из Task
    trigger=CronTrigger(hour=9),          # триггер (см. ниже)
    args=("user@example.com",),           # позиционные аргументы
    kwargs={"format": "pdf"},             # именованные аргументы
    paused=False,                         # приостановлено?
    coalesce=CoalescePolicy.latest,       # политика объединения пропусков
    misfire_grace_time=None,              # переопределение для расписания
    max_jitter=None,                      # макс. случайная задержка (сек)
    job_executor="default",               # исполнитель для этого расписания
    metadata={}                           # произвольные JSON-данные
)

Типы планировщиков (Schedulers)

Класс	Назначение	Особенности
`BlockingScheduler`	Автономные скрипты	Блокирует основной поток
`BackgroundScheduler`	Веб-приложения (Flask, Django)	Работает в фоновом потоке
`AsyncIOScheduler`	asyncio-приложения	Интеграция с event loop
`GeventScheduler`	gevent-приложения	Работает через greenlets
`TornadoScheduler`	Tornado-приложения	Интеграция с IOLoop
`TwistedScheduler`	Twisted-приложения	Интеграция с reactor
`QtScheduler`	Qt-приложения	Интеграция с QApplication

Базовый планировщик — методы

from apscheduler.schedulers.base import BaseScheduler

# Состояния планировщика
STATE_STOPPED = 0   # остановлен
STATE_RUNNING = 1   # запущен и обрабатывает задачи
STATE_PAUSED = 2    # запущен, но не обрабатывает

scheduler = BaseScheduler(
    logger="apscheduler",                 # логгер
    timezone="UTC",                       # часовой пояс по умолчанию
    jobstore_retry_interval=5.0,          # сек между повторами при ошибках БД
    job_defaults={                        # значения по умолчанию для задач
        "max_instances": 1,
        "misfire_grace_time": 30
    },
    jobstores={                           # хранилища задач
        "default": MemoryJobStore()
    },
    executors={                           # исполнители
        "default": ThreadPoolExecutor(10)
    }
)

scheduler.configure(prefix="apscheduler.") # переконфигурация (только когда не запущен)

scheduler.start(paused=False)             # запуск (paused=True — не обрабатывать задачи)
scheduler.pause()                         # приостановить обработку задач
scheduler.resume()                        # возобновить обработку задач
scheduler.shutdown(wait=True)             # остановка (wait=True — дождаться завершения задач)

Добавление и управление задачами (APScheduler 3.x — классический API)

# Добавление задачи
job = scheduler.add_job(
    func,                                 # вызываемый объект или строка "module:function"
    trigger,                              # триггер: 'date', 'interval', 'cron' или объект
    args=(1, 2),                          # позиционные аргументы
    kwargs={"x": 10},                     # именованные аргументы
    id="unique_job_id",                   # уникальный ID (генерируется автоматически)
    name="Job description",               # описание
    max_instances=1,                      # макс. параллельных экземпляров
    misfire_grace_time=30,                # допустимое опоздание в секундах
    coalesce=True,                        # объединять пропущенные запуски
    replace_existing=False,               # заменять существующую задачу с тем же ID
    jobstore="default",                   # имя хранилища
    executor="default",                   # имя исполнителя
    trigger_args=None,                    # аргументы для триггера (при передаче строкой)
)

# Управление задачей
job.pause()                               # приостановить
job.resume()                              # возобновить
job.remove()                              # удалить
job.modify(max_instances=2)               # изменить параметры
job.reschedule(CronTrigger(hour=10))      # сменить триггер

# Получение задач
scheduler.get_job(job_id)                 # получить задачу по ID
scheduler.get_jobs()                      # список всех задач
scheduler.pause_job(job_id)               # приостановить по ID
scheduler.resume_job(job_id)              # возобновить по ID
scheduler.remove_job(job_id)              # удалить по ID
scheduler.remove_all_jobs()               # удалить все задачи
scheduler.print_jobs()                    # вывести все задачи в stdout

Триггеры

DateTrigger — однократный запуск

from apscheduler.triggers.date import DateTrigger

DateTrigger(run_date=datetime(2026, 12, 31, 23, 59))

IntervalTrigger — периодический

from apscheduler.triggers.interval import IntervalTrigger

IntervalTrigger(
    seconds=10,                           # интервал в секундах
    minutes=0, hours=0, days=0, weeks=0,  # альтернативные единицы
    start_date=None,                      # дата начала (включительно)
    end_date=None,                        # дата окончания (включительно)
    timezone=None,                        # часовой пояс
    jitter=None                           # случайная задержка ±N сек
)

CronTrigger — гибкое расписание (UNIX cron)

from apscheduler.triggers.cron import CronTrigger

CronTrigger(
    year=None, month=None, day=None,      # 4-digit год, 1-12, 1-31
    week=None, day_of_week=None,          # ISO неделя (1-53), 0-6 или mon,tue...
    hour=None, minute=None, second=None,  # 0-23, 0-59, 0-59
    start_date=None, end_date=None,       # границы
    timezone=None,                        # часовой пояс
    jitter=None                           # случайная задержка ±N сек
)

# Альтернатива — из строки crontab
CronTrigger.from_crontab("30 2 * * *")    # каждую ночь в 2:30

Хранилища задач (Job Stores)

Класс	Назначение	Особенности
`MemoryJobStore`	Тесты / dev	Потеря задач при перезапуске
`SQLAlchemyJobStore`	Production (реляционные БД)	PostgreSQL, MySQL, SQLite
`RedisJobStore`	Production (высокая доступность)	Требует redis-py
`MongoDBJobStore`	Production	Требует pymongo
`ZooKeeperJobStore`	Распределённые системы	Требует kazoo

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

SQLAlchemyJobStore(
    url="postgresql://user:pass@localhost/db",
    tablename="apscheduler_jobs",
    pickle_protocol=pickle.HIGHEST_PROTOCOL
)

Исполнители (Executors)

Класс	Для каких задач	Особенности
`ThreadPoolExecutor`	I/O-bound (сеть, БД, файлы)	Потоки, GIL не проблема
`ProcessPoolExecutor`	CPU-bound (расчёты, обработка)	Процессы, накладные расходы выше
`AsyncIOExecutor`	asyncio-задачи	Интеграция с event loop

from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

ThreadPoolExecutor(max_workers=10)        # до 10 потоков
ProcessPoolExecutor(max_workers=4)        # до 4 процессов

События (Events) — мониторинг

from apscheduler.events import (
    EVENT_JOB_EXECUTED,
    EVENT_JOB_ERROR,
    EVENT_JOB_MISSED,
    EVENT_SCHEDULER_STARTED,
    EVENT_SCHEDULER_SHUTDOWN
)

# Подписка на события
def my_listener(event):
    if event.code == EVENT_JOB_ERROR:
        print(f"Job {event.job_id} failed: {event.exception}")

scheduler.add_listener(my_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)

Типы событий:
- EVENT_SCHEDULER_STARTED / EVENT_SCHEDULER_SHUTDOWN — жизнь планировщика
- EVENT_JOB_SUBMITTED / EVENT_JOB_EXECUTED / EVENT_JOB_ERROR — выполнение задачи
- EVENT_JOB_MISSED — пропущенное выполнение
- EVENT_ALL — все события

Dramatiq. Шпаргалка

Базовые функции

import dramatiq

# Глобальный брокер
broker = dramatiq.get_broker()            # получить текущий брокер
dramatiq.set_broker(broker)               # установить глобальный брокер

# Глобальный кодировщик (сериализация)
encoder = dramatiq.get_encoder()          # получить кодировщик
dramatiq.set_encoder(encoder)             # установить кодировщик

Actor — декоратор и класс

@dramatiq.actor
def my_actor(x, y):
    return x + y

# С параметрами
@dramatiq.actor(
    queue_name="default",                 # очередь (по умолчанию "default")
    priority=0,                           # приоритет (0 — наивысший)
    actor_name="custom_name",             # имя актора (по умолчанию — имя функции)
    broker=None,                          # брокер для этого актора
    max_retries=3,                        # макс. повторов при ошибке
    min_backoff=1000,                     # мин. задержка перед повтором (мс)
    max_backoff=60000,                    # макс. задержка перед повтором (мс)
    store_results=False,                  # сохранять результат? (требуется Results middleware)
    time_limit=None,                      # таймаут выполнения (мс)
    throws=None,                          # исключения, при которых НЕ делать повторов
)
class MyActor(dramatiq.Actor):            # или через наследование
    def perform(self, x, y):
        return x + y

Свойства Actor

actor.logger                              # логгер актора
actor.fn                                  # исходная функция
actor.broker                              # брокер
actor.actor_name                          # имя актора
actor.queue_name                          # очередь
actor.priority                            # приоритет
actor.options                             # произвольные опции

Отправка сообщений

# Простая отправка
my_actor.send(1, 2)                       # отправить задачу в очередь

# С опциями
my_actor.send_with_options(
    args=(1, 2),                          # позиционные аргументы
    kwargs={},                            # именованные аргументы
    delay=60000,                          # задержка в миллисекундах
    priority=1,                           # приоритет (0 — наивысший)
    queue_name="high_priority",           # переопределить очередь
    on_success="success_actor",           # актор при успехе
    on_failure="failure_actor",           # актор при ошибке
    max_retries=5,                        # переопределить повторы
    time_limit=30000,                     # таймаут выполнения (мс)
    throws=(ValueError,),                 # исключения БЕЗ повторов
    broker=None,                          # переопределить брокер
)

# Построение сообщения (для композиции)
msg = my_actor.message(1, 2)              # создать объект Message

Композиция: Pipeline и Group

from dramatiq import pipeline, group

# Pipeline — цепочка: результат первого → аргумент второго
pipeline(
    actor1.message(1, 2),                 # первый актор
    actor2.message(),                     # второй (получает результат первого)
).run()                                   # выполнить

# Group — параллельное выполнение
group(
    actor1.message(1, 2),
    actor1.message(3, 4),
    actor1.message(5, 6),
).run()                                   # выполнить все параллельно

Middleware — системные плагины

from dramatiq.middleware import (
    Retries,                              # повторы при ошибках
    TimeLimit,                            # таймауты
    Pipelines,                            # поддержка pipeline
    ShutdownNotifications,                # уведомления о завершении
    CurrentMessage,                       # доступ к текущему сообщению
    Callbacks,                            # колбэки on_success/on_failure
    Prometheus,                           # метрики для Prometheus
)

# Подключение middleware к брокеру
from dramatiq.brokers.redis import RedisBroker

broker = RedisBroker(
    middleware=[                           # список middleware
        Retries(max_retries=3),
        TimeLimit(time_limit=30000),
        Pipelines(),
        Prometheus(port=9191),             # экспорт метрик на порт 9191
    ]
)
dramatiq.set_broker(broker)

Результаты (Results middleware)

from dramatiq.results import Results
from dramatiq.results.backends import RedisBackend

# Настройка
result_backend = RedisBackend(
    host="localhost",
    port=6379,
    db=0,
    ttl=3600,                              # время жизни результата (сек)
)

broker.add_middleware(Results(backend=result_backend))

# Актор с сохранением результата
@dramatiq.actor(store_results=True)       # включить сохранение
def compute(x, y):
    return x + y

# Получение результата
message = compute.send(1, 2)
result = compute.get_result(               # дождаться и получить результат
    message,
    block=True,                           # блокировать до получения
    timeout=5000,                         # таймаут (мс)
)

Тестирование: StubBroker

from dramatiq.brokers.stub import StubBroker

broker = StubBroker()
dramatiq.set_broker(broker)

# В тестах
my_actor.send(1, 2)
broker.join("default")                    # дождаться обработки всех сообщений в очереди
                                          # fail_fast=True по умолчанию в Dramatiq 2.0

# Eager-брокер для синхронного выполнения (dev/тесты)
from dramatiq_eager_broker import EagerBroker
broker = EagerBroker()                    # выполняет задачи синхронно, без очереди

Брокеры

RedisBroker

from dramatiq.brokers.redis import RedisBroker

RedisBroker(
    url="redis://localhost:6379/0",       # URL подключения
    namespace="dramatiq",                 # пространство имён для ключей
    queue_prefix="dq",                    # префикс для очередей
    max_connections=10,                   # макс. соединений в пуле
    socket_timeout=None,                  # таймаут сокета
    socket_connect_timeout=None,          # таймаут подключения
    heartbeat_interval=60,                # интервал heartbeat (сек)
)

RabbitMQBroker

from dramatiq.brokers.rabbitmq import RabbitmqBroker

RabbitmqBroker(
    url="amqp://guest:guest@localhost:5672/",
    namespace="dramatiq",
    queue_prefix="dq",
    prefetch=10,                          # количество сообщений на воркер
    heartbeat=60,                         # интервал heartbeat (сек)
)

CLI — запуск воркеров

# Базовый запуск
dramatiq my_module

# С параметрами
dramatiq my_module \
    --broker redis \                      # тип брокера (redis / rabbitmq)
    --processes 4 \                       # количество процессов-воркеров
    --threads 8 \                         # потоков на процесс
    --queues default,high_priority \      # какие очереди слушать
    --path /app \                         # добавить в PYTHONPATH
    --pid-file /tmp/dramatiq.pid \        # файл с PID
    --log-file /var/log/dramatiq.log \    # файл лога
    --verbose                             # подробный вывод

Важные нюансы при работе с API

APScheduler

Job не инстанциируется вручную — только через scheduler.add_job()
misfire_grace_time — если None, задача выполняется при любом опоздании; если 0 — пропускается при любом опоздании
coalesce=True — при нескольких пропущенных запусках выполняется только один раз
Ссылки на функции — можно передавать как строку "module:function" для ленивой загрузки
Сериализация — при использовании постоянных хранилищ аргументы должны быть pickle-совместимы

Dramatiq

Аргументы Actor — должны быть JSON-сериализуемы (bool, int, float, str, list, dict)
fail_fast=True в StubBroker.join() — по умолчанию с Dramatiq 2.0, выбрасывает исключение при dead-letter
Prometheus middleware — больше не входит в стандартный набор, требует установки dramatiq[prometheus]
Результаты — требуют явного подключения Results middleware
Асинхронные акторы — поддерживаются через async_to_sync обёртку

APScheduler и Dramatiq

Введение

Обзор и стек технологий

APScheduler: In-Process планировщик

Dramatiq: Распределённая очередь задач

Возможности и API

APScheduler: API и компоненты

Dramatiq: API и компоненты

Нюансы и подводные камни

APScheduler

Dramatiq

Производительность и нагрузка

Сравнительное бенчмарки

Потребление памяти

Оптимизация

Сравнение с аналогами

Общая классификация

APScheduler vs аналоги

Dramatiq vs аналоги

Когда что выбирать

Выбираем APScheduler, если:

Выбираем Dramatiq, если:

Комбинированное использование

Практические примеры использования

APScheduler в действии

APScheduler 4.0: асинхронный подход

Dramatiq в действии

Интеграция с веб-фреймворками

APScheduler + FastAPI (корректная работа с жизненным циклом)

APScheduler + Flask

Dramatiq + Django

Dramatiq + Flask

Проблема Actor в Dramatiq

Мониторинг и наблюдаемость

Dramatiq + Prometheus

Кастомные middleware для мониторинга

APScheduler: логирование

Развёртывание и эксплуатация

APScheduler в production

Dramatiq в production

Миграция с Celery на Dramatiq

Сравнение синтаксиса

Ключевые отличия при миграции

Когда миграция оправдана

Связка APScheduler + Dramatiq: лучшее из двух миров

Архитектура

Пример реализации

Преимущества такого подхода

Итог

APScheduler: идеальный выбор, когда...

Dramatiq: идеальный выбор, когда...

Комбинированное использование

Заключение

Шпаргалка по API

APScheduler. Шпаргалка

Основные структуры данных (APScheduler 4.x)

Task — описание задачи

TaskDefaults — значения по умолчанию

Schedule — расписание задачи

Типы планировщиков (Schedulers)

Базовый планировщик — методы

Добавление и управление задачами (APScheduler 3.x — классический API)

Триггеры

DateTrigger — однократный запуск

IntervalTrigger — периодический

CronTrigger — гибкое расписание (UNIX cron)

Хранилища задач (Job Stores)

Исполнители (Executors)

События (Events) — мониторинг

Dramatiq. Шпаргалка

Базовые функции

Actor — декоратор и класс

Свойства Actor

Отправка сообщений

Композиция: Pipeline и Group

Middleware — системные плагины

Результаты (Results middleware)

Тестирование: StubBroker

Брокеры

RedisBroker

`Task` — описание задачи

`TaskDefaults` — значения по умолчанию

`Schedule` — расписание задачи