mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zeekay
95d8461196
- Update quality-gate.yml to only lint hanzo_mcp and tests directories
- Add ruff config to pyproject.toml:
- Exclude vscode-extension and node_modules
- Ignore lint rules that conflict with test patterns:
B023 (loop variable binding), B904 (raise from),
E741 (ambiguous vars), F401/F821/F841 (unused/undefined in tests)
- Apply per-file ignores for tests/ and __init__.py
- Apply ruff fixes for import sorting and format
79 lines
2.0 KiB
Python
79 lines
2.0 KiB
Python
"""Event loop configuration with optional uvloop support.
|
|
|
|
This module provides utilities for configuring the asyncio event loop
|
|
with optional uvloop support for improved performance on Linux/macOS.
|
|
|
|
Uses hanzo_async for unified async I/O configuration.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from typing import Optional
|
|
|
|
from hanzo_async import configure_loop, using_uvloop
|
|
|
|
|
|
def configure_event_loop(*, quiet: bool = False) -> Optional[str]:
|
|
"""Configure the event loop with uvloop if available.
|
|
|
|
This should be called early in the application startup, before
|
|
any async code runs.
|
|
|
|
Args:
|
|
quiet: If True, suppress info messages about uvloop status
|
|
|
|
Returns:
|
|
The name of the event loop policy being used, or None if default
|
|
"""
|
|
# Use hanzo_async for unified configuration
|
|
if configure_loop():
|
|
if not quiet:
|
|
import logging
|
|
|
|
try:
|
|
import uvloop
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.debug(f"Using uvloop {uvloop.__version__} for event loop")
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import uvloop
|
|
|
|
return f"uvloop-{uvloop.__version__}"
|
|
except ImportError:
|
|
return None
|
|
|
|
return None
|
|
|
|
|
|
def get_event_loop_info() -> dict:
|
|
"""Get information about the current event loop configuration.
|
|
|
|
Returns:
|
|
Dictionary with event loop details
|
|
"""
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
except RuntimeError:
|
|
loop = None
|
|
|
|
info = {
|
|
"loop_class": type(loop).__name__ if loop else None,
|
|
"loop_module": type(loop).__module__ if loop else None,
|
|
"platform": sys.platform,
|
|
"using_uvloop": using_uvloop(),
|
|
}
|
|
|
|
# Check if uvloop is available
|
|
try:
|
|
import uvloop
|
|
|
|
info["uvloop_available"] = True
|
|
info["uvloop_version"] = uvloop.__version__
|
|
except ImportError:
|
|
info["uvloop_available"] = False
|
|
|
|
return info
|