Agnostic IPC for Python programs
Project description
Zonis
A coro based callback system for many to one IPC setups.
pip install zonis
Simplistic example
Client:
import asyncio
from zonis import Client
async def main():
client = Client(url="localhost:12344/ws")
@client.route()
async def ping():
print("Pong'd!")
return "pong"
await client.start()
await client.block_until_closed()
asyncio.run(main())
Server:
import asyncio
import json
from fastapi import FastAPI
from starlette.websockets import WebSocket, WebSocketDisconnect
from zonis import UnknownClient
from zonis.server import Server
app = FastAPI()
server = Server(using_fastapi_websockets=True)
@app.get("/")
async def index():
try:
response = await server.request("ping")
return {"data": response} # Returns pong
except UnknownClient as e:
return e
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
d: str = await websocket.receive_text()
identifier = await server.parse_identify(json.loads(d), websocket)
try:
await asyncio.Future()
except WebSocketDisconnect:
await server.disconnect(identifier)
print(f"Closed connection for {identifier}")
See the examples directory for more use cases.
For documentation, please see here.
Support
Want realtime help? Join the discord here.
Funding
Want a feature added quickly? Want me to help build your software using this?
Sponsor me here
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
zonis-2.1.0.tar.gz
(11.5 kB
view hashes)
Built Distribution
zonis-2.1.0-py3-none-any.whl
(14.2 kB
view hashes)