Synchronicity
Synchronicity is a Python library designed to simplify the development of libraries that need to offer both synchronous (blocking) and asynchronous (non-blocking) APIs from a single async implementation. It achieves this by creating an event loop on a separate thread, wrapping functions, generators, and classes, allowing them to be called synchronously or asynchronously via a `.aio` attribute. The current version is 0.12.1, and it appears to be actively maintained with regular updates.
Warnings
- gotcha When a class is wrapped by `synchronizer.wrap`, instances become proxies. Direct access to attributes of the original class might not work; you may need to define getter methods or `@properties` on the original class for them to be reachable on the wrapper.
- gotcha Synchronicity isolates library execution to its own event loop and thread, which helps prevent accidental blocking of a user's main event loop. However, any long-running, non-async calls within *your* original async functions (or outside Synchronicity's wrapper) can still block the event loop where they execute, negating async benefits.
- gotcha All synchronized code runs on a different thread and event loop. While this provides isolation benefits, it introduces minor overhead due to thread switching and inter-thread communication.
- gotcha Wrapping classes with `synchronizer.wrap` creates a new class (with the same name). This can lead to unexpected type-checking issues or runtime errors if you mix usage of the original class and the wrapped class without careful consideration of the type transformation.
- gotcha Synchronicity's `wrap` decorator works for classes that are context managers, but it does not support wrapping functions that return context managers.
Install
-
pip install synchronicity
Imports
- Synchronizer
from synchronicity import Synchronizer
Quickstart
import asyncio
from synchronicity import Synchronizer
synchronizer = Synchronizer()
@synchronizer.wrap
async def my_async_function(x):
await asyncio.sleep(0.1)
return x**2
# --- Synchronous usage (blocking) ---
print(f"Synchronous call: my_async_function(5) = {my_async_function(5)}") # Blocks until result is ready
# --- Asynchronous usage (non-blocking) ---
async def main_async():
print(f"Asynchronous call (awaiting): await my_async_function.aio(10) = {await my_async_function.aio(10)}")
# Example with multiple concurrent calls from an async context
results = await asyncio.gather(
my_async_function.aio(2),
my_async_function.aio(3),
my_async_function.aio(4)
)
print(f"Concurrent async calls: {results}")
asyncio.run(main_async())