Mastering Python Second Edition Release Code
This commit is contained in:
Rick van Hattem
@@ -0,0 +1,4 @@
|
||||
Chapter 7, Async IO
|
||||
##############################################################################
|
||||
|
||||
| Multithreading without Threads demonstrates the usage of asynchronous functions using async def and await so external resources no longer stall your Python processes.
|
||||
@@ -0,0 +1,14 @@
|
||||
import time
|
||||
import asyncio
|
||||
|
||||
|
||||
async def slow_blocking_function(sleep):
|
||||
# Note: you should never use time.sleep() in an async function
|
||||
# Always use asyncio.sleep instead
|
||||
time.sleep(sleep)
|
||||
|
||||
|
||||
print('Slow:')
|
||||
asyncio.run(slow_blocking_function(0.5), debug=True)
|
||||
print('Fast:')
|
||||
asyncio.run(slow_blocking_function(0.05), debug=True)
|
||||
@@ -0,0 +1,72 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def main():
|
||||
print('Hello from main')
|
||||
yield from asyncio.sleep(1)
|
||||
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
loop.run_until_complete(main())
|
||||
loop.close()
|
||||
|
||||
#################################################################
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def main():
|
||||
print('Hello from main')
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
loop.run_until_complete(main())
|
||||
loop.close()
|
||||
|
||||
#################################################################
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def main():
|
||||
print('Hello from main')
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
|
||||
#################################################################
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def main():
|
||||
print('Hello from main')
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
try:
|
||||
loop.run_until_complete(main())
|
||||
finally:
|
||||
# Run the loop again to finish pending tasks
|
||||
loop.run_until_complete(asyncio.sleep(0))
|
||||
|
||||
asyncio.set_event_loop(None)
|
||||
loop.close()
|
||||
|
||||
#################################################################
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def main():
|
||||
print('Hello from main')
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
loop.run_until_complete(main())
|
||||
@@ -0,0 +1,49 @@
|
||||
>>> import time
|
||||
>>> import asyncio
|
||||
|
||||
>>> def normal_sleep():
|
||||
... print('before sleep')
|
||||
... time.sleep(1)
|
||||
... print('after sleep')
|
||||
|
||||
|
||||
>>> def normal_sleeps(n):
|
||||
... for _ in range(n):
|
||||
... normal_sleep()
|
||||
|
||||
|
||||
# Normal execution
|
||||
|
||||
>>> start = time.time()
|
||||
>>> normal_sleeps(2)
|
||||
before sleep
|
||||
after sleep
|
||||
before sleep
|
||||
after sleep
|
||||
>>> print(f'duration: {time.time() - start:.0f}')
|
||||
duration: 2
|
||||
|
||||
##################################################################
|
||||
|
||||
>>> async def asyncio_sleep():
|
||||
... print('before sleep')
|
||||
... await asyncio.sleep(1)
|
||||
... print('after sleep')
|
||||
|
||||
|
||||
>>> async def asyncio_sleeps(n):
|
||||
... coroutines = []
|
||||
... for _ in range(n):
|
||||
... coroutines.append(asyncio_sleep())
|
||||
...
|
||||
... await asyncio.gather(*coroutines)
|
||||
|
||||
>>> start = time.time()
|
||||
>>> asyncio.run(asyncio_sleeps(2))
|
||||
before sleep
|
||||
before sleep
|
||||
after sleep
|
||||
after sleep
|
||||
>>> print(f'duration: {time.time() - start:.0f}')
|
||||
duration: 1
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
>>> import asyncio
|
||||
>>> import selectors
|
||||
|
||||
|
||||
>>> selector = selectors.SelectSelector()
|
||||
>>> loop = asyncio.SelectorEventLoop(selector)
|
||||
>>> asyncio.set_event_loop(loop)
|
||||
@@ -0,0 +1,13 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
class UvLoopPolicy(asyncio.DefaultEventLoopPolicy):
|
||||
def new_event_loop(self):
|
||||
try:
|
||||
from uvloop import Loop
|
||||
return Loop()
|
||||
except ImportError:
|
||||
return super().new_event_loop()
|
||||
|
||||
|
||||
asyncio.set_event_loop_policy(UvLoopPolicy())
|
||||
@@ -0,0 +1,32 @@
|
||||
>>> import time
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> def printer(name):
|
||||
... print(f'Started {name} at {loop.time() - offset:.1f}')
|
||||
... time.sleep(0.2)
|
||||
... print(f'Finished {name} at {loop.time() - offset:.1f}')
|
||||
|
||||
>>> loop = asyncio.new_event_loop()
|
||||
>>> _ = loop.call_at(loop.time() + .2, printer, 'call_at')
|
||||
>>> _ = loop.call_later(.1, printer, 'call_later')
|
||||
>>> _ = loop.call_soon(printer, 'call_soon')
|
||||
>>> _ = loop.call_soon_threadsafe(printer, 'call_soon_threadsafe')
|
||||
|
||||
# Make sure we stop after a second
|
||||
|
||||
>>> _ = loop.call_later(1, loop.stop)
|
||||
|
||||
# Store the offset because the loop requires time to start
|
||||
|
||||
>>> offset = loop.time()
|
||||
|
||||
>>> loop.run_until_complete(asyncio.sleep(0))
|
||||
Started call_soon at 0.0
|
||||
Finished call_soon at 0.2
|
||||
Started call_soon_threadsafe at 0.2
|
||||
Finished call_soon_threadsafe at 0.4
|
||||
Started call_later at 0.4
|
||||
Finished call_later at 0.6
|
||||
Started call_at at 0.6
|
||||
Finished call_at at 0.8
|
||||
@@ -0,0 +1,27 @@
|
||||
>>> import time
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> def executor_sleep():
|
||||
... print('before sleep')
|
||||
... time.sleep(1)
|
||||
... print('after sleep')
|
||||
|
||||
|
||||
>>> async def executor_sleeps(n):
|
||||
... loop = asyncio.get_running_loop()
|
||||
... futures = []
|
||||
... for _ in range(n):
|
||||
... future = loop.run_in_executor(None, executor_sleep)
|
||||
... futures.append(future)
|
||||
...
|
||||
... await asyncio.gather(*futures)
|
||||
|
||||
>>> start = time.time()
|
||||
>>> asyncio.run(executor_sleeps(2))
|
||||
before sleep
|
||||
before sleep
|
||||
after sleep
|
||||
after sleep
|
||||
>>> print(f'duration: {time.time() - start:.0f}')
|
||||
duration: 1
|
||||
@@ -0,0 +1,26 @@
|
||||
import time
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
|
||||
|
||||
def executor_sleep():
|
||||
print('before sleep')
|
||||
time.sleep(1)
|
||||
print('after sleep')
|
||||
|
||||
|
||||
async def executor_sleeps(n):
|
||||
loop = asyncio.get_running_loop()
|
||||
futures = []
|
||||
with concurrent.futures.ProcessPoolExecutor() as pool:
|
||||
for _ in range(n):
|
||||
future = loop.run_in_executor(pool, executor_sleep)
|
||||
futures.append(future)
|
||||
|
||||
await asyncio.gather(*futures)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
start = time.time()
|
||||
asyncio.run(executor_sleeps(2))
|
||||
print(f'duration: {time.time() - start:.0f}')
|
||||
@@ -0,0 +1,139 @@
|
||||
While the `sleep` command is available on most systems, Windows is
|
||||
the notable exception. The Windows alternative for the `sleep`
|
||||
command is the `timeout` command which is not the same but serves
|
||||
the same purpose for these examples.
|
||||
|
||||
Alternatively I can recommand the Git for Windows installer which
|
||||
allows you to install "optional Unix tools".
|
||||
|
||||
>>> import time
|
||||
>>> import subprocess
|
||||
|
||||
|
||||
>>> def subprocess_sleep():
|
||||
... print(f'Started sleep at: {time.time() - start:.1f}')
|
||||
... process = subprocess.Popen(['sleep', '0.1'])
|
||||
... process.wait()
|
||||
... print(f'Finished sleep at: {time.time() - start:.1f}')
|
||||
|
||||
|
||||
>>> start = time.time()
|
||||
|
||||
First, we run this completely sequentially:
|
||||
>>> for _ in range(2):
|
||||
... subprocess_sleep()
|
||||
Started sleep at: 0.0
|
||||
Finished sleep at: 0.1
|
||||
Started sleep at: 0.1
|
||||
Finished sleep at: 0.2
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
|
||||
|
||||
>>> import time
|
||||
>>> import subprocess
|
||||
|
||||
|
||||
>>> def subprocess_sleep():
|
||||
... print(f'Started sleep at: {time.time() - start:.1f}')
|
||||
... return subprocess.Popen(['sleep', '0.1'])
|
||||
|
||||
|
||||
>>> start = time.time()
|
||||
|
||||
Now we start all processes immediately and only wait for output:
|
||||
>>> processes = []
|
||||
>>> for _ in range(2):
|
||||
... processes.append(subprocess_sleep())
|
||||
Started sleep at: 0.0
|
||||
Started sleep at: 0.0
|
||||
|
||||
The processes should be running in the background now:
|
||||
>>> for process in processes:
|
||||
... returncode = process.wait()
|
||||
... print(f'Finished sleep at: {time.time() - start:.1f}')
|
||||
Finished sleep at: 0.1
|
||||
Finished sleep at: 0.1
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
|
||||
>>> import time
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> async def async_process_sleep():
|
||||
... print(f'Started sleep at: {time.time() - start:.1f}')
|
||||
... process = await asyncio.create_subprocess_exec('sleep', '0.1')
|
||||
... await process.wait()
|
||||
... print(f'Finished sleep at: {time.time() - start:.1f}')
|
||||
|
||||
|
||||
>>> async def main():
|
||||
... coroutines = []
|
||||
... for _ in range(2):
|
||||
... coroutines.append(async_process_sleep())
|
||||
... await asyncio.gather(*coroutines)
|
||||
|
||||
|
||||
>>> start = time.time()
|
||||
>>> asyncio.run(main())
|
||||
Started sleep at: 0.0
|
||||
Started sleep at: 0.0
|
||||
Finished sleep at: 0.1
|
||||
Finished sleep at: 0.1
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
|
||||
>>> import time
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> async def run_python_script(script):
|
||||
... process = await asyncio.create_subprocess_exec(
|
||||
... 'python3',
|
||||
... stdout=asyncio.subprocess.PIPE,
|
||||
... stdin=asyncio.subprocess.PIPE,
|
||||
... )
|
||||
... print(f'Executing: {script!r}')
|
||||
... stdout, stderr = await process.communicate(script)
|
||||
... print(f'stdout: {stdout!r}')
|
||||
|
||||
|
||||
>>> asyncio.run(run_python_script(b'print("Hi~")'))
|
||||
Executing: b'print("Hi~")'
|
||||
stdout: b'Hi~\n'
|
||||
|
||||
------------------------------------------------------------------
|
||||
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> async def run_script():
|
||||
... process = await asyncio.create_subprocess_exec(
|
||||
... 'python3',
|
||||
... stdout=asyncio.subprocess.PIPE,
|
||||
... stdin=asyncio.subprocess.PIPE,
|
||||
... )
|
||||
...
|
||||
... # Write a simple Python script to the interpreter
|
||||
... process.stdin.write(b'print("Hi~")')
|
||||
...
|
||||
... # Make sure the stdin is flushed asynchronously
|
||||
... await process.stdin.drain()
|
||||
... # And send the end of file so the Python interpreter will
|
||||
... # start processing the input. Without this the process will
|
||||
... # stall forever.
|
||||
... process.stdin.write_eof()
|
||||
...
|
||||
... # Fetch the lines from the stdout asynchronously
|
||||
... async for line in process.stdout:
|
||||
... # Decode the output from bytes and strip the whitespace
|
||||
... # (newline) at the right
|
||||
... print('stdout:', line.rstrip())
|
||||
...
|
||||
... # Wait for the process to exit
|
||||
... await process.wait()
|
||||
|
||||
|
||||
>>> asyncio.run(run_script())
|
||||
stdout: b'Hi~'
|
||||
@@ -0,0 +1,62 @@
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> HOST = '127.0.0.1'
|
||||
>>> PORT = 1234
|
||||
|
||||
|
||||
>>> async def echo_client(message):
|
||||
... # Open the connection to the server
|
||||
... reader, writer = await asyncio.open_connection(HOST, PORT)
|
||||
...
|
||||
... print(f'Client sending {message!r}')
|
||||
... writer.write(message)
|
||||
...
|
||||
... # We need to drain and write the EOF to stop sending
|
||||
... writer.write_eof()
|
||||
... await writer.drain()
|
||||
...
|
||||
... async for line in reader:
|
||||
... print(f'Client received: {line!r}')
|
||||
...
|
||||
... writer.close()
|
||||
|
||||
|
||||
>>> async def echo(reader, writer):
|
||||
... # Read all lines from the reader and send them back
|
||||
... async for line in reader:
|
||||
... print(f'Server received: {line!r}')
|
||||
... writer.write(line)
|
||||
... await writer.drain()
|
||||
...
|
||||
... writer.close()
|
||||
|
||||
|
||||
>>> async def echo_server():
|
||||
... # Create a TCP server that listens on `HOST`/`PORT` and
|
||||
... # calls `echo` when a client connects.
|
||||
... server = await asyncio.start_server(echo, HOST, PORT)
|
||||
...
|
||||
... # Start listening
|
||||
... async with server:
|
||||
... await server.serve_forever()
|
||||
|
||||
|
||||
>>> async def main():
|
||||
... # Create and run the echo server
|
||||
... server_task = asyncio.create_task(echo_server())
|
||||
...
|
||||
... # Wait a little for the server to start
|
||||
... await asyncio.sleep(0.01)
|
||||
...
|
||||
... # Create a client and send the message
|
||||
... await echo_client(b'test message')
|
||||
...
|
||||
... # Kill the server
|
||||
... server_task.cancel()
|
||||
|
||||
|
||||
>>> asyncio.run(main())
|
||||
Client sending b'test message'
|
||||
Server received: b'test message'
|
||||
Client received: b'test message'
|
||||
@@ -0,0 +1,14 @@
|
||||
>>> import asyncio
|
||||
>>> import aiofiles
|
||||
|
||||
>>> async def main():
|
||||
... async with aiofiles.open('aiofiles.txt', 'w') as fh:
|
||||
... await fh.write('Writing to file')
|
||||
...
|
||||
... async with aiofiles.open('aiofiles.txt', 'r') as fh:
|
||||
... async for line in fh:
|
||||
... print(line)
|
||||
|
||||
|
||||
>>> asyncio.run(main())
|
||||
Writing to file
|
||||
@@ -0,0 +1,21 @@
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> class AsyncGenerator:
|
||||
... def __init__(self, iterable):
|
||||
... self.iterable = iterable
|
||||
...
|
||||
... async def __aiter__(self):
|
||||
... for item in self.iterable:
|
||||
... yield item
|
||||
|
||||
|
||||
>>> async def main():
|
||||
... async_generator = AsyncGenerator([4, 2])
|
||||
...
|
||||
... async for item in async_generator:
|
||||
... print(f'Got item: {item}')
|
||||
|
||||
>>> asyncio.run(main())
|
||||
Got item: 4
|
||||
Got item: 2
|
||||
@@ -0,0 +1,25 @@
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> class AsyncContextManager:
|
||||
... async def __aenter__(self):
|
||||
... print('Hi :)')
|
||||
...
|
||||
... async def __aexit__(self, exc_type, exc_value, traceback):
|
||||
... print('Bye :(')
|
||||
|
||||
|
||||
>>> async def main():
|
||||
... async_context_manager = AsyncContextManager()
|
||||
...
|
||||
... print('Before with')
|
||||
... async with async_context_manager:
|
||||
... print('During with')
|
||||
... print('After with')
|
||||
|
||||
>>> asyncio.run(main())
|
||||
Before with
|
||||
Hi :)
|
||||
During with
|
||||
Bye :(
|
||||
After with
|
||||
@@ -0,0 +1,40 @@
|
||||
>>> import asyncio
|
||||
|
||||
|
||||
>>> class SomeClass:
|
||||
... def __init__(self, *args, **kwargs):
|
||||
... print('Sync init')
|
||||
...
|
||||
... async def init(self, *args, **kwargs):
|
||||
... print('Async init')
|
||||
...
|
||||
... @classmethod
|
||||
... async def create(cls, *args, **kwargs):
|
||||
... # Create an instance of `SomeClass` which calls the
|
||||
... # sync init: `SomeClass.__init__(*args, **kwargs)`
|
||||
... self = cls(*args, **kwargs)
|
||||
... # Now we can call the async init:
|
||||
... await self.init(*args, **kwargs)
|
||||
... return self
|
||||
...
|
||||
... async def close(self):
|
||||
... print('Async destructor')
|
||||
...
|
||||
... def __del__(self):
|
||||
... print('Sync destructor')
|
||||
|
||||
|
||||
>>> async def main():
|
||||
... # Note that we use `SomeClass.create()` instead of
|
||||
... # `SomeClass()` so we also run `SomeClass().init()`
|
||||
... some_class = await SomeClass.create()
|
||||
... print('Using the class here')
|
||||
... await some_class.close()
|
||||
... del some_class
|
||||
|
||||
>>> asyncio.run(main())
|
||||
Sync init
|
||||
Async init
|
||||
Using the class here
|
||||
Async destructor
|
||||
Sync destructor
|
||||
@@ -0,0 +1,5 @@
|
||||
async def printer():
|
||||
print('This is a coroutine')
|
||||
|
||||
|
||||
printer()
|
||||
@@ -0,0 +1,10 @@
|
||||
import time
|
||||
import asyncio
|
||||
|
||||
|
||||
async def main():
|
||||
# Oh no... a synchronous sleep from asyncio code
|
||||
time.sleep(0.2)
|
||||
|
||||
|
||||
asyncio.run(main(), debug=True)
|
||||
@@ -0,0 +1,13 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
async def throw_exception():
|
||||
raise RuntimeError()
|
||||
|
||||
|
||||
async def main():
|
||||
# ignoring an exception from an async def
|
||||
asyncio.create_task(throw_exception())
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,17 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
async def sub_printer():
|
||||
print('Hi from the sub-printer')
|
||||
|
||||
|
||||
async def printer():
|
||||
print('Before creating the sub-printer task')
|
||||
asyncio.create_task(sub_printer())
|
||||
print('After creating the sub-printer task')
|
||||
|
||||
|
||||
async def main():
|
||||
asyncio.create_task(printer())
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,18 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
async def sub_printer():
|
||||
print('Hi from the sub-printer')
|
||||
|
||||
|
||||
async def printer():
|
||||
print('Before creating the sub-printer task')
|
||||
asyncio.create_task(sub_printer())
|
||||
print('After creating the sub-printer task')
|
||||
|
||||
|
||||
async def main():
|
||||
asyncio.create_task(printer())
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,30 @@
|
||||
import asyncio
|
||||
|
||||
|
||||
async def sub_printer():
|
||||
print('Hi from the sub-printer')
|
||||
|
||||
|
||||
async def printer():
|
||||
print('Before creating the sub-printer task')
|
||||
asyncio.create_task(sub_printer())
|
||||
print('After creating the sub-printer task')
|
||||
|
||||
|
||||
async def main():
|
||||
asyncio.create_task(printer())
|
||||
await shutdown()
|
||||
|
||||
|
||||
async def shutdown(timeout=5):
|
||||
tasks = []
|
||||
# Collect all tasks from `asyncio`
|
||||
for task in asyncio.all_tasks():
|
||||
# Make sure we skip our current task so we don't loop
|
||||
if task is not asyncio.current_task():
|
||||
tasks.append(task)
|
||||
|
||||
for future in asyncio.as_completed(tasks, timeout=timeout):
|
||||
await future
|
||||
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user