How to Create and Use Scheduled Jobs with the scheduled_job Decorator in HoshinoBot

Use the scheduled_job method on a Service instance as a decorator to register async functions with HoshinoBot's APScheduler, which automatically handles timezone configuration, error logging, and graceful execution.

HoshinoBot, an open-source QQ bot framework based on NoneBot, provides a robust scheduling system through its Service class. The scheduled_job decorator allows developers to register periodic background tasks—such as data updates, RSS polling, or message broadcasting—without managing the underlying APScheduler configuration manually.

Understanding the scheduled_job Decorator Architecture

The scheduled_job decorator is implemented as an instance method of the Service class in hoshino/service.py (lines 41-58). When you create a service instance, this method becomes available to wrap asynchronous functions and register them with NoneBot's APScheduler.

The decorator performs three critical functions:

• Default Configuration: Automatically sets timezone='Asia/Shanghai', misfire_grace_time=60, and coalesce=True to ensure reliable execution across timezones and bot restarts.
• Error Isolation: Wraps the coroutine in a try-except block that logs exceptions without crashing the scheduler.
• Execution Logging: Records job start and completion times for debugging purposes.

Basic Syntax and Trigger Types

To create a scheduled job, instantiate a Service and apply the scheduled_job decorator with APScheduler trigger arguments.

from hoshino import Service, priv

sv = Service("my-service", use_priv=priv.NORMAL)

@sv.scheduled_job('cron', hour='*/2', minute='0')
async def periodic_task():
    sv.logger.info("Executing scheduled task")
    # Your async logic here

The decorator accepts any APScheduler trigger type:

• cron: Calendar-based scheduling (e.g., hour='5', minute='0' for daily at 5:00 AM)
• interval: Time-delta based (e.g., minutes=3 for every 3 minutes)
• date: One-time execution at a specific datetime

Practical Implementation Patterns

Daily Data Updates with Jitter

For tasks that should run daily but avoid thundering herd problems, combine cron scheduling with the jitter parameter. The pcr_data_updater.py module demonstrates this pattern:

@sv.scheduled_job('cron', hour='5', jitter=300)  # 5:00 AM ±5 minutes

async def update_pcr_data():
    # Update Princess Connect Re:Dive character data

    await refresh_character_database()

High-Frequency Polling with Broadcast

When monitoring external feeds, use short intervals and broadcast results to enabled groups. The mikan.py module polls an RSS feed every 3 minutes:

@sv.scheduled_job('cron', minute='*/3', second='15')
async def mikan_poller():
    new_items = await check_rss_feed()
    if new_items:
        await sv.broadcast(
            f"New anime available: {new_items[0].title}",
            TAG='mikan_update',
            interval_time=0.3
        )

Error Resilience and Logging

The decorator automatically handles exceptions. Even if your coroutine fails, the scheduler continues running:

@sv.scheduled_job('interval', minutes=5)
async def fragile_monitor():
    # This exception is caught and logged, not raised

    raise ConnectionError("API unreachable")

The error appears in logs as:

ERROR: Scheduled job fragile_monitor occured when doing scheduled job fragile_monitor.
Traceback (most recent call last):
  ...
ConnectionError: API unreachable

Real-World Examples in the Repository

Several production modules in ice9coffee/hoshinobot rely on scheduled_job:

• hoshino/modules/priconne/pcr_data_updater.py (lines 45-47): Daily character data synchronization at 05:00 with 300-second jitter.
• hoshino/modules/mikan/mikan.py (lines 61-63): RSS polling every 3 minutes at second 15.
• hoshino/modules/priconne/comic.py (lines 99-101): Comic update checks every 5 minutes at second 25.

These implementations demonstrate the decorator's flexibility across different trigger types and use cases.

Summary

• The scheduled_job decorator is a method of the Service class that registers async functions with APScheduler.
• It automatically configures timezone (Asia/Shanghai), misfire grace time (60s), and coalescing to ensure reliable execution.
• The decorator supports all APScheduler triggers: cron, interval, and date.
• Exceptions are caught and logged without stopping the scheduler, ensuring high availability.
• Real-world usage in pcr_data_updater.py, mikan.py, and comic.py demonstrates patterns for daily updates, high-frequency polling, and error resilience.

Frequently Asked Questions

Can I use the scheduled_job decorator outside of a Service class?

No. The scheduled_job method is an instance method of the Service class defined in hoshino/service.py. You must create a service instance (e.g., sv = Service("name")) and use @sv.scheduled_job() to register jobs. This ensures the job inherits the service's configuration and logging context.

What happens if my scheduled job raises an exception?

The decorator wraps your coroutine in a try-except block that catches all exceptions, logs the full traceback using the service logger, and allows the scheduler to continue running. As implemented in hoshino/service.py, the error message format is: Scheduled job {func_name} occured when doing scheduled job {func_name}. This prevents one faulty job from crashing the entire bot.

How do I schedule a job to run at a specific time every day?

Use the cron trigger with hour and minute arguments. For example, @sv.scheduled_job('cron', hour='5', minute='0') runs daily at 5:00 AM. You can also add jitter=300 to introduce a random delay of up to 300 seconds, which is useful for avoiding simultaneous requests when multiple instances run. This pattern is used in hoshino/modules/priconne/pcr_data_updater.py.

Can I pass arguments to the scheduled function?

The decorated function receives no arguments from the scheduler by design. However, you can use closures (variables from the outer scope) or define the function within a class method to access self. For service-wide data, access the service instance (sv) or use global variables within the module. The mikan.py module demonstrates accessing external state and broadcasting results without receiving scheduler arguments.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}