Skip to main content
Python Software Foundation 20th Year Anniversary Fundraiser  Donate today!

Idiomatic asyncio utilities

Project description


PyPI release version Supported Python versions Test Status Code Coverage

Idiomatic asyncio utilties

NOTE: This project is under early stage of development. The public APIs may break version by version.


I also recommend to try the following asyncio libraries for your happier life.

  • async_timeout: Provides a light-weight timeout wrapper that does not spawn subtasks.
  • aiojobs: Provides a concurrency-limited scheduler for asyncio tasks with graceful shutdown.
  • trio: An alternative implementation of asynchronous IO stack for Python, with focus on cancellation scopes and task groups called "nursery".


Async Context Manager

This is an asynchronous version of contextlib.contextmanager to make it easier to write asynchronous context managers without creating boilerplate classes.

import asyncio
import aiotools

async def mygen(a):
   await asyncio.sleep(1)
   yield a + 1
   await asyncio.sleep(1)

async def somewhere():
   async with mygen(1) as b:
       assert b == 2

Note that you need to wrap yield with a try-finally block to ensure resource releases (e.g., locks), even in the case when an exception is ocurred inside the async-with block.

import asyncio
import aiotools

lock = asyncio.Lock()

async def mygen(a):
   await lock.acquire()
       yield a + 1

async def somewhere():
       async with mygen(1) as b:
           raise RuntimeError('oops')
   except RuntimeError:
       print('caught!')  # you can catch exceptions here.

You can also create a group of async context managers, which are entered/exited all at once using asyncio.gather().

import asyncio
import aiotools

async def mygen(a):
   yield a + 10

async def somewhere():
   ctxgrp = aiotools.actxgroup(mygen(i) for i in range(10))
   async with ctxgrp as values:
       assert len(values) == 10
       for i in range(10):
           assert values[i] == i + 10

Async Server

This implements a common pattern to launch asyncio-based server daemons.

import asyncio
import aiotools

async def echo(reader, writer):
   data = await
   await writer.drain()

async def myworker(loop, pidx, args):
   server = await asyncio.start_server(echo, '', 8888,
       reuse_port=True, loop=loop)
   print(f'[{pidx}] started')
   yield  # wait until terminated
   await server.wait_closed()
   print(f'[{pidx}] terminated')

if __name__ == '__main__':
   # Run the above server using 4 worker processes.
   aiotools.start_server(myworker, num_workers=4)

It handles SIGINT/SIGTERM signals automatically to stop the server, as well as lifecycle management of event loops running on multiple processes. Internally it uses aiotools.fork module to get kernel support to resolve potential signal/PID related races via PID file descriptors on supported versions (Python 3.9+ and Linux kernel 5.4+).

Async TaskGroup

A TaskGroup object manages the lifecycle of sub-tasks spawned via its create_task() method by guarding them with an async context manager which exits only when all sub-tasks are either completed or cancelled.

This is motivated from trio's nursery API and a draft implementation is adopted from EdgeDB's Python client library.

import aiotools

async def do():
    async with aiotools.TaskGroup() as tg:
    # at this point, all subtasks are either cancelled or done.

Async Timer

import aiotools

i = 0

async def mytick(interval):
   i += 1

async def somewhere():
   t = aiotools.create_timer(mytick, 1.0)
   await t

t is an asyncio.Task object. To stop the timer, call t.cancel(); await t. Please don't forget await-ing t because it requires extra steps to cancel and await all pending tasks. To make your timer function to be cancellable, add a try-except clause catching asyncio.CancelledError since we use it as a termination signal.

You may add TimerDelayPolicy argument to control the behavior when the timer-fired task takes longer than the timer interval. DEFAULT is to accumulate them and cancel all the remainings at once when the timer is cancelled. CANCEL is to cancel any pending previously fired tasks on every interval.

import asyncio
import aiotools

async def mytick(interval):
   await asyncio.sleep(100)  # cancelled on every next interval.

async def somewhere():
   t = aiotools.create_timer(mytick, 1.0, aiotools.TimerDelayPolicy.CANCEL)
   await t

Virtual Clock

It provides a virtual clock that advances the event loop time instantly upon any combination of asyncio.sleep() calls in multiple coroutine tasks, by temporarily patching the event loop selector.

This is also used in our timer test suite.

import aiotools
import pytest

async def test_sleeps():
    loop = aiotools.compat.get_running_loop()
    vclock = aiotools.VirtualClock()
    with vclock.patch_loop():
        print(loop.time())  # -> prints 0
        await asyncio.sleep(3600)
        print(loop.time())  # -> prints 3600


.. towncrier release notes start

