Syncer
Syncer is a Python library designed to simplify the conversion of asynchronous functions to synchronous ones. It abstracts away the boilerplate of `asyncio.get_event_loop().run_until_complete()`, making it easier to call `async def` functions from synchronous contexts, particularly useful in testing or when integrating with synchronous codebases. The current version is 2.0.3, with releases occurring periodically to maintain compatibility and add features.
Warnings
- gotcha While `syncer.sync` allows running async code from a synchronous context, it does so by managing an `asyncio` event loop. If `sync()` is called from within an already running `async def` function or an existing event loop, it will raise a `RuntimeError` (e.g., 'Cannot run loop while another loop is running'). It's intended for bridging from non-async contexts.
- gotcha `syncer` converts an async function or coroutine to run synchronously, but it does not magically make blocking I/O operations (e.g., `requests.get()`, `time.sleep()`) non-blocking if they occur within the async function's execution. If your async function internally calls long-running synchronous code, `syncer.sync` will still block the thread it's running on, potentially impacting performance or responsiveness.
- gotcha Be mindful of the distinction between passing an `async def` function itself versus a coroutine object. `sync(async_func)` returns a *synchronous callable* that you then execute (e.g., `sync(async_func)()`). In contrast, `sync(async_func())` immediately executes the *coroutine object* returned by calling `async_func()` and returns its final result. Incorrect usage can lead to unexpected behavior or `TypeError` if you forget to call the returned synchronous function.
Install
-
pip install syncer
Imports
- sync
from syncer import sync
Quickstart
import asyncio
from syncer import sync
async def fetch_data():
await asyncio.sleep(0.01) # Simulate an async I/O operation
return "Async data fetched!"
# --- Usage 1: Run an async function synchronously (returns a callable) ---
synchronous_fetch = sync(fetch_data)
result_function_call = synchronous_fetch() # Call the returned synchronous function
print(f"Result from function call: {result_function_call}")
# --- Usage 2: Run a coroutine synchronously (returns the coroutine's result) ---
result_coroutine_run = sync(fetch_data())
print(f"Result from coroutine run: {result_coroutine_run}")
# --- Usage 3: As a decorator for synchronous test methods ---
import unittest
class MySyncTests(unittest.TestCase):
@sync
async def test_async_fetch(self):
data = await fetch_data()
self.assertEqual(data, "Async data fetched!")
# To run the tests (optional, for demonstration):
# if __name__ == '__main__':
# unittest.main()