sniffio: Sniff out which async library your code is running under

You’re writing a library. You’ve decided to be ambitious, and support multiple async I/O packages, like Trio, and asyncio, and … You’ve written a bunch of clever code to handle all the differences. But… how do you know which piece of clever code to run?

This is a tiny package whose only purpose is to let you detect which async library your code is running under.

This library is maintained by the Trio project, as a service to the async Python community as a whole.

Usage

sniffio.current_async_library() → str

Detect which async library is currently running.

The following libraries are currently supported:

Library Requires Magic string
Trio Trio v0.6+ "trio"
Curio
"curio"
asyncio   "asyncio"
Trio-asyncio v0.8.2+ "trio" or "asyncio", depending on current mode
Returns:A string like "trio".
Raises:AsyncLibraryNotFoundError – if called from synchronous context, or if the current async library was not recognized.

Examples

from sniffio import current_async_library

async def generic_sleep(seconds):
    library = current_async_library()
    if library == "trio":
        import trio
        await trio.sleep(seconds)
    elif library == "asyncio":
        import asyncio
        await asyncio.sleep(seconds)
    # ... and so on ...
    else:
        raise RuntimeError(f"Unsupported library {library!r}")
exception sniffio.AsyncLibraryNotFoundError

Adding support to a new async library

If you’d like your library to be detected by sniffio, it’s pretty easy.

Step 1: Pick the magic string that will identify your library. To avoid collisions, this should match your library’s PEP 503 normalized name on PyPI.

Step 2: There’s a special threading.local object:

thread_local.name

Make sure that whenever your library is calling a coroutine throw(), send(), or close() that this is set to your identifier string. In most cases, this will be as simple as:

from sniffio import thread_local

# Your library's step function
def step(...):
     old_name, thread_local.name = thread_local.name, "my-library's-PyPI-name"
     try:
         result = coro.send(None)
     finally:
         thread_local.name = old_name

Step 3: Send us a PR to add your library to the list of supported libraries above.

That’s it!

There are libraries that directly drive a sniffio-naive coroutine from another, outer sniffio-aware coroutine such as trio_asyncio. These libraries should make sure to set the correct value while calling a synchronous function that will go on to drive the sniffio-naive coroutine.

from sniffio import thread_local

# Your library's compatibility loop
async def main_loop(self, ...) -> None:
     ...
     handle: asyncio.Handle = await self.get_next_handle()
     old_name, thread_local.name = thread_local.name, "asyncio"
     try:
         result = handle._callback(obj._args)
     finally:
         thread_local.name = old_name