1.2.1 (2021-01-12)


  • Avoid side effects of custom clone() function and resorts back to the combinatino of os.fork() and os.pidfd_open() for now (#25)

1.2.0 (2021-01-12)

Breaking Changes

  • server: The use_threading argument for start_server() is completely deprecated. (#23)


  • Now the primary target is Python 3.9, though we still support from Python 3.6 (#22)
  • fork: Add a new module fork to support PID file descriptors in Linux 5.4+ and a POSIX-compatible fallback to asynchornously fork the Python process without signal/PID races. (#22)
  • server: Completely rewrote the module using the new fork module with handling of various edge cases such as async failures of sibiling child processes (#23)

1.1.1 (2020-12-16)


  • Fix a potential memory leak with TaskGroup when it's used for long-lived asyncio tasks. (#21)

1.1.0 (2020-10-18)


  • Add a current_taskgroup context-variable to the taskgroup module (only available for Python 3.7 or later)


  • Fix missing auto-import of taskgroup module exports in the aiotools root package.

1.0.0 (2020-10-18)


  • Adopt an implementation of the taskgroup API as aiotools.taskgroup from EdgeDB (#18)
  • Add timer.VirtualClock which provides a virtual clock that makes a block of asyncio codes using asyncio.sleep() to complete instantly and deterministically (#19)


  • Adopt towncrier for changelog management (#15)
  • Migrate to GitHub Actions for CI (#19)

0.9.1 (2020-02-25)

  • A maintenance release to fix up the defer module exports in the aiotools namespace.

0.9.0 (2020-02-25)

  • defer: A new module that emulates Golang's defer() API with asyncio awareness.

0.8.5 (2019-11-19)

  • server: Rewrite internals of the worker main functions to use native async with instead of manually unrolling __aenter__() and __aexit__() dunder methods, to keep the code simple and avoid potential bugs.

0.8.4 (2019-11-18)

  • Python 3.8 is now officially supported.
  • server: Fix errors when multiprocessing.set_start_method("spawn") is used.
    • NOTE: This is now the default for macOS since Python 3.8.
    • KNOWN ISSUE: #12
  • Remove some packaging hacks in and let setuptools read the version from a separate aiotools/VERSION text file.

0.8.3 (2019-10-07)

  • context: Fix aclosing()'s __aexit__() exception arguments.

0.8.2 (2019-08-28)

  • context, server: Catch asyncio.CancelledError along with BaseException to make the cancellation behavior consistent in Python 3.6, 3.7, and 3.8.

0.8.1 (2019-02-24)

  • server: Fix yields of the received stop signal in main/worker context managers when using threaded workers.

0.8.0 (2018-11-18)

  • server: Updated stop signal handling and now user-defined worker/main context managers have a way to distinguish the stop signal received. See the updated docs for more details.

0.7.3 (2018-10-16)

  • This ia a technical release to fix a test case preventing the automated CI release procedure.

0.7.2 (2018-10-16)

  • Improve support for Python 3.6/3.7 using a small compatibility module against asyncio.
  • func: Add expire_after option to lru_cache() function.

0.7.1 (2018-08-24)

  • Minor updates to the documentation

0.7.0 (2018-08-24)

  • Add support for Python 3.7
  • context: Updated to work like Python 3.7
  • context: Deprecated AsyncContextDecorator stuffs in Python 3.7+
  • context: Added an alias to contextlib.AsyncExitStack in the standard library.

0.6.0 (2018-04-10)

  • Introduce a new module aiotools.iter with aiter() function which corresponds to an async version of the builtin iter().

0.5.4 (2018-02-01)

  • server: Remove use of unncessary setpgrp syscall, which is also blocked by Docker's default seccomp profile!

0.5.3 (2018-01-12)

  • server: Ooops! (a finally block should have been an else block)

0.5.2 (2018-01-12)

  • server: Improve inner beauty (code readability)
  • server: Improve reliability and portability of worker-to-main interrupts

0.5.1 (2018-01-11)

  • server: Fix a race condition related to handling of worker initialization errors with multiple workers

0.5.0 (2017-11-08)

  • func: Add lru_cache() which is a coroutine version of functools.lru_cache()

0.4.5 (2017-10-14)

  • server: Fix a race condition related to signal handling in the multiprocessing module during termination
  • server: Improve error handling during initialization of workers (automatic shutdown of other workers and the main loop after logging the exception)

0.4.4 (2017-09-12)

  • Add a new module aiotools.func with apartial() function which is an async version of functools.partial() in the standard library

0.4.3 (2017-08-06)

  • Add aclosing() context manager like closing() in the standard library
  • Speed up Travis CI builds for packaging
  • Now provide README in rst as well as CHANGES (this file)

0.4.2 (2017-08-01)

  • server: Fix spawning subprocesses in child workers
  • Add support for uvloop

0.4.0 (2017-08-01)

  • Add use_threading argument to
  • Add initial documentation (which currently not served on due to Python version problem)

0.3.2 (2017-07-31)

  • Add extra_procs argument to start_server() function
  • Add socket and ZeroMQ server examples
  • Improve CI configs

0.3.1 (2017-07-26)

  • Improve CI scripts
  • Adopt editorconfig

0.3.0 (2017-04-26)

  • Add start_server() function using multiprocessing with automatic children lifecycle management
  • Clarify the semantics of AsyncContextGroup using asyncio.gather() with return_exceptions=True

0.2.0 (2017-04-20)

  • Add abstract types for AsyncContextManager
  • Rename AsyncGenContextManager to AsyncContextManager
  • Add AsyncContextGroup

0.1.1 (2017-04-14)

  • Initial release

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for aiotools, version 1.2.1
Filename, size File type Python version Upload date Hashes
Filename, size aiotools-1.2.1-py3-none-any.whl (28.3 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size aiotools-1.2.1.tar.gz (31.1 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page