A high-performance inter-process communication library designed to work with the latest version of discord.py
Project description
Better IPC
A high-performance inter-process communication library designed to work with the latest version of discord.py
This library is heavily based on discod-ext-ipc, which is no longer maintained.
Installation
Stable version (Guaranteed to work)
python -m pip install better-ipc
Unstable version (Active getting updates)
python -m pip install -U git+https://github.com/MiroslavRosenov/better-ipc
Inside your Discord client (with decorator)
import logging
import discord
from discord.ext import commands, ipc
from discord.ext.ipc.Server import route
from discord.ext.ipc.errors import IPCError
class Routes(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
if not hasattr(bot, "ipc"):
bot.ipc = ipc.Server(self.bot, host="127.0.0.1", port=2300, secret_key="your_secret_key_here")
bot.ipc.start(self)
@commands.Cog.listener()
async def on_ipc_ready(self):
logging.info("Ipc is ready")
@commands.Cog.listener()
async def on_ipc_error(self, endpoint: str, error: IPCError):
logging.error(endpoint, "raised", error, file=sys.stderr)
@route()
async def get_user_data(self, data):
user = self.bot.get_user(data.user_id)
return user._to_minimal_user_json() # THE OUTPUT MUST BE JSON SERIALIZABLE!
async def setup(bot):
await bot.add_cog(Routes(bot))
Inside your Discord client (with manual endpoint register)
import logging
import discord
from discord.ext import commands, ipc
from discord.ext.ipc.Server import route
from discord.ext.ipc.errors import IPCError
class Routes(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
if not hasattr(bot, "ipc"):
bot.ipc = ipc.Server(self.bot, host="127.0.0.1", port=2300, secret_key="your_secret_key_here")
bot.ipc.start(self)
for name, function in inspect.getmembers(self):
if name.startswith("get_"): # ATTENTION: Every function that stats with `get_` will be registered as endpoint
bot.ipc.endpoints[name] = function
@commands.Cog.listener()
async def on_ipc_ready(self):
logging.info("Ipc is ready")
@commands.Cog.listener()
async def on_ipc_error(self, endpoint: str, error: IPCError):
logging.error(endpoint, "raised", error, file=sys.stderr)
async def get_user_data(self, data):
user = self.bot.get_user(data.user_id)
return user._to_minimal_user_json() # THE OUTPUT MUST BE JSON SERIALIZABLE!
async def setup(bot):
await bot.add_cog(Routes(bot))
Inside your web application
from quart import Quart
from discord.ext.ipc import Client
app = Quart(__name__)
IPC = Client(
host="127.0.0.1",
port=2300,
secret_key="your_secret_key_here"
) # These params must be the same as the ones in the client
@app.route('/')
async def main():
data = await ipc_client.request("get_user_data", user_id=383946213629624322)
return str(data)
if __name__ == '__main__':
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
app.ipc = loop.run_until_complete(IPC.start(loop=loop)) # `Client.start()` returns new Client instance
app.run(loop=loop)
finally:
loop.run_until_complete(app.ipc.close()) # Closes the session, doesn't close the loop
loop.close()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
better-ipc-1.0.2.tar.gz
(9.5 kB
view hashes)
Built Distribution
better_ipc-1.0.2-py3-none-any.whl
(11.5 kB
view hashes)
Close
Hashes for better_ipc-1.0.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a57d7b0c33c2d323fbe9feb102b6d290a99b99dbfb9ebf95088ec9e07e4d4154 |
|
MD5 | b96dd59861cf6cc7600c28a7b2c13eea |
|
BLAKE2b-256 | 817c5f54bd0aa249119bc848f3cd0317b0a4e05be0ca43af255086960eb06c63 |