I'm developping a Twitch Bot for approximatively a year now. Over the time, the bot became bigger and bigger to add features. Now the bot can manage multiple interfaces including Discord, Twitch, Twitch API, Streamlabs... and have a web interface to receive all the callbacks from OAuth authentification APIs and to present some statistics on an HTML page to the streamer.
However, the bigger the bot gets, the more problems I experience. The fact is, I don't think my current architecture is good. This is how it's done currently :
First, I instantiate a Core
class, this class will contain instance of all the bots and interface and store the variables shared between them. It also contains a global asyncio lock for the database (since the entire bot is running on a single async loop, I need to be sure only one interface is talking with the database at a same time).
class Core:
def __init__(self):
# Define session lock for database access
self.session_lock: asyncio.locks.Lock = asyncio.Lock()
# Store bots & API access
self.discord_bot: Optional[discord.Client] = None
self.twitch_bot: Optional[twitchio.ext.commands.Bot] = None # Manage Twitch IRC chat
self.twitch_api: TwitchApi = None # Manage Twitch API
# Other interfaces ...
# Some shared attributes which are read by all interfaces ...
Then, I instantiate all my interfaces by passing them the core. Each interface will register into the core by itself when instantiated. This is an example of an interface initialization (Discord here) :
class DiscordBot(commands.Bot):
def __init__(self, core: Core, **options):
super().__init__(**options)
self.core: Core = core
self.core.discord_bot = self
And the instanciation phase in my main script:
core = Core()
discord_bot = DiscordBot(core)
twitch_bot = TwitchChatBot(core, os.environ['TWITCH_BOT_TMI_TOKEN'], [os.environ['TWITCH_CHANNEL_NAME']])
loop = asyncio.get_event_loop()
loop.create_task(twitch_bot.connect())
loop.create_task(discord_bot.start(os.environ["DISCORD_BOT_TOKEN"]))
loop.run_forever()
Here is a global diagram of my architecture and how it is managed :
This architecture is really convenient because it allows me to have a really simple bridge between interfaces. For example, if I want to post a message on Discord from my Twitch bot, I can simply call self.core.discord_bot.get_channel(...).send()
. Same thing on the other direction.
But I feel like this architecture is no longer sustainable. Currently the Core
class contains more than 5000 lines of code and more than 60 methods shared between all the interfaces. I would like to explode it in multiple file but it's a mess. Moreover, I increasingly think that running all interfaces on the same asynchronous loop is not a good idea in the long run.
I thought of solutions, like separating all the interfaces into different processes. But then how to manage the synchronization between these processes without doing something complicated (I take my example of posting a message on Discord from the Twitch bot). I also looked at solutions like Redis for the synchronisation but I don't really know if it could answer all my concerns... I also thought about using the python import module to directly import a Core instance (and thus not having to register the Core in each interface), but I'm not sure if it would work and if it's a good practice.
I take another example, in the Core
class is instanciated a class processing the interactions between the bot and the community. It is used to vary the interactions with the users (a sort of primitive chatbot). This class MUST be shared between all my interfaces because I want the reactions of the Discord bot to react according to what is happening on Twitch.
Anyway, I would like your expert opinion on this. How would you have organized all this? Thanks :)