Skip to main content

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

Authors

zaptools was written by Nathan Mejia

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

zaptools-0.0.1.tar.gz (7.0 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page