Category : python-asyncio

I have an asynchronous function that makes http requests and receives json: async def query_async(name): async with aiohttp.ClientSession() as session: async with session.get(‘https://example.com/’ + name) as resp: response = await resp.json() return response I want to call this function in synchronous code, like this: def prepare_result(): name = ‘some_url’ response = asyncio.run(query_async(name)) … Questions: will ..

Read more

In the python 3 script I need to enter a few commands under root priveleges via asyncssh. Something like this code sample: async with asyncssh.connect(host=IP, port=PORT, username=ORDINARY_USER, password=PASSWORD, known_hosts=None) as conn: result_su = await conn.run(‘echo "verysecretpass" | su’, check=False) result_1 = await conn.run("root_command1", check=False) print(result_1.stdout) result_2 = await conn.run("root_command2", check=False) print(result_2.stdout) This sample is not ..

Read more

I have this code example: import asyncio import aiosqlite async def write_row(db): index = 0 while True: index += 1 await db.execute(f’insert into files(name) values (?)’, (‘file’ + str(index),)) # here it stops if index%10 == 0: await db.commit() await asyncio.sleep(0.5) async def main(): db = await aiosqlite.connect(‘test.sqlite3’) print(db) await db.execute(‘create table if not exists ..

Read more

asyncio.sleep()‘s blocking cousin, time.sleep(), cannot guarantee that it will sleep for the requested amount of time. The actual suspension time may be less than that requested because any caught signal will terminate the sleep() following execution of that signal’s catching routine. asyncio.sleep()‘s documentation does not mention a similar limitation. Is asyncio.sleep() able to make stronger ..

Read more

I have a class that looks like this: class DatabaseTest: def __init__(self): self.headers = { "host": os.environ["HOST"], "username": os.environ["USERNAME"], "password": os.environ["PASSWORD"] } self.db = None self.cursor = None asyncio.create_task(self._connect()) async def _connect(self): self.db = await aiomysql.connect(self.headers[‘host’], self.headers[‘username’], self.headers[‘password’]) self.cursor = await self.db.cursor() await self.cursor.execute(‘use bot’) async def _insertData(self, guildid): sql = f’insert into data(guild) values ..

Read more