Async Helpers
The pydoover utilities module provides several functions and decorators for writing code that works seamlessly in both synchronous and asynchronous contexts. These are particularly useful when building libraries or applications that need to support both execution models.
Import
from pydoover.utils import (
call_maybe_async,
maybe_async,
get_is_async,
wrap_try_except,
wrap_try_except_async,
)
call_maybe_async()
An async helper function that can call either synchronous or asynchronous functions, with options for running as a task or in an executor.
Signature
async def call_maybe_async(
func,
*args,
as_task: bool = False,
in_executor: bool = True,
**kwargs
)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
func | callable | required | The function to call (sync or async) |
*args | any | - | Positional arguments to pass to the function |
as_task | bool | False | If True, run as an asyncio task |
in_executor | bool | True | If True, run sync functions in an executor |
**kwargs | any | - | Keyword arguments to pass to the function |
Return Value
Returns the result of the function call. If as_task=True, returns an asyncio.Task or asyncio.Future object instead.
Behavior
The function handles different scenarios:
- Async function: Awaits the coroutine directly (or creates a task if
as_task=True) - Sync function with executor: Runs in the default executor to avoid blocking
- Sync function without executor: Calls the function directly
Examples
Basic Usage
import asyncio
from pydoover.utils import call_maybe_async
def sync_operation(x, y):
return x + y
async def async_operation(x, y):
await asyncio.sleep(0.1)
return x * y
async def main():
# Call sync function (runs in executor by default)
result1 = await call_maybe_async(sync_operation, 2, 3)
print(result1) # 5
# Call async function
result2 = await call_maybe_async(async_operation, 2, 3)
print(result2) # 6
asyncio.run(main())
Running as a Task
async def main():
# Create a task that runs in the background
task = await call_maybe_async(
async_operation,
10, 20,
as_task=True
)
# Do other work...
print("Task is running in background")
# Wait for the task to complete
result = await task
print(f"Result: {result}")
Running Without Executor
async def main():
# Run sync function directly (may block the event loop)
result = await call_maybe_async(
sync_operation,
5, 10,
in_executor=False
)
print(result) # 15
Important Notes
- When using
in_executor=Truewith a sync function, kwargs are not supported and a warning will be logged if provided - The
as_task=Trueoption is useful for fire-and-forget operations or when you want to run multiple operations concurrently
maybe_async()
A decorator that provides a unified interface for methods that have both sync and async implementations. When the object is in an async context, it automatically calls the async variant of the method.
Signature
def maybe_async() -> Callable
How It Works
- Decorate your sync method with
@maybe_async() - Create an async variant with the same name plus
_asyncsuffix - The decorator checks the instance's
_is_asyncattribute to determine which version to call
Example
from pydoover.utils import maybe_async
class DataFetcher:
def __init__(self, is_async: bool = False):
self._is_async = is_async
@maybe_async()
def fetch_data(self, url: str):
"""Sync version - makes blocking HTTP request"""
import requests
response = requests.get(url)
return response.json()
async def fetch_data_async(self, url: str):
"""Async version - makes non-blocking HTTP request"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
Using in Sync Context
# Sync usage
fetcher = DataFetcher(is_async=False)
data = fetcher.fetch_data("https://api.example.com/data")
print(data)
Using in Async Context
import asyncio
async def main():
# Async usage - automatically calls fetch_data_async
fetcher = DataFetcher(is_async=True)
data = await fetcher.fetch_data("https://api.example.com/data")
print(data)
asyncio.run(main())
Force Sync in Async Context
async def main():
fetcher = DataFetcher(is_async=True)
# Force using sync version even in async context
data = fetcher.fetch_data("https://api.example.com/data", run_sync=True)
print(data)
Usage with pydoover Applications
The maybe_async() decorator is particularly useful when building pydoover applications that need to support both sync and async execution modes:
from pydoover.docker import Application
from pydoover.utils import maybe_async
class MyApplication(Application):
@maybe_async()
def process_data(self, data):
# Sync processing
return self._transform(data)
async def process_data_async(self, data):
# Async processing with non-blocking operations
await asyncio.sleep(0) # Yield to event loop
return self._transform(data)
def _transform(self, data):
# Shared transformation logic
return {k: v * 2 for k, v in data.items()}
get_is_async()
Determines whether the code is running in an async context.
Signature
def get_is_async(is_async: bool = None) -> bool
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
is_async | bool | None | Override value; if provided, returns this value |
Return Value
Returns True if running in an async context (event loop is running), False otherwise.
Example
from pydoover.utils import get_is_async
import asyncio
def check_context():
if get_is_async():
print("Running in async context")
else:
print("Running in sync context")
# Sync context
check_context() # "Running in sync context"
# Async context
async def main():
check_context() # "Running in async context"
asyncio.run(main())
Override Behavior
# Force a specific value result = get_is_async(is_async=True) # Always True result = get_is_async(is_async=False) # Always False
wrap_try_except()
Wraps a synchronous function call to catch and log exceptions without propagating them.
Signature
def wrap_try_except(func, *args, **kwargs)
Parameters
| Parameter | Type | Description |
|---|---|---|
func | callable | The function to call |
*args | any | Positional arguments for the function |
**kwargs | any | Keyword arguments for the function |
Return Value
Returns the function's return value, or None if an exception occurred.
Example
from pydoover.utils import wrap_try_except
def risky_operation(x):
if x < 0:
raise ValueError("x must be non-negative")
return x ** 2
# Success case
result = wrap_try_except(risky_operation, 5)
print(result) # 25
# Exception case - logged but not raised
result = wrap_try_except(risky_operation, -1)
print(result) # None (exception was caught and logged)
wrap_try_except_async()
Async version of wrap_try_except() for wrapping async function calls.
Signature
async def wrap_try_except_async(func, *args, **kwargs)
Example
import asyncio
from pydoover.utils import wrap_try_except_async
async def risky_async_operation(x):
await asyncio.sleep(0.1)
if x < 0:
raise ValueError("x must be non-negative")
return x ** 2
async def main():
# Success case
result = await wrap_try_except_async(risky_async_operation, 5)
print(result) # 25
# Exception case - logged but not raised
result = await wrap_try_except_async(risky_async_operation, -1)
print(result) # None
asyncio.run(main())
Best Practices
When to Use call_maybe_async()
- When you need to call callbacks that could be either sync or async
- When building event handlers or plugin systems that accept user-provided functions
- When integrating with external libraries that may provide sync or async interfaces
When to Use maybe_async()
- When building library code that needs to support both sync and async users
- When you have significant differences between sync and async implementations
- When working with the pydoover Application framework
Executor Considerations
When using call_maybe_async() with in_executor=True:
- Thread safety: Ensure the sync function is thread-safe
- Blocking operations: Long-blocking operations should use the executor
- Shared state: Be careful with shared mutable state across threads
import asyncio
from pydoover.utils import call_maybe_async
def cpu_intensive_task(data):
# This will run in a thread pool, not blocking the event loop
result = 0
for i in range(1000000):
result += i
return result
async def main():
# Run CPU-intensive task without blocking
result = await call_maybe_async(cpu_intensive_task, [1, 2, 3])
print(result)
Related Pages
- Utilities Overview - Overview of all utility functions
- Application Lifecycle - Understanding sync vs async in applications