python-asyncio

SqlAlchemy asyncio orm: How to query the database

by

session.query is the old API. The asynchronous version uses select and accompanying methods. from sqlalchemy.future import select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker engine = create_async_engine(_build_async_db_uri(CONFIG.general.sqlalchemy_db_uri)) async_session = sessionmaker( engine, expire_on_commit=False, class_=AsyncSession ) CACHE = {} async def _load_all(): global CACHE try: async with async_session() as session: q = select(TableClass) result = … Read more

how to cache asyncio coroutines

by

Maybe a bit late, but I’ve started a new package that may help: https://github.com/argaen/aiocache. Contributions/comments are always welcome. An example: import asyncio from collections import namedtuple from aiocache import cached from aiocache.serializers import PickleSerializer Result = namedtuple(‘Result’, “content, status”) @cached(ttl=10, serializer=PickleSerializer()) async def async_main(): print(“First ASYNC non cached call…”) await asyncio.sleep(1) return Result(“content”, 200) if … Read more

How to cancel all remaining tasks in gather if one fails?

by

The problem with your implementation is that it calls sleepers.cancel() after sleepers has already raised. Technically the future returned by gather() is in a completed state, so its cancellation must be no-op. To correct the code, you just need to cancel the children yourself instead of trusting gather‘s future to do it. Of course, coroutines … Read more

Await an async function in Python debugger

by

Here a modified version of what I posted at https://stackoverflow.com/a/67847257/893857 since it also solves this problem. I found a solution using nest_asyncio. If one has the following async example script: import asyncio import nest_asyncio async def bar(x): return x + 1 async def foo(): import ipdb; ipdb.set_trace() if __name__==”__main__”: loop = asyncio.get_event_loop() nest_asyncio.apply(loop) loop.run_until_complete(foo()) One … Read more

How to mock aiohttp.client.ClientSession.get async context manager

by

In your link, there is an edit: EDIT: A GitHub issue mentioned in this post has been resolved and as of version 0.11.1 asynctest supports asynchronous context managers out of the box. Since asynctest==0.11.1, it was changed, a working example is: import random from aiohttp import ClientSession from asynctest import CoroutineMock, patch async def get_random_photo_url(): … Read more

FastAPI runs api-calls in serial instead of parallel fashion

by

As per FastAPI’s documentation: When you declare a path operation function with normal def instead of async def, it is run in an external threadpool that is then awaited, instead of being called directly (as it would block the server). also, as described here: If you are using a third party library that communicates with … Read more

Python 3.5 async/await with real code example

by

If a third-party library is not compatible with async/await then obviously you can’t use it easily. There are two cases: Let’s say that the function in the library is asynchronous and it gives you a callback, e.g. def fn(…, clb): … So you can do: def on_result(…): … fn(…, on_result) In that case you can … Read more

Use asyncio and Tkinter (or another GUI lib) together without freezing the GUI

by

Trying to run both event loops at the same time is a dubious proposition. However, since root.mainloop simply calls root.update repeatedly, one can simulate mainloop by calling update repeatedly as an asyncio task. Here is a test program that does so. I presume adding asyncio tasks to the tkinter tasks would work. I checked that … Read more

Hata!: SQLSTATE[HY000] [1045] Access denied for user 'divattrend_liink'@'localhost' (using password: YES)