An async library for subscribing to the Pusher WebSocket protocol.
You can install aiopusher
via pip from PyPI:
pip install aiopusher
Or with Poetry:
poetry add aiopusher
Here are some examples of using aiopusher
:
import asyncio
from aiopusher import Pusher
async def main():
async with Pusher('<your-app-key>') as client:
channel = await client.subscribe('<channel-name>')
channel.bind('<event-name>', lambda data: print(data))
# Run forever (or until manually stopped)
while True:
await asyncio.sleep(1)
asyncio.run(main())
async def main():
client = Pusher('<your-app-key>')
await client.connect()
channel = await client.subscribe('<channel-name>')
channel.bind('<event-name>', lambda data: print(data))
while True:
await asyncio.sleep(1)
await client.disconnect() # Yes, I know this cannot technically be reached
import asyncio
from aiopusher import Pusher
client = Pusher('<your-app-key>')
@client.event('<channel-name>', '<event-name>')
async def handle_event(data):
print(data)
async def main():
await client.connect()
# Run forever (or until manually stopped)
while True:
await asyncio.sleep(1)
asyncio.run(main())
import asyncio
from aiopusher import Pusher
async def main():
options = {
host: 'api.example.com', # default: 'ws.pusherapp.com'
"userAuthentication": {
"endpoint": "/auth",
"transport": "ajax",
}
}
async with Pusher('<your-app-key>', options) as client:
channel = await client.subscribe('<channel-name>')
channel.bind('<event-name>', lambda data: print(data))
# Run forever (or until manually stopped)
while True:
await asyncio.sleep(1)
asyncio.run(main())
import asyncio
from aiopusher import Pusher, SingletonClient
async def main():
client = Pusher('<your-app-key>')
SingletonClient.set_client(client)
await client.connect()
channel = await SingletonClient.get_client().subscribe('<channel-name>')
channel.bind('<event-name>', lambda data: print(data))
while True:
await asyncio.sleep(1)
If you want to contribute to this project, please read CONTRIBUTING.md.