livekit--agents
wehub-resource-sync
43 行
1.2 KiB
Python
43 行
1.2 KiB
Python
import functools
|
|
import inspect
|
|
import logging
|
|
from collections.abc import Callable
|
|
from typing import Any, TypeVar, cast
|
|
|
|
F = TypeVar("F", bound=Callable[..., Any])
|
|
|
|
|
|
def log_exceptions(msg: str = "", logger: logging.Logger = logging.getLogger()) -> Callable[[F], F]: # noqa: B008
|
|
def deco(fn: F) -> F:
|
|
if inspect.iscoroutinefunction(fn):
|
|
|
|
@functools.wraps(fn)
|
|
async def async_fn_logs(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return await fn(*args, **kwargs)
|
|
except Exception:
|
|
err = f"Error in {fn.__name__}"
|
|
if msg:
|
|
err += f" – {msg}"
|
|
logger.exception(err)
|
|
raise
|
|
|
|
return cast(F, async_fn_logs)
|
|
|
|
else:
|
|
|
|
@functools.wraps(fn)
|
|
def fn_logs(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return fn(*args, **kwargs)
|
|
except Exception:
|
|
err = f"Error in {fn.__name__}"
|
|
if msg:
|
|
err += f" – {msg}"
|
|
logger.exception(err)
|
|
raise
|
|
|
|
return cast(F, fn_logs)
|
|
|
|
return deco
|