Routines Ext

Routines are helpers designed to make running async background tasks in TwitchIO easier. Overall Routines are a QoL and are designed to be simple and easy to use.

Recipes

A simple routine:

This routine will run every 5 seconds for 5 iterations.

from twitchio.ext import routines


@routines.routine(seconds=5.0, iterations=5)
async def hello(arg: str):
    print(f'Hello {arg}!')


hello.start('World')

Routine with a before/after_routine hook:

This routine will run a hook before starting, this can be useful for setting up state before the routine runs. The before_routine hook will only be called once. Similarly after_routine will be called once at the end of the routine.

from twitchio.ext import routines


@routines.routine(hours=1)
async def hello():
    print('Hello World!')

@hello.before_routine
async def hello_before():
    print('I am run first!')


hello.start()

Routine with an error handler:

This example shows a routine with a non-default error handler; by default all routines will stop on error. You can change this behaviour by adding stop_on_error=False to your routine start function.

from twitchio.ext import routines


@routines.routine(minutes=10)
async def hello(arg: str):
    raise RuntimeError

@hello.error
async def hello_on_error(error: Exception):
    print(f'Hello routine raised: {error}.')


hello.start('World', stop_on_error=True)

Routine which runs at a specific time:

This routine will run at the same time everyday. If a naive datetime is provided, your system local time is used.

The below example shows a routine which will first be ran on the 1st, June 2021 at 9:30am system local time. It will then be ran every 24 hours after the initial date, until stopped.

If the date has already passed, the routine will run at the next specified time. For example: If today was the 2nd, June 2021 8:30am and your datetime was scheduled to run on the 1st, June 2021 at 9:30am, your routine will first run on 2nd, June 2021 at 9:30am.

In simpler terms, datetimes in the past only care about the time, not the date. This can be useful when scheduling routines that don’t need to be started on a specific date.

import datetime

from twitchio.ext import routines


@routines.routine(time=datetime.datetime(year=2021, month=6, day=1, hour=9, minute=30))
async def hello(arg: str):
    print(f'Hello {arg}!')


hello.start('World')

API Reference

Attributes

• completed_iterations
• remaining_iterations
• start_time

Methods

• @after_routine
• @before_routine
• defcancel
• defchange_interval
• @error
• asyncon_error
• defrestart
• defstart
• defstop

class twitchio.ext.routines.Routine(*, coro: Callable, iterations: Optional[int] = None, time: Optional[datetime] = None, delta: Optional[float] = None, wait_first: Optional[bool] = False)

The main routine class which helps run async background tasks on a schedule.

Examples

@routine(seconds=5, iterations=3)
async def test(arg):
    print(f'Hello {arg}')

test.start('World!')

Warning

This class should not be instantiated manually. Use the decorator routine() instead.

after_routine(coro: Callable) → None

A decorator to assign a coroutine to run after the routine has ended.

before_routine(coro: Callable) → None

A decorator to assign a coroutine to run before the routine starts.

cancel() → None

Cancel the routine effective immediately and non-gracefully.

Note

Consider using stop() if a graceful stop, which will complete the current iteration, is desired.

change_interval(*, seconds: Optional[float] = 0, minutes: Optional[float] = 0, hours: Optional[float] = 0, time: Optional[datetime] = None, wait_first: Optional[bool] = False) → None

Method which schedules the running interval of the task to change.

Parameters

• seconds (Optional[float]) – The seconds to wait before the next iteration of the routine.

• minutes (Optional[float]) – The minutes to wait before the next iteration of the routine.

• hours (Optional[float]) – The hours to wait before the next iteration of the routine.

• time (Optional[datetime.datetime]) – A specific time to run this routine at. If a naive datetime is passed, your system local time will be used.

• wait_first (Optional[bool]) – Whether to wait the specified time before running the first iteration at the new interval. Defaults to False.

Raises

RuntimeError – Raised when the time argument and any hours, minutes or seconds is passed together.

Warning

The time argument can not be passed in conjunction with hours, minutes or seconds. This behaviour is intended as it allows the time to be exact every day.

error(coro: Callable)

A decorator to assign a coroutine as the error handler for this routine.

The error handler takes in one argument: the exception caught.

coroutine on_error(error: Exception)

The default error handler for this routine. Can be overwritten with error().

restart(*args, **kwargs) → None

Restart the currently running routine.

Parameters

• stop_on_error (Optional[bool]) – Whether or not to stop and cancel the routine on error. Defaults to True.

• force (Optional[bool]) – If True the restart will cancel the currently running routine effective immediately and restart. If False a graceful stop will occur, which allows the routine to finish it’s current iteration. Defaults to True.

• *args – The args to pass to the routine.

• **kwargs – The kwargs to pass to the routine.

Note

This does not return the internal task unlike start().

start(*args, **kwargs) → Task

Start the routine and return the created task.

Parameters

• stop_on_error (Optional[bool]) – Whether or not to stop and cancel the routine on error. Defaults to True.

• *args – The args to pass to the routine.

• **kwargs – The kwargs to pass to the routine.

Returns

asyncio.Task – The created internal asyncio task.

Raises

RuntimeError – Raised when this routine is already running when start is called.

stop() → None

Stop the routine gracefully.

Note

This allows the current iteration to complete before the routine is cancelled. If immediate cancellation is desired consider using cancel() instead.

property completed_iterations: int

A count of completed iterations.

property remaining_iterations: Optional[int]

A count of remaining iterations.

property start_time: Optional[datetime]

The time the routine was started.

Note

This does not reset when restarting, stopping or cancelling the routine.

twitchio.ext.routines.routine(*, seconds: Optional[float] = 0, minutes: Optional[float] = 0, hours: Optional[float] = 0, time: Optional[datetime] = None, iterations: Optional[int] = None, wait_first: Optional[bool] = False)

A decorator to assign a coroutine as a Routine.

Parameters

• seconds (Optional[float]) – The seconds to wait before the next iteration of the routine.

• minutes (Optional[float]) – The minutes to wait before the next iteration of the routine.

• hours (Optional[float]) – The hours to wait before the next iteration of the routine.

• time (Optional[datetime.datetime]) – A specific time to run this routine at. If a naive datetime is passed, your system local time will be used.

• iterations (Optional[int]) – The amount of iterations to run this routine before stopping. If set to None or 0, the routine will run indefinitely.

• wait_first (Optional[bool]) – Whether to wait the specified time before running the first iteration. This has no effect when the time argument is used. Defaults to False.

Raises

• RuntimeError – Raised when the time argument and any hours, minutes or seconds is passed together.

• TypeError – Raised when used on a non-coroutine.

Warning

The time argument can not be passed in conjunction with hours, minutes or seconds. This behaviour is intended as it allows the time to be exact every day.