|
3 | 3 | from collections.abc import Callable |
4 | 4 | from datetime import datetime |
5 | 5 | from functools import wraps |
6 | | -from typing import Final, ParamSpec, TypeVar |
| 6 | +from typing import Any, Final, ParamSpec, TypeVar |
7 | 7 |
|
8 | 8 | from pydantic import BaseModel, Field, NonNegativeFloat, PrivateAttr |
9 | 9 |
|
@@ -73,28 +73,31 @@ def else_reset(self) -> None: |
73 | 73 | P = ParamSpec("P") |
74 | 74 | R = TypeVar("R") |
75 | 75 |
|
| 76 | +F = TypeVar("F", bound=Callable[..., Any]) |
76 | 77 |
|
77 | | -def silence_exceptions( |
78 | | - exceptions: tuple[type[BaseException], ...] |
79 | | -) -> Callable[[Callable[P, R]], Callable[P, R]]: |
80 | | - def decorator(func: Callable[..., R]) -> Callable[..., R]: |
81 | | - @wraps(func) |
82 | | - def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> R | None: |
83 | | - try: |
84 | | - return func(*args, **kwargs) |
85 | | - except exceptions: |
86 | | - return None |
87 | 78 |
|
88 | | - @wraps(func) |
89 | | - async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R | None: |
| 79 | +def silence_exceptions(exceptions: tuple[type[BaseException], ...]) -> Callable[[F], F]: |
| 80 | + def _decorator(func_or_coro: F) -> F: |
| 81 | + |
| 82 | + if inspect.iscoroutinefunction(func_or_coro): |
| 83 | + |
| 84 | + @wraps(func_or_coro) |
| 85 | + async def _async_wrapper(*args, **kwargs) -> Any: |
| 86 | + try: |
| 87 | + assert inspect.iscoroutinefunction(func_or_coro) # nosec |
| 88 | + return await func_or_coro(*args, **kwargs) |
| 89 | + except exceptions: |
| 90 | + return None |
| 91 | + |
| 92 | + return _async_wrapper # type: ignore[return-value] # decorators typing is hard |
| 93 | + |
| 94 | + @wraps(func_or_coro) |
| 95 | + def _sync_wrapper(*args, **kwargs) -> Any: |
90 | 96 | try: |
91 | | - assert inspect.iscoroutinefunction(func) # nosec |
92 | | - return await func(*args, **kwargs) |
| 97 | + return func_or_coro(*args, **kwargs) |
93 | 98 | except exceptions: |
94 | 99 | return None |
95 | 100 |
|
96 | | - if inspect.iscoroutinefunction(func): |
97 | | - return async_wrapper # type: ignore |
98 | | - return sync_wrapper # type: ignore |
| 101 | + return _sync_wrapper # type: ignore[return-value] # decorators typing is hard |
99 | 102 |
|
100 | | - return decorator |
| 103 | + return _decorator |
0 commit comments