![pypi badge](https://img.shields.io/pypi/v/synchronicity.svg?style=flat)
Python 3 has some amazing support for async programming but it's arguably made it a bit harder to develop libraries. Are you tired of implementing synchronous and asynchronous methods doing basically the same thing? This might be a simple solution for you.
Installing
pip install synchronicity
Background: why is anything like this needed
Let's say you have an asynchronous function
async def f(x):
await asyncio.sleep(1.0)
return x**2
And let's say (for whatever reason) you want to offer a synchronous API to users. For instance maybe you want to make it easy to run your code in a basic script, or a user is building something that's mostly CPU-bound, so they don't want to bother with asyncio.
A "simple" way to create a synchronous equivalent would be to implement a set of synchronous functions where all they do is call asyncio.run on an asynchronous function. But this isn't a great solution for more complex code:
- It's kind of tedious grunt work to have to do this for every method/function
- asyncio.run doesn't work with generators
- In many cases, you need to preserve an event loop running between calls.
The last case is particularly challenging. For instance, let's say you are implementing a client to a database that needs to have a persistent connection, and you want to built it in asyncio:
class DBConnection:
def __init__(self, url):
self._url = url
async def connect(self):
self._connection = await connect_to_database(self._url)
async def query(self, q):
return await self._connection.run_query(q)
How do you expose a synchronous interface to this code? The problem is that wrapping connect
and query
in asyncio.run won't work since you need to preserve the event loop across calls. It's clear we need something slightly more advanced.
How to use this library
This library offers a simple Synchronizer
class that creates an event loop on a separate thread, and wraps functions/generators/classes so that synchronous execution happens on that thread. When you call anything, it will detect if you're running in a synchronous or asynchronous context, and behave correspondingly.
- In the synchronous case, it will simply block until the result is available (note that you can make it return a future as well, see later)
- In the asynchronous case, it works just like the usual business of calling asynchronous code
from synchronicity import Synchronizer
synchronizer = Synchronizer()
@synchronizer.create_blocking
async def f(x):
await asyncio.sleep(1.0)
return x**2
ret = f(42)
print('f(42) =', ret)
async def g():
ret = await f(42)
print('f(42) =', ret)
More advanced examples
Generators
The decorator also works on generators:
@synchronizer.create_blocking
async def f(n):
for i in range(n):
await asyncio.sleep(1.0)
yield i
for ret in f(10):
print(ret)
Synchronizing whole classes
It also operates on classes by wrapping every method on the class:
@synchronizer.create_blocking
class DBConnection:
def __init__(self, url):
self._url = url
async def connect(self):
self._connection = await connect_to_database(self._url)
async def query(self, q):
return await self._connection.run_query(q)
db_conn = DBConnection('tcp://localhost:1234')
db_conn.connect()
data = db_conn.query('select * from foo')
Returning futures
You can also make functions return a Future
object by adding _future=True
to any call. This can be useful if you want to dispatch many calls from a blocking context, but you want to resolve them roughly in parallel:
from synchronicity import Synchronizer
synchronizer = Synchronizer()
@synchronizer.create_blocking
async def f(x):
await asyncio.sleep(1.0)
return x**2
futures = [f(i, _future=True) for i in range(10)]
rets = [fut.result() for fut in futures]
print('first ten squares:', rets)
Using with with other asynchronous code
This library can also be useful in purely asynchronous settings, if you have multiple event loops, or if you have some section that is CPU-bound, or some critical code that you want to run on a separate thread for safety. All calls to synchronized functions/generators are thread-safe by design. This makes it a useful alternative to loop.run_in_executor for simple things. Note however that each synchronizer only runs one thread.
Context managers
You can synchronize context manager classes just like any other class and the special methods will be handled properly.
There's also a function decorator @synchronizer.asynccontextmanager
which behaves just like https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager but works in both synchronous and asynchronous contexts.
Gotchas
- It works for classes that are context managers, but not for functions returning a context manager
- It creates a new class (with the same name) when wrapping classes, which might lead to typing problems if you have any any un-synchronized usage of the same class
- No idea how this interacts with typing annotations
- If a class is "synchronized", it wraps all the methods on the class, but this typically means you can't reach into attributes and run asynchronous code on it: you might get errors such as "attached to a different loop"
- Note that all synchronized code will run on a different thread, and a different event loop, so calling the code might have some minor extra overhead
TODOs
- Support the opposite case, i.e. you have a blocking function/generator/class/object, and you want to call it asynchronously (this is relatively simple to do for plain functions using loop.run_in_executor, but Python has no built-in support for generators, and it would be nice to transform a whole class
- More documentation
- Make it possible to annotate methods selectively to return futures
- Maybe make it possible to synchronize objects on the fly, not just classes
This library is limb-amputating edge
This is code I broke out of a personal projects, and it's not been battle-tested. There is a small test suite that you can run using pytest.
Release process
Should automate this...
- Make a new branch
release-X.Y.Z
from main - Bump version in pyproject.toml to
X.Y.Z
- Commit that change and create a PR
- Merge the PR once green
- Checkout main
git tag -a vX.Y.Z -m "* release bullets"
- git push --tags
TWINE_USERNAME=__token__ TWINE_PASSWORD="$PYPI_TOKEN_SYNCHRONICITY" make publish