Chat - Introduction to Middleware#

In this tutorial, we will go over a few examples on how to use and write your own chat command middleware.

Basics#

Command Middleware can be understood as a set of filters which decide if a chat command should be executed by a user. A basic example would be the idea to limit the use of certain commands to just a few chat rooms or restricting the use of administrative commands to just the streamer.

There are two types of command middleware:

  1. global command middleware: this will be used to check any command that might be run

  2. single command middleware: this will only be used to check a single command if it might be run

Example setup#

The following basic chat example will be used in this entire tutorial

 1 import asyncio
 2 from twitchAPI import Twitch
 3 from twitchAPI.chat import Chat, ChatCommand
 4 from twitchAPI.oauth import UserAuthenticationStorageHelper
 5 from twitchAPI.types import AuthScope
 6
 7
 8 APP_ID = 'your_app_id'
 9 APP_SECRET = 'your_app_secret'
10 SCOPES = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
11 TARGET_CHANNEL = ['your_first_channel', 'your_second_channel']
12
13
14 async def command_one(cmd: ChatCommand):
15     await cmd.reply('This is the first command!')
16
17
18 async def command_two(cmd: ChatCommand):
19     await cmd.reply('This is the second command!')
20
21
22 async def run():
23     twitch = await Twitch(APP_ID, APP_SECRET)
24     helper = UserAuthenticationStorageHelper(twitch, SCOPES)
25     await helper.bind()
26     chat = await Chat(twitch, initial_channel=TARGET_CHANNEL)
27
28     chat.register_command('one', command_one)
29     chat.register_command('two', command_two)
30
31     chat.start()
32     try:
33         input('press Enter to shut down...\n')
34     except KeyboardInterrupt:
35         pass
36     finally:
37         chat.stop()
38         await twitch.close()
39
40
41 asyncio.run(run())

Global Middleware#

Given the above example, we now want to restrict the use of all commands in a way that only user user1 can use them and that they can only be used in your_first_channel.

The highlighted lines in the code below show how easy it is to set this up:

 1import asyncio
 2from twitchAPI import Twitch
 3from twitchAPI.chat import Chat, ChatCommand
 4from twitchAPI.chat.middleware import UserRestriction, ChannelRestriction
 5from twitchAPI.oauth import UserAuthenticationStorageHelper
 6from twitchAPI.types import AuthScope
 7
 8
 9APP_ID = 'your_app_id'
10APP_SECRET = 'your_app_secret'
11SCOPES = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
12TARGET_CHANNEL = ['your_first_channel', 'your_second_channel']
13
14
15async def command_one(cmd: ChatCommand):
16    await cmd.reply('This is the first command!')
17
18
19async def command_two(cmd: ChatCommand):
20    await cmd.reply('This is the second command!')
21
22
23async def run():
24    twitch = await Twitch(APP_ID, APP_SECRET)
25    helper = UserAuthenticationStorageHelper(twitch, SCOPES)
26    await helper.bind()
27    chat = await Chat(twitch, initial_channel=TARGET_CHANNEL)
28    chat.register_command_middleware(UserRestriction(allowed_users=['user1']))
29    chat.register_command_middleware(ChannelRestriction(allowed_channel=['your_first_channel']))
30
31    chat.register_command('one', command_one)
32    chat.register_command('two', command_two)
33
34    chat.start()
35    try:
36        input('press Enter to shut down...\n')
37    except KeyboardInterrupt:
38        pass
39    finally:
40        chat.stop()
41        await twitch.close()
42
43
44asyncio.run(run())

Single Command Middleware#

Given the above example, we now want to only restrict !one to be used by the streamer of the channel its executed in.

The highlighted lines in the code below show how easy it is to set this up:

 1import asyncio
 2from twitchAPI import Twitch
 3from twitchAPI.chat import Chat, ChatCommand
 4from twitchAPI.chat.middleware import StreamerOnly
 5from twitchAPI.oauth import UserAuthenticationStorageHelper
 6from twitchAPI.types import AuthScope
 7
 8
 9APP_ID = 'your_app_id'
10APP_SECRET = 'your_app_secret'
11SCOPES = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
12TARGET_CHANNEL = ['your_first_channel', 'your_second_channel']
13
14
15async def command_one(cmd: ChatCommand):
16    await cmd.reply('This is the first command!')
17
18
19async def command_two(cmd: ChatCommand):
20    await cmd.reply('This is the second command!')
21
22
23async def run():
24    twitch = await Twitch(APP_ID, APP_SECRET)
25    helper = UserAuthenticationStorageHelper(twitch, SCOPES)
26    await helper.bind()
27    chat = await Chat(twitch, initial_channel=TARGET_CHANNEL)
28
29    chat.register_command('one', command_one, command_middleware=[StreamerOnly()])
30    chat.register_command('two', command_two)
31
32    chat.start()
33    try:
34        input('press Enter to shut down...\n')
35    except KeyboardInterrupt:
36        pass
37    finally:
38        chat.stop()
39        await twitch.close()
40
41
42asyncio.run(run())

Using Execute Blocked Handlers#

Execute blocked handlers are a function which will be called whenever the execution of a command was blocked.

You can define a default handler to be used for any middleware that blocks a command execution and/or set one per middleware that will only be used when that specific middleware blocked the execution of a command.

Note: You can mix and match a default handler with middleware specific handlers as much as you want.

Using a default handler#

