File size: 2,971 Bytes
eccf061 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """
Async utilities for safe async/sync integration
"""
import asyncio
import functools
from typing import Any, Callable, Coroutine
import logging
logger = logging.getLogger(__name__)
class AsyncRunner:
"""Safely run async functions in sync context"""
@staticmethod
def run_async(coro: Coroutine) -> Any:
"""
Run async coroutine in sync context safely
Args:
coro: Async coroutine to run
Returns:
Result of the coroutine
"""
try:
# Try to get existing event loop
loop = asyncio.get_running_loop()
# We're already in async context - create task
logger.debug("Running in existing async context")
future = asyncio.create_task(coro)
return future
except RuntimeError:
# No running loop, create one
logger.debug("Creating new event loop for async execution")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
finally:
loop.close()
@staticmethod
def sync_wrapper(async_func: Callable) -> Callable:
"""
Decorator to make async function callable from sync context
Args:
async_func: Async function to wrap
Returns:
Sync-compatible function
"""
@functools.wraps(async_func)
def wrapper(*args, **kwargs) -> Any:
return AsyncRunner.run_async(async_func(*args, **kwargs))
return wrapper
def async_to_sync(async_func: Callable) -> Callable:
"""
Convenience decorator to convert async function to sync
Usage:
@async_to_sync
async def my_async_function():
...
# Can now be called synchronously
result = my_async_function()
"""
return AsyncRunner.sync_wrapper(async_func)
class SafeEventLoop:
"""Context manager for safe event loop handling"""
def __init__(self, create_new: bool = False):
self.create_new = create_new
self.loop = None
self.original_loop = None
def __enter__(self):
if self.create_new:
self.original_loop = asyncio.get_event_loop_policy().get_event_loop()
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
else:
try:
self.loop = asyncio.get_running_loop()
except RuntimeError:
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
return self.loop
def __exit__(self, exc_type, exc_val, exc_tb):
if self.create_new and self.original_loop:
asyncio.set_event_loop(self.original_loop)
if not self.loop.is_closed():
self.loop.close() |