concurrency-in-python-with-asyncio / concurrency-in-python-with-asyncio

Code for the Manning book Concurrency in Python with Asyncio

Concurrency in Python with Asyncio

Source code to the Manning book Concurrency in Python with Asyncio.

Running

This code ran successfully with Python version 3.10.2. Using a different version may give you different results or may not work.

Steps to Run Code Listing

1. Install Python 3.10.2, installers available at the bottom of the page at: https://www.python.org/downloads/release/python-3102/

2. Create a virtual environment and activate it. Instructions available here.

3. Install the dependencies for the project by running pip install -r requirements.txt within your virtual environment.

4. Several functions are in the util module - for this to import properly you need to include this module in your PYTHONPATH. You can do this by running export PYTHONPATH=${PYTHONPATH}:concurrency-in-python-with-asyncio/util within your terminal, changing concurrency-in-python-with-asyncio to whichever path you have cloned this repository on your local machine.

5. You should now be able to run any code listing successfully with python3 scriptname.py, for example, python3 chapter_01/listing_1_1.py will run the first code listing from the first chapter.

concurrency-in-python-with-asyncio's Issues

chapter_14 /listing_14_8.py

line18:
self._done_callback(result)

does it mean:
self.add_done_callback(result)

Introducing coroutines (p. 26)

Hi Matt,

In the section "Pausing execution with the await keyword" you wrote the following:

... The await expression will also pause the coroutine where it is contained in until the coroutine we awaited finishes and returns a result.
When the coroutine we awaited finishes, we’ll have access to the result it returned, and the containing coroutine will “wake up” to handle the result.

As I understand, not each await can pause the task (in terms of switching to another task), and the code can be paused and switched to another task only for await where we can actually wait for the result from "outside" (for example, IO with non-blocking sockets).

