Triggers Extension
Overview
coc.py’s events are an extremely powerful framework, but they are not particularly well suited for periodic bulk-update
style tasks, and employing the use of APScheduler
or similar modules feels excessive for such a simple job. That is
where the triggers extension for coc.py comes into play.
This extension provides you with powerful and easy to use decorators that turn your coroutines into periodically repeating tasks without the need for any additional modifications. It is as simple as putting a trigger decorator on your existing coroutine functions. The triggers extension comes with:
two types of triggers:
IntervalTrigger
andCronTrigger
,customisable error handlers for each trigger and a global
@on_error()
fallback handler,extensive logging that can seamlessly be integrated with your existing logger,
integrated tools to apply your repeating function across an iterable, and
a framework that is easy to extend, allowing you to create your own custom triggers if you need to.
API Reference
IntervalTrigger
The IntervalTrigger
will continuously loop the decorated function, sleeping for a defined number of seconds
between executions. For convenience, this trigger defines .hourly()
and .daily()
class methods to instantiate
triggers with a sleep time of 1 hour and 24 hours, respectively.
- class coc.ext.triggers.IntervalTrigger(*, seconds: int, iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A decorator class to repeat a function every seconds seconds after the previous execution finishes
- seconds
how many seconds to wait between trigger runs
- Type:
int
- iter_args
an optional list of arguments. The decorated function will be called once per list element, and the element will be passed to the decorated function as the first positional argument. If no iter_args are defined, nothing (especially not None) will be injected into the decorated function
- Type:
Optional[
list
]
- on_startup
whether to trigger a run of the decorated function on startup. Defaults to True
- Type:
Optional[
bool
]
- autostart
whether to automatically start the trigger. Auto-starting it may cause required components to not have fully loaded and initialized. If you choose to disable autostart (which is the default), you can use coc.ext.triggers.start_triggers() to manually kick the trigger execution off once you have loaded all required resources
- Type:
Optional[
bool
]
- error_handler
an optional coroutine function that will be called on each error incurred during the trigger execution. The handler will receive three arguments:
- function_name:
str
the name of the failing trigger’s decorated function
- arg: Optional[
Any
] the failing iter_args element or None if no iter_args are defined
- exception:
Exception
the exception that occurred
- Type:
Optional[
coc.ext.triggers.CoroFunction
]
- function_name:
- logger
an optional logger instance implementing the logging.Logger functionality. Debug, warning and error logs about the trigger execution will be sent to this logger
- Type:
Optional[
logging.Logger
]
- loop
an optional event loop that the trigger execution will be appended to. If no loop is provided, the trigger will provision one using asyncio.get_event_loop()
- Type:
Optional[
asyncio.AbstractEventLoop
]
- kwargs
any additional keyword arguments that will be passed to the decorated function every time it is called
Example
@IntervalTrigger(seconds=600, iter_args=['#2PP', '#2PPP']) async def download_current_war(clan_tag: str): # use your coc client to fetch war data, store it to a file or database, ... pass
- classmethod daily(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs with a 24-hour break between executions
- classmethod hourly(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs with a one-hour break between executions
CronTrigger
The CronTrigger
allows you to specify a standard dialect Cron schedule string to dictate the trigger’s
executions. This allows you to specify highly specialised schedules to e.g. fetch clan game points before and after
the clan games, legend league rankings before and after season reset and much more. For convenience, a set of class
methods to instantiate triggers with common patters have been provided:
hourly()
implements the0 * * * *
schedule,daily()
implements0 0 * * *
,weekly()
implements0 0 * * 0
, andmonthly()
implements0 0 1 * *
.
- class coc.ext.triggers.CronTrigger(*, cron_schedule: Union[CronSchedule, str], iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A decorator class to repeat a function based on a Cron schedule
- cron_schedule
the Cron schedule to follow
- Type:
Union[
str
,coc.ext.triggers.CronSchedule
]
- iter_args
an optional list of arguments. The decorated function will be called once per list element, and the element will be passed to the decorated function as the first positional argument. If no iter_args are defined, nothing (especially not None) will be injected into the decorated function
- Type:
Optional[
list
]
- on_startup
whether to trigger a run of the decorated function on startup. Defaults to True
- Type:
Optional[
bool
]
- autostart
whether to automatically start the trigger. Auto-starting it may cause required components to not have fully loaded and initialized. If you choose to disable autostart (which is the default), you can use coc.ext.triggers.start_triggers() to manually kick the trigger execution off once you have loaded all required resources
- Type:
Optional[
bool
]
- error_handler
an optional coroutine function that will be called on each error incurred during the trigger execution. The handler will receive three arguments:
- function_name:
str
the name of the failing trigger’s decorated function
- arg: Optional[
Any
] the failing iter_args element or None if no iter_args are defined
- exception:
Exception
the exception that occurred
- Type:
Optional[
coc.ext.triggers.CoroFunction
]
- function_name:
- logger
an optional logger instance implementing the logging.Logger functionality. Debug, warning and error logs about the trigger execution will be sent to this logger
- Type:
Optional[
logging.Logger
]
- loop
an optional event loop that the trigger execution will be appended to. If no loop is provided, the trigger will provision one using asyncio.get_event_loop()
- Type:
Optional[
asyncio.AbstractEventLoop
]
- kwargs
any additional keyword arguments that will be passed to the decorated function every time it is called
Example
@CronTrigger(cron_schedule='0 0 * * *', iter_args=['#2PP', '#2PPP']) async def download_current_war(clan_tag: str): # use your coc client to fetch war data, store it to a file or database, ... pass
- classmethod daily(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs at the start of every day
- classmethod hourly(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs at the start of every hour
- classmethod monthly(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs at the start of every month
- classmethod weekly(iter_args: Optional[list] = None, on_startup: bool = True, autostart: bool = False, error_handler: Optional[Callable[[], Coroutine[Any, Any, Any]]] = None, logger: Optional[Logger] = None, loop: Optional[AbstractEventLoop] = None, **kwargs)
A shortcut to create a trigger that runs at the start of every week (Sunday at 00:00)
Starting the Triggers
By default, triggers don’t start on their own. This is because you typically want to load other resources before
running a trigger, e.g. log in to the coc dev site, start your Discord bot or boot up a database connection. If a
trigger fired right away, the initial runs would likely fail due to unavailability of these resources. Due to how
Python works, a trigger would run the moment the interpreter reaches the definition, usually well before you intend
to actually start (you can see that illustrated in the examples as well). That is why by default, all triggers are
set to autostart=False
.
The triggers extension provides a coc.ext.triggers.start_triggers()
function to manually kick off trigger
execution from within your code once you’re ready to start processing. If you don’t need any additional resources to
load in first or have otherwise made sure that your triggers won’t fire early, you can set them to autostart=True
and omit the call to coc.ext.triggers.start_triggers()
. If you have a mixture of auto-started and not
auto-started triggers, coc.ext.triggers.start_triggers()
will only start the ones that aren’t already running.
- async coc.ext.triggers.start_triggers()
Manually start all triggers with autostart=False (which is the default value)
Example
# define a trigger @CronTrigger(cron_schedule='0 0 * * *', iter_args=['#2PP', '#2PPP'], autostart=False) async def download_current_war(clan_tag: str): # use your coc client to fetch war data, store it to a file or database, ... pass if __name__ = '__main__': # login to coc.py and/or load other required resources here event_loop = asyncio.get_event_loop() # then start trigger execution event_loop.run_until_complete(start_triggers()) # set the loop to run forever so that it keeps executing the triggers event_loop.run_forever()
Error Handling
The triggers extension offers two ways to deal with error handling:
passing a handler function directly to the trigger’s
error_handler
parameter. This allows you to specify individual error handlers for each repeated task if you need to.designating a global error handler by decorating a function with
coc.ext.triggers.on_error()
. This will be used as a fallback by all triggers that don’t have a dedicated error handler passed to them during initialisation.
- coc.ext.triggers.on_error() Callable[[], Callable[[str, Any, Exception], Coroutine[Any, Any, Any]]]
A decorator function that designates a function as the global fallback error handler for all exceptions during trigger executions.
Notes
This handler declaration should occur before any trigger declarations to avoid a RuntimeWarning about a potentially undeclared error handler, though that warning can safely be ignored.
Any function decorated by this must be a coroutine and accept three parameters:
- function_name:
str
the name of the failing trigger’s decorated function
- arg: Optional[
Any
] the failing iter_args element or None if no iter_args are defined
- exception:
Exception
the exception that occurred
- Return type:
the decorated handler function
Example
@on_error() async def handle_trigger_exception(function_name: str, arg: Any, exception: Exception): # send a Discord message, do some data cleanup, ... pass
- function_name:
An error handler function must be defined with async def
and accept three parameters in the following order:
a function_name string. The name of the failing trigger’s decorated function will be passed to this parameter.
an arg of arbitrary type (defined by what is passed to the trigger’s iter_args parameter). The failing element of the trigger’s iter_args will be passed to this argument, if any are defined. Otherwise, this parameter will receive
None
.an exception. This parameter will be passed the exception that occurred during the execution of the trigger.
Additional arguments can statically be passed to the error handler making use of functools.partial
, if needed.
Logging
Each trigger can be provided with a class implementing the logging.Logger
functionality. If set, the logger will
receive the following events:
info: trigger execution starts and finishes, along with the next scheduled run times.
warning: if a trigger missed it’s next scheduled run time.
error: if an exception occurs during the execution of a trigger. If both a logger and an error handler are set up, both will receive information about this event.
Other Parameters
Similar to the events API, triggers allow you to specify a list of elements you want the decorated function to be
spread over. If you specify the iter_args
parameter when initialising a trigger, it will call the decorated
function once for each element of that parameter. Each element will be positionally passed into the function’s first
argument. If you prefer to keep your logic inside the function or load it from somewhere else, simply don’t pass the
iter_args
parameter. That will let the trigger know not to inject any positional args.
The boolean on_startup
flag allows you to control the trigger behaviour on startup. If it is set to True
, the
trigger will fire immediately and resume its predefined schedule afterwards. If on_startup
is False
, the
trigger will remain dormant until its first scheduled run time.
The autostart
option allows you to decide whether a trigger should automatically start on application startup.
If autostart is disabled, triggers can be started using coc.ext.triggers.start_triggers()
once all
dependencies and required resources are loaded. Refer to Starting the Triggers for details.
The loop
parameter allows you to pass your own asyncio event loop to attach the trigger execution to. If omitted,
the current event loop will be used.
You can also specify additional key word arguments (**kwargs
). Any extra arguments will be passed to the decorated
function as key word arguments on each call.
Example
Below are some usage examples for the triggers extension:
1import asyncio
2import logging
3import os
4
5import coc
6from coc.ext.triggers import CronSchedule, CronTrigger, IntervalTrigger, on_error, start_triggers
7
8
9coc_client = coc.Client()
10cron = CronSchedule('0 0 * * *')
11event_loop = asyncio.get_event_loop()
12logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s',
13 datefmt='%Y-%m-%d %H:%M:%S')
14_logger = logging.getLogger()
15MOCK_DATABASE = {'cg_contribution': {}, 'donations': {}}
16
17
18@on_error()
19async def default_error_handler(function_name, arg, exception):
20 print('Default error handler engaged')
21 print('Default handler:', function_name, arg, exception)
22
23
24async def special_error_handler(function_name, arg, exception):
25 print('Special error handler engaged')
26 print('Special handler:', function_name, arg, exception)
27
28
29@CronTrigger(cron_schedule='0 0 * * 5', iter_args=['#2PP', '#2PPP'], on_startup=False, loop=event_loop, logger=_logger)
30async def cache_cg_contribution_before_raid_weekend(player_tag: str):
31 """This trigger showcases the `on_startup=False` option and the use of `iter_args`"""
32
33 player = await coc_client.get_player(player_tag)
34 MOCK_DATABASE['cg_contribution'][player.tag] = player.clan_capital_contributions
35
36 print('capital gold contributions', MOCK_DATABASE['cg_contribution'])
37
38
39@CronTrigger(cron_schedule=cron, iter_args=['#2PP'], loop=event_loop, logger=_logger)
40async def daily_donation_downloader(clan_tag: str):
41 """This trigger showcases the use of a :class:`CronSchedule` object and `iter_args`"""
42
43 clan = await coc_client.get_clan(clan_tag)
44 async for member in clan.get_detailed_members():
45 MOCK_DATABASE['donations'][member.tag] = member.donations
46
47 print('donation status', MOCK_DATABASE['donations'])
48
49
50@IntervalTrigger.hourly(loop=event_loop, logger=_logger, error_handler=special_error_handler)
51async def test_special_error_handling():
52 """This trigger showcases the convenience class methods and the use of a dedicated error handler"""
53
54 return 1/0
55
56
57@IntervalTrigger(seconds=10, iter_args=[1, 0], autostart=True, loop=event_loop, logger=_logger)
58async def test_default_error_handling(divisor: int):
59 """This trigger demonstrates the use of `autostart=True` (this is fine because it has no dependencies
60 on other resources) in combination with `iter_args` and the default error handler defined by @on_error.
61 It also is the trigger with the lowest repeat timer to showcase the fact that triggers indeed do repeat
62 """
63
64 return 2/divisor
65
66
67async def main():
68 try:
69 await coc_client.login(os.environ.get('DEV_SITE_EMAIL'), os.environ.get('DEV_SITE_PASSWORD'))
70 except coc.InvalidCredentials as error:
71 exit(error)
72
73
74if __name__ == "__main__":
75 try:
76 # using the loop context, run the main function
77 event_loop.run_until_complete(main())
78
79 # wait a few seconds to simulate other resources loading
80 event_loop.run_until_complete(asyncio.sleep(3))
81
82 # note how this print statement will show up in your console AFTER the auto-started trigger fired
83 print('NOTE: Auto-started trigger "test_default_error_handling" has already fired at this point')
84
85 # then start trigger execution
86 event_loop.run_until_complete(start_triggers())
87
88 # set the loop to run forever so that it keeps executing the triggers
89 # NOTE: this line is blocking, no code after will be executed
90 event_loop.run_forever()
91 print('This line will never make it to your console')
92 except KeyboardInterrupt:
93 pass
Extending this Extension
If you find yourself in need of scheduling logic that none of the included triggers can provide, you can easily
create a trigger class that fits your needs by importing the BaseTrigger
from this extension, creating a
subclass and overwriting the next_run
property. The property needs to return a timezone-aware
datetime.datetime
object indicating when the trigger should run next based on the current system time.
from coc.ext.triggers import BaseTrigger
from datetime import datetime, timedelta
from random import randint
class RandomTrigger(BaseTrigger):
def __init__(self,
*, # disable positional arguments
seconds: int,
iter_args: Optional[list] = None,
on_startup: bool = True,
autostart: bool = False,
error_handler: Optional[CoroFunction] = None,
logger: Optional[logging.Logger] = None,
loop: Optional[asyncio.AbstractEventLoop] = None,
**kwargs):
super().__init__(iter_args=iter_args, on_startup=on_startup, autostart=autostart,
error_handler=error_handler, logger=logger, loop=loop, **kwargs)
@property
def next_run(self) -> datetime:
"""randomly triggers every 50-100 seconds"""
return datetime.now().astimezone() + timedelta(seconds=randint(50, 101))