How to call a async function contained in a class? How to call a async function contained in a class? python python

How to call a async function contained in a class?


Finally I could find the right way to do it (special thanks to @dirn)

#!/usr/bin/env python3import sys, jsonimport asynciofrom websockets import connectclass EchoWebsocket:    async def __aenter__(self):        self._conn = connect('wss://ws.binaryws.com/websockets/v3')        self.websocket = await self._conn.__aenter__()                return self    async def __aexit__(self, *args, **kwargs):        await self._conn.__aexit__(*args, **kwargs)    async def send(self, message):        await self.websocket.send(message)    async def receive(self):        return await self.websocket.recv()class mtest:    def __init__(self):        self.wws = EchoWebsocket()        self.loop = asyncio.get_event_loop()    def get_ticks(self):        return self.loop.run_until_complete(self.__async__get_ticks())    async def __async__get_ticks(self):        async with self.wws as echo:            await echo.send(json.dumps({'ticks_history': 'R_50', 'end': 'latest', 'count': 1}))            return await echo.receive()

And this in main.py:

from testws import *a = mtest()foo = a.get_ticks()print (foo)print ("async works like a charm!")foo = a.get_ticks()print (foo)

This is the output:

root@ubupc1:/home/dinocob# python3 test.py{"count": 1, "end": "latest", "ticks_history": "R_50"}async works like a charm!{"count": 1, "end": "latest", "ticks_history": "R_50"}

Any tip to improve it is welcomed! ;)


Your question and answer are great!They helped me a lot!

Based on your code I was able to create the following class,better matching my need:

import asynciofrom websockets import connectclass TestClient:    def __init__(self, URL):        self.URL = URL        self.conn = None        self.loop = asyncio.get_event_loop()    async def send(self, message):        if self.conn == None:            self.conn = await connect(self.URL)        await self.conn.send(message)    async def receive(self):        return await self.conn.recv()    def ping(self):        return self.loop.run_until_complete(self._ping())    async def _ping(self):        await self.send("Hello World")        return await self.receive()test = TestClient("wss://echo.websocket.org")print(test.ping())