A default handler will be called whenever the execution of a command is blocked by a middleware which has no specific handler set.

You can define a simple handler which just replies to the user as follows using the global middleware example:

handle_command_blocked() will be called if the execution of either !one or !two is blocked, regardless by which of the two middlewares.

 1import asyncio
 2from twitchAPI import Twitch
 3from twitchAPI.chat import Chat, ChatCommand
 4from twitchAPI.chat.middleware import UserRestriction, ChannelRestriction
 5from twitchAPI.oauth import UserAuthenticationStorageHelper
 6from twitchAPI.types import AuthScope
 7
 8
 9APP_ID = 'your_app_id'
10APP_SECRET = 'your_app_secret'
11SCOPES = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
12TARGET_CHANNEL = ['your_first_channel', 'your_second_channel']
13
14
15async def command_one(cmd: ChatCommand):
16    await cmd.reply('This is the first command!')
17
18
19async def command_two(cmd: ChatCommand):
20    await cmd.reply('This is the second command!')
21
22
23async def handle_command_blocked(cmd: ChatCommand):
24    await cmd.reply(f'You are not allowed to use {cmd.name}!')
25
26
27async def run():
28    twitch = await Twitch(APP_ID, APP_SECRET)
29    helper = UserAuthenticationStorageHelper(twitch, SCOPES)
30    await helper.bind()
31    chat = await Chat(twitch, initial_channel=TARGET_CHANNEL)
32    chat.register_command_middleware(UserRestriction(allowed_users=['user1']))
33    chat.register_command_middleware(ChannelRestriction(allowed_channel=['your_first_channel']))
34
35    chat.register_command('one', command_one)
36    chat.register_command('two', command_two)
37    chat.default_command_execution_blocked_handler = handle_command_blocked
38
39    chat.start()
40    try:
41        input('press Enter to shut down...\n')
42    except KeyboardInterrupt:
43        pass
44    finally:
45        chat.stop()
46        await twitch.close()
47
48
49asyncio.run(run())

Using a middleware specific handler#

A middleware specific handler can be used to change the response based on which middleware blocked the execution of a command. Note that this can again be both set for command specific middleware as well as global middleware. For this example we will only look at global middleware but the method is exactly the same for command specific one.

To set a middleware specific handler, you have to set execute_blocked_handler. For the preimplemented middleware in this library, you can always pass this in the init of the middleware.

In the following example we will be responding different based on which middleware blocked the command.

 1import asyncio
 2from twitchAPI import Twitch
 3from twitchAPI.chat import Chat, ChatCommand
 4from twitchAPI.chat.middleware import UserRestriction, ChannelRestriction
 5from twitchAPI.oauth import UserAuthenticationStorageHelper
 6from twitchAPI.types import AuthScope
 7
 8
 9APP_ID = 'your_app_id'
10APP_SECRET = 'your_app_secret'
11SCOPES = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
12TARGET_CHANNEL = ['your_first_channel', 'your_second_channel']
13
14
15async def command_one(cmd: ChatCommand):
16    await cmd.reply('This is the first command!')
17
18
19async def command_two(cmd: ChatCommand):
20    await cmd.reply('This is the second command!')
21
22
23async def handle_blocked_user(cmd: ChatCommand):
24    await cmd.reply(f'Only user1 is allowed to use {cmd.name}!')
25
26
27async def handle_blocked_channel(cmd: ChatCommand):
28    await cmd.reply(f'{cmd.name} can only be used in channel your_first_channel!')
29
30
31async def run():
32    twitch = await Twitch(APP_ID, APP_SECRET)
33    helper = UserAuthenticationStorageHelper(twitch, SCOPES)
34    await helper.bind()
35    chat = await Chat(twitch, initial_channel=TARGET_CHANNEL)
36    chat.register_command_middleware(UserRestriction(allowed_users=['user1'],
37                                                     execute_blocked_handler=handle_blocked_user))
38    chat.register_command_middleware(ChannelRestriction(allowed_channel=['your_first_channel'],
39                                                        execute_blocked_handler=handle_blocked_channel))
40
41    chat.register_command('one', command_one)
42    chat.register_command('two', command_two)
43
44    chat.start()
45    try:
46        input('press Enter to shut down...\n')
47    except KeyboardInterrupt:
48        pass
49    finally:
50        chat.stop()
51        await twitch.close()
52
53
54asyncio.run(run())

Write your own Middleware#

You can also write your own middleware to implement custom logic, you only have to extend the class BaseCommandMiddleware.

In the following example, we will create a middleware which allows the command to execute in 50% of the times its executed.

from typing import Callable, Optional, Awaitable

class MyOwnCoinFlipMiddleware(BaseCommandMiddleware):

   # it is best practice to add this part of the init function to be compatible with the default middlewares
   # but you can also leave this out should you know you dont need it
   def __init__(self, execute_blocked_handler: Optional[Callable[[ChatCommand], Awaitable[None]]] = None):
     self.execute_blocked_handler = execute_blocked_handler

   async def can_execute(cmd: ChatCommand) -> bool:
      # add your own logic here, return True if the command should execute and False otherwise
      return random.choice([True, False])

   async def was_executed(cmd: ChatCommand):
      # this will be called whenever a command this Middleware is attached to was executed, use this to update your internal state
      # since this is a basic example, we do nothing here
      pass

Now use this middleware as any other:

chat.register_command('ban-me', execute_ban_me, command_middleware=[MyOwnCoinFlipMiddleware()])