tessl/pypi-asgiref
ASGI specs, helper code, and adapters for bridging synchronous and asynchronous Python web applications
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
sync-async.mddocs/
Sync/Async Conversion
Core functionality for converting between synchronous and asynchronous functions, handling context propagation, thread management, and event loop integration automatically. These utilities enable seamless integration between sync and async codebases.
Capabilities
Function Decorators
Convert functions between sync and async execution contexts with automatic context variable propagation and proper thread/event loop management.
def sync_to_async(func=None, *, thread_sensitive=True, executor=None):
"""
Convert a synchronous function to an asynchronous one.
Parameters:
- func: callable, the synchronous function to convert
- thread_sensitive: bool, whether to run in thread-sensitive mode (default True)
- executor: Executor, custom executor for running sync code (optional)
Returns:
SyncToAsync wrapper that can be called with await
"""
def async_to_sync(awaitable=None, *, force_new_loop=False):
"""
Convert an asynchronous function to a synchronous one.
Parameters:
- awaitable: callable, the async function/coroutine to convert
- force_new_loop: bool, whether to force creation of new event loop (default False)
Returns:
AsyncToSync wrapper that can be called synchronously
"""Conversion Classes
Direct access to the underlying conversion classes for advanced use cases and custom integration.
class SyncToAsync:
"""Generic utility class to convert sync functions to async."""
def __init__(self, func, thread_sensitive=True, executor=None):
"""
Initialize sync-to-async converter.
Parameters:
- func: callable, synchronous function to wrap
- thread_sensitive: bool, run in thread-sensitive mode
- executor: Executor, custom executor instance
"""
def __call__(self, *args, **kwargs):
"""
Execute wrapped function asynchronously.
Returns:
Coroutine that will execute the sync function
"""
def __get__(self, parent, objtype):
"""Descriptor protocol support for method binding."""
class AsyncToSync:
"""Generic utility class to convert async functions to sync."""
def __init__(self, awaitable, force_new_loop=False):
"""
Initialize async-to-sync converter.
Parameters:
- awaitable: callable, async function to wrap
- force_new_loop: bool, force new event loop creation
"""
def __call__(self, *args, **kwargs):
"""
Execute wrapped function synchronously.
Returns:
Result of the async function executed synchronously
"""
def __get__(self, parent, objtype):
"""Descriptor protocol support for method binding."""Thread Context Management
Advanced context management for thread-sensitive operations and cross-context execution.
class ThreadSensitiveContext:
"""Async context manager for thread-sensitive mode."""
async def __aenter__(self):
"""Enter thread-sensitive async context."""
async def __aexit__(self, exc, value, tb):
"""Exit thread-sensitive async context."""Coroutine Function Utilities
Utilities for detecting and marking coroutine functions, particularly useful for framework integration.
def iscoroutinefunction(func):
"""
Check if something is a coroutine function.
Parameters:
- func: callable, function to check
Returns:
bool: True if func is a coroutine function
"""
def markcoroutinefunction(func):
"""
Decorator to mark functions as coroutine functions.
Parameters:
- func: callable, function to mark
Returns:
callable: The marked function
"""Usage Examples
Basic Function Conversion
from asgiref.sync import sync_to_async, async_to_sync
import time
import asyncio
# Convert sync function to async
@sync_to_async
def sync_operation():
time.sleep(1) # Simulate blocking operation
return "sync result"
# Convert async function to sync
@async_to_sync
async def async_operation():
await asyncio.sleep(1) # Simulate async operation
return "async result"
# Usage in async context
async def async_main():
result = await sync_operation()
print(result) # "sync result"
# Usage in sync context
def sync_main():
result = async_operation()
print(result) # "async result"Thread-Sensitive vs Thread-Insensitive
from asgiref.sync import sync_to_async
import threading
# Thread-sensitive (default) - runs in same thread
@sync_to_async(thread_sensitive=True)
def thread_sensitive_func():
return threading.current_thread().name
# Thread-insensitive - can run in any thread
@sync_to_async(thread_sensitive=False)
def thread_insensitive_func():
return "Can run anywhere"
async def demonstrate():
# Will run in specific thread context
result1 = await thread_sensitive_func()
# Can run in thread pool
result2 = await thread_insensitive_func()Custom Executor
from asgiref.sync import sync_to_async
from concurrent.futures import ThreadPoolExecutor
# Use custom executor
custom_executor = ThreadPoolExecutor(max_workers=2)
@sync_to_async(executor=custom_executor)
def cpu_intensive_task():
# CPU-bound work
return sum(i * i for i in range(1000000))
async def main():
result = await cpu_intensive_task()
print(f"Result: {result}")Method Conversion
from asgiref.sync import sync_to_async, async_to_sync
class DatabaseManager:
@sync_to_async
def get_user(self, user_id):
# Simulate database query
return {"id": user_id, "name": f"User {user_id}"}
@async_to_sync
async def async_get_user(self, user_id):
# Simulate async database query
await asyncio.sleep(0.1)
return {"id": user_id, "name": f"Async User {user_id}"}
# Usage
db = DatabaseManager()
async def async_usage():
user = await db.get_user(123)
print(user)
def sync_usage():
user = db.async_get_user(456)
print(user)Install with Tessl CLI
npx tessl i tessl/pypi-asgiref