The only example I found which confirms it is from the awesome series of Łukasz Langa (https://youtu.be/1LTHbmed3D4?t=2162), where the task is not rescheduled on the line with our coroutine's call (return await example(count - 1)) but only on asyncio.sleep calls.

So in our case 2 calls of the coroutine add_one is like any ordinary function's call, and await does not affect on the pausing and switching between tasks (if there are any more), but only for calling like the function.

Therefore there are no "wake up" events here, only calling the functions sequentially and returning results, and all the code will be executed in 1 iteration of the loop without any opportunity to switch to another tasks (if any).
And you can confirm it using the same tracing technique as in the video.

Is it correct? Did you see where this feature is specified in the documentation or mentioned anywhere?
It seems like very important note to understand how await is working, but I can't find a lot of information about it.

Thank you.

Asyncio

Understanding processes, threads, multithreading, and multiprocessing (p. 10)

Hi Matt,

Thanks a lot for your amazing book!

I have a question about Listing 1.3:

import threading

def hello_from_thread():
    print(f'Hello from thread {threading.current_thread()}!')

hello_thread = threading.Thread(target=hello_from_thread)
hello_thread.start()

total_threads = threading.active_count()
thread_name = threading.current_thread().name

print(f'Python is currently running {total_threads} thread(s)')
print(f'The current thread is {thread_name}')

hello_thread.join()

You wrote that as a result we have the output including this line:

Hello from thread <Thread(Thread-1, started 123145541312512)>!
Python is currently running 2 thread(s)
The current thread is MainThread

But sometimes when I execute this code I can see not 2 threads but 1 thread in the output.
Do I understand correctly that sometimes this additional thread can be finished before printing (due to the context switch) so we have only 1 thread in the output?

Thank you.

Listing 10.8

There is the row in get_products_with_inventory function:

logging.exception(f'Error getting inventory for id {product_id}', exc_info=inventory_tasks_to_product_id[done_task].exception())

Then inventory_tasks_to_product_id defined as this:

inventory_tasks_to_product_id = {
        asyncio.create_task(
            inventory_circuit.request(session, product["product_id"])
        ): product["product_id"]
        for product in product_response
    }

The problem is that inventory_tasks_to_product_id[done_task] is just a number. So it doesn't have .exception method. It should be probably simply done_task.

logging.exception(f'Error getting inventory for id {product_id}', exc_info=done_task.exception())

Creating a REST API with aiohttp (p. 225)

Hi Matt,

I found that the example of creating a product (Listing 9.4) is not working as expected, because of the problem which is the same that in the issue #10 - there is an import of the create_database_pool and destroy_database_pool functions from Listing 9.2, where there are no checks like if __name__ == "__main__", so the app is starting twice. We can make sure of this if after starting of the server we will press CTRL+C once (if will stop the first server from Listing 9.2), and after that our request to create product will work.

Thank you.

Asynchronous context managers (p. 78)

Hi Matt,

Probably there is a typo in the following note:

After this statement, aenter will execute, and we’ll close our connection.

It seems that aexit is described here, not aenter.

Thank you.

Using process pool executors with asyncio (p. 135)

Hi Matthew (and all who read it)!

On page 135 you write:

We’ll submit multiple count tasks to the executor and wait for them all to finish with gather. run_in_executor only takes a callable and does not allow us to supply function arguments; so, to get around this, we’ll use partial function application to build countdown calls with 0 arguments.

And give this code (listing_6_5):

async def main():
    with ProcessPoolExecutor() as process_pool:
        loop: AbstractEventLoop = asyncio.get_running_loop()
        nums = [1, 3, 5, 22, 100000000]
        calls: List[partial[int]] = [partial(countdown, num) for num in nums]
        call_coros = []

        for call in calls:
            call_coros.append(loop.run_in_executor(process_pool, call))

        results = await asyncio.gather(*call_coros)

        for result in results:
            print(result)

I changed it on (gave function argument into run_in_executor):

async def main():
    with ProcessPoolExecutor() as process_pool:
        loop: AbstractEventLoop = asyncio.get_running_loop()
        nums = [1, 3, 5, 22, 100000000]
        call_coros = []

        for num in nums:
            call_coros.append(loop.run_in_executor(process_pool, countdown, num))

        results = await asyncio.gather(*call_coros)

        for result in results:
            print(result)

and it still works!

Can you explain this moment? (May be I do something wrong).

Best regards!

chapter_2/listing_2_12.py

Hi Matt!

It is probably better to include the following line in the finally block to confirm whether the task was cancelled:

print(f'Was the task cancelled? {delay_task.cancelled()}')

This ensures that you can accurately determine whether the task has been cancelled, especially when users manipulate the time values in the code.

async def main():
    delay_task = asyncio.create_task(delay(2))
    try:
        result = await asyncio.wait_for(delay_task, timeout=1)
        print(result)
    except asyncio.exceptions.TimeoutError:
        print('Got a timeout!')
    finally:
        print(f'Was the task cancelled? {delay_task.cancelled()}')

Executing queries concurrently with connection pools (p. 110)

Hi Matt,

I noticed that using original code, brand names are inserted with the newline at the end of each name:

def load_common_words() -> List[str]:
    with open('common_words.txt') as common_words:
        return common_words.readlines()

Probably it will be better to strip these names before inserting to database, something like that:

def load_common_words() -> List[str]:
    return [line.strip() for line in open('common_words.txt')]

ASGI with Starlette (p. 230)

Hi Matt,

When I try to reproduce the example of the counter based on the web sockets (start the server and open HTML page), I have the following error (I'm using Python 3.10.4):
TypeError: As of 3.10, the *loop* parameter was removed from Lock() since it is no longer necessary

It seems that this recommendation is working, and when I upgrade websockets package from 9.1 to 10.4 version (the latest version), it is working as expected.

Thank you.

Executing queries concurrently with connection pools (p. 111)

Hi Matt,

I found that if we inserted brands using Listing 5.5 and run Listing 5.6 to insert products and SKUs after it, then there will be 200 rows in the brand table, because code from Listing 5.5 is imported and (because there are no checks likes if __name__=="__main__") executed again.
Also connection is not closed in Listing 5.5.
So the main function and its execution in Listing 5.5 can be changed to avoid these problems:

async def main():
    common_words = load_common_words()
    connection = await asyncpg.connect(host='127.0.0.1',
                                       port=5432,
                                       user='postgres',
                                       database='products',
                                       password='password')
    await insert_brands(common_words, connection)
    await connection.close()


if __name__ == '__main__':
    asyncio.run(main())

Thank you.

Getting error in listings 7.13, 7.14, 7.15

Getting error in the program from listings 7.13, 7.14, 7.15
I'm getting message of error [Errno 1] [SSL: APPLICATION_DATA_AFTER_CLOSE_NOTIFY] application data after close notify (_ssl.c:2672)
n-1 times where n is number putted in the field 'self._request_field'.
And I have no idea what that means.
When i tried to recreate this error with parts of code pertaining only to aiohttp (to make requests outside of Tkinter app just using ClientSession and asyncio.gather) i didn't had such errors. Environment was the same.

RuntimeError: asyncio.run() cannot be called from a running event loop

When I'm running the very first listenings (2.3, 2.4) I have a runtime error.

import asyncio


async def coroutine_add_one(number: int) -> int:
    return number + 1
    

result = asyncio.run(coroutine_add_one(1))

print(result)

Output:

RuntimeError                              Traceback (most recent call last)
c:\Users\mokoto\Desktop\solution\asyn_test.ipynb Cell 2 in <cell line: 8>()
      [4](vscode-notebook-cell:/c%3A/Users/mokoto/Desktop/solution/asyn_test.ipynb#W1sZmlsZQ%3D%3D?line=3) async def coroutine_add_one(number: int) -> int:
      [5](vscode-notebook-cell:/c%3A/Users/mokoto/Desktop/solution/asyn_test.ipynb#W1sZmlsZQ%3D%3D?line=4)     return number + 1
----> [8](vscode-notebook-cell:/c%3A/Users/mokoto/Desktop/solution/asyn_test.ipynb#W1sZmlsZQ%3D%3D?line=7) result = asyncio.run(coroutine_add_one(1))
     [10](vscode-notebook-cell:/c%3A/Users/mokoto/Desktop/solution/asyn_test.ipynb#W1sZmlsZQ%3D%3D?line=9) print(result)

File c:\Users\mokoto\AppData\Local\Programs\Python\Python310\lib\asyncio\runners.py:33, in run(main, debug)
      9 """Execute the coroutine and return the result.
     10 
     11 This function runs the passed coroutine, taking care of
   (...)
     30     asyncio.run(main())
     31 """
     32 if events._get_running_loop() is not None:
---> 33     raise RuntimeError(
     34         "asyncio.run() cannot be called from a running event loop")
     36 if not coroutines.iscoroutine(main):
     37     raise ValueError("a coroutine was expected, got {!r}".format(main))

RuntimeError: asyncio.run() cannot be called from a running event loop

How to resolve this? Python version is 3.10.4