Routines

Attributes
Methods
class twitchio.ext.routines.Routine(*, name: str | None = None, coro: Callable[[...], Coroutine[Any, Any, Any]], max_attempts: int | None = None, iterations: int | None, wait_remainder: bool = False, stop_on_error: bool = False, wait_first: bool, time: datetime.datetime | None, delta: datetime.timedelta | None)

The TwitchIO Routine class which runs asynchronously in the background on a timed loop.

Note

You should not instantiate this class manually, instead use the routine() decorator instead.

Examples

# This routine will run every minute until stopped or canceled.

@routines.routine(delta=datetime.timedelta(minutes=1))
async def my_routine() -> None:
    print("Hello World!")

my_routine.start()

# Pass some arguments to a routine...

@routines.routine(delta=datetime.timedelta(minutes=1))
async def my_routine(hello: str) -> None:
    print(f"Hello {hello}")

my_routine.start("World!")

# Only run the routine three of times...

@routines.routine(delta=datetime.timedelta(minutes=1), iterations=3)
async def my_routine(hello: str) -> None:
    print(f"Hello {hello}")

my_routine.start("World!")
property args: tuple[Any, ...]

Property returning any positional arguments passed to the routine via start().

property kwargs: Any

Property returning any keyword arguments passed to the routine via start().

start(*args: Any, **kwargs: Any) → Task[None]

Method to start the Routine in the background.

Note

You can not start an already running task. See: restart() instead.

Parameters

  • *args (Any) – Any positional args passed to this method will also be passed to your Routine callback.

  • **kwargs (Any) – Any keyword arguments passed to this method will also be passed to your Routine callback.

Returns

The internal background task associated with this Routine.

Return type

asyncio.Task

stop() → None

Method to stop the currently running task after it completes its current iteration.

Unlike cancel() this will schedule the task to stop after it completes its next iteration.

cancel() → None

Method to immediately cancel the currently running Routine.

Unlike stop(), this method is not graceful and will cancel the task in its current iteration.

restart(*, force: bool = True) → None

Method which restarts the Routine.

If the Routine has not yet been started, this method immediately returns.

Parameters

force (bool) – Whether to cancel the currently running routine and restart it immediately. If this is False the Routine will start after it’s current iteration. Defaults to True.

before_routine(func: Callable[[...], Coroutine[Any, Any, Any]]) → None

This function is a decorator.

Decorator used to set a coroutine to run before the Routine has started.

Any arguments passed to start() will also be passed to this coroutine callback.

after_routine(func: Callable[[...], Coroutine[Any, Any, Any]]) → None

This function is a decorator.

Decorator used to set a coroutine to run after the Routine has been stopped, canceled or completed.

Any arguments passed to start() will also be passed to this coroutine callback.

async on_error(error: Exception) → None

This function is a coroutine.

Default error handler for this Routine. You can override this with the error() decorator.

By default this handler will log the exception.

Parameters

error (Exception) – The exception raised in the routine.

error(func: Callable[[...], Coroutine[Any, Any, Any]]) → None

This function is a decorator.

Decorator used to set a coroutine as an error handler for this Routine.

property completed_iterations: int

Property returning a int of the successfully completed iterations.

Note

Iterations where an error occured do not count towards completed iterations.

property remaining_iterations: int | None

Property returning a int or None of the remaining iterations.

If iterations was not set this returns None.

property current_iteration: int

Property returning the current iteration count as a int.

property last_iteration: datetime.datetime | None

Property which returns a datetime.datetime of when the current iteration started.

next_iteration() → float

Method which returns the time until the next iteration as a float of seconds.

change_interval(*, delta: datetime.timedelta | None = None, time: datetime.datetime | None = None, wait_first: bool = False) → None

Method to change the running interval of a currently running routine.

Parameters

  • delta (datetime.timedelta | None) – A datetime.timedelta of time to wait per iteration of the Routine. If this is None, you must pass the time parameter. Defaults to None.

  • time (datetime.datetime | None) – A datetime.datetime to schedule an run each iteration of the Routine. The Routine will run at the same time everyday. If this is None, you must pass the delta parameter. Defaults to None.

  • wait_first (bool) – An optional bool indicating whether the currently running routine should complete it’s current iteration before restarting. Defaults to False which will immediately cancel the currently running iteration and restart the routine with the new times provided.

twitchio.ext.routines.routine(*, delta: datetime.timedelta | None = None, time: datetime.datetime | None = None, name: str | None = None, iterations: int | None = None, wait_first: bool = False, wait_remainder: bool = False, max_attempts: int | None = 5, stop_on_error: bool = False) → Callable[[Callable[[...], Coroutine[Any, Any, Any]]], Routine]

This function is a decorator.

A decorator to assign a coroutine as a Routine.

Important

One parameter of either time or delta must be passed. Both parameters CAN NOT be used together.

Parameters

  • delta (datetime.timedelta | None) – A datetime.timedelta of time to wait per iteration of the Routine. If this is None, you must pass the time parameter. Defaults to None.

  • time (datetime.datetime | None) – A datetime.datetime to schedule an run each iteration of the Routine. The Routine will run at the same time everyday. If this is None, you must pass the delta parameter. Defaults to None.

  • name (str | None) – An optional name to use for this routine. Defaults to None which uses the repr of the Routine.

  • iterations (int | None) – An optional int amount of iterations to run this routine for. For example if set to 3, the Routine will only run 3 times and complete. If set to None the routine will continue running until stopped or canceled. Defaults to None.

  • wait_first (bool) – Whether the Routine should wait the amount of time specified in time or delta before starting. Defaults to False, which will run the first iteration immediately after start() is called.

  • wait_remainder (bool) – Whether to wait only the remaining time to the next iteration after the current iteration completes. For example if your Routine is scheduled to run every minute, and your iteration took 30 seconds, the routine will only wait an additional 30 seconds before running its next iteration. Defaults to False which means that the Routine will wait the full time after each iteration. Note: this has no effect when the time parameter is passed.

  • max_attempts (int | None) – The maximum times the Routine should be allowed to error consecuitevely before stopping. When an error occurs and this parameter is set, the routine will sleep for a small backoff interval before retrying the iteration. If the max_attempts counter is hit after multiple consecutive errors, the routine will stop. This counter resets when a successful iteration occurs. Note: max_attempts has no effect when stop_on_error is set. If set to None, the routine will backoff indefinitely. Defaults to 5.

  • stop_on_error (bool) – A bool indicating whether the Routine should immediately stop when encountering an error. This parameter takes precedence over max_attempts. Defaults to False.

Raises

  • TypeError – The function decorated is not a coroutine function.

  • RuntimeError – The time and delta arguments were both passed together. Only one of these parameters can be used.

  • RuntimeError – Both the time and delta arguments are missing. One of these parameters must be used.