Hintergrund-Tasks¶

Bostie schläft, während ein Wecker klingelt

@bot.every führt eine asynchrone Funktion ohne Argumente nach Zeitplan aus, solange der Bot bereit ist.

import datetime as dt
import logging

import bostie
from bostie import activity
from bostie.events import TaskError

log = logging.getLogger("my-bot")
bot = bostie.Bot()
TOKYO = dt.timezone(dt.timedelta(hours=9))


@bot.every(seconds=300)
async def show_guild_count() -> None:
    count = len(bot.cache.guilds())
    await bot.update_presence(activities=[activity.watching(f"{count} servers")])


@bot.every(time=[dt.time(9, tzinfo=TOKYO), dt.time(21, tzinfo=TOKYO)])
async def remind() -> None:
    log.info("twice a day, at 9:00 and 21:00 in Tokyo")


@bot.every(time=dt.time(0, 0))
async def midnight_utc() -> None:
    log.info("every day at midnight UTC")


@bot.listen
async def task_failed(failure: TaskError) -> None:
    log.error("%s failed", failure.task.callback.__name__, exc_info=failure.error)

every(seconds=300) führt die Funktion aus, sobald der Bot bereit ist, und danach alle 300 Sekunden. Der Zeitplan driftet nicht: Jeder Lauf beginnt beim nächsten Vielfachen von seconds ab dem ersten, und ein Lauf, der länger als seconds dauert, überspringt die Takte, die er überlappt hat.

every(time=...) führt sie jeden Tag zu einer Uhrzeit aus, oder zu jeder von mehreren. Ein datetime.time ohne tzinfo gilt in UTC. Mit einem festen Offset, wie oben, bleibt es bei diesem Offset; mit einer zoneinfo.ZoneInfo folgt es den Sommerzeitwechseln der Zone. Übergib seconds oder time, nicht beides.

Einen Task steuern¶

import logging

import bostie
from bostie import Context
from bostie.checks import is_owner

log = logging.getLogger("my-bot")
bot = bostie.Bot()


@bot.every(seconds=60)
async def check_feeds() -> None:
    log.info("checking the feeds, run %d", check_feeds.count)


@check_feeds.before
async def open_feeds() -> None:
    log.info("feed checks started")


@check_feeds.after
async def close_feeds() -> None:
    log.info("feed checks stopped")


@bot.command(checks=[is_owner])
async def feeds(ctx: Context, minutes: int | None = None) -> None:
    """Pause feed checks, or check every so many minutes."""
    if minutes is None:
        check_feeds.stop()
        await ctx.respond("Feed checks paused.", ephemeral=True)
        return
    check_feeds.change_interval(seconds=minutes * 60)
    check_feeds.start()
    await ctx.respond(f"Checking feeds every {minutes} minutes.", ephemeral=True)

@bot.every gibt einen Task zurück, und ein await auf check_feeds() führt die Funktion weiterhin einmal aus. Seine Methoden ändern den Zeitplan, während der Bot läuft:

`Task`	Bewirkt
`stop()`	bricht den Task ab, während eines Laufs oder zwischen Läufen, und verhindert, dass er mit dem Bot startet, bis `start()` kommt; vor dem Start des Bots aufgerufen, wartet der Task auf `start()`
`start()`	führt den Task aus, solange der Bot bereit ist, ab sofort, wenn er es ist; ein laufender Task läuft weiter wie bisher
`restart()`	bricht den Task ab und startet ihn neu, sodass ein Intervall-Task sofort läuft
`change_interval(seconds=...)` oder `change_interval(time=...)`	wechselt zum neuen Zeitplan; das Warten beginnt ab der Änderung neu, oder ab dem Ende eines laufenden Durchgangs
`count`	wie viele Läufe seit dem letzten Start des Tasks beendet wurden
`seconds`, `times`	der Zeitplan, dem er folgt
`@task.before`, `@task.after`	registrieren `async def hook()`, das jedes Mal abgewartet wird, wenn der Task startet und wenn er stoppt, sei es durch `stop()`, `restart()`, einen ausgeschlossenen Router oder `close()`

Eine Exception in einem Hook wird wie eine in einem Lauf als TaskError gesendet. Wirft der before-Hook eine Exception, bleibt der Task bis zu restart() gestoppt.

Wann Tasks laufen¶

Tasks starten, wenn der Bot Started sendet, und close() bricht sie ab, bevor es den HTTP-Client schließt. Ein Task, der an einem später eingebundenen Router registriert ist, startet, sobald dieser eingebunden wird, und einer, dessen Router ausgeschlossen wird, wird abgebrochen. Eine App, die nur von ASGIApp bereitgestellt wird, startet ihre Tasks, wenn die App startet.

Eine Funktion lässt sich nur einmal als Task registrieren. Um denselben Code nach zwei Zeitplänen auszuführen, registriere zwei Funktionen.

Wenn ein Task eine Exception wirft¶

Eine Exception in einem Lauf wird als TaskError-Event gesendet, mit task, dem Task, der sie geworfen hat, und error, und der Task läuft zu seinem nächsten Zeitpunkt erneut. Ohne TaskError-Listener loggt bostie die Exception samt Traceback.