Python Implementation to ZapTools WebSockets
Project description
PreAlfa 2
Wrapper around WebSocket to ZapTools Clients
Usage
How to use zaptools
from fastapi import FastAPI
from zaptools import FastApiZapAdapter, SocketClient
app : FastAPI = FastAPI()
adapter = FastApiZapAdapter(app= app, route= "/" )
@adapter.on_client_connected
async def on_connected(client : SocketClient , adapter : FastApiZapAdapter):
print("client connected : ", client.id_connection)
await client.send_event("myEvent", "Data Payload to send")
@adapter.on_event("event")
async def event(payload, client:SocketClient , adapter : FastApiZapAdapter ):
print(payload)
await client.send_event("anyEvent", {"msg" : "a Json Format"})
@adapter.on_client_disconnected
async def client_disconected(client : SocketClient , adapter : FastApiZapAdapter):
print("client disconnected: ", client.id_connection)
zaptools is only compatible with FastAPI apps, so we need to create a FastAPI app and then create a instance of FastApiZapAdapter, the constructor need the app (FastApi app) and a specific route (“/”)
app : FastAPI = FastAPI()
adapter = FastApiZapAdapter(app= app, route= "/" )
Now we can use the adapter to define the function to be called when a event is triggered, for example is we need to check if a new client is connected to our Socket server, and then response to client:
@adapter.on_client_connected
async def on_connected(client : SocketClient , adapter : FastApiZapAdapter):
print("client connected : ", client.id_connection)
await client.send_event("myEvent", "Data Payload to send")
the SocketClient is a class that have id_connection and send_event method, remember ever await the send_event.
Also, you can define a function to be called when a client is disconnected:
@adapter.on_client_disconnected
async def client_disconected(client : SocketClient , adapter : FastApiZapAdapter):
print("client disconnected: ", client.id_connection)
We can register a event, in this case we have the payload:
@adapter.on_event("eventName")
async def event(payload, client:SocketClient , adapter : FastApiZapAdapter ):
print(payload)
await client.send_event("eventName", {"msg" : "a Json Format"})
And finally, start the FastAPI app normally (uvicorn)
Installation
pip install zaptools
Requirements
FastApi
Compatibility
Python 3.7+
Licence
MIT
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.