Skip to content

Commit

Permalink
fix(crons): Fix type hints for monitor decorator (#2944)
Browse files Browse the repository at this point in the history
Fixes GH-2939
  • Loading branch information
szokeasaurusrex authored Apr 8, 2024
1 parent 669ed17 commit 4729d53
Showing 1 changed file with 44 additions and 13 deletions.
57 changes: 44 additions & 13 deletions sentry_sdk/crons/_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,11 @@

if TYPE_CHECKING:
from typing import (
Any,
Awaitable,
Callable,
cast,
overload,
ParamSpec,
TypeVar,
Union,
Expand All @@ -17,22 +20,50 @@


class MonitorMixin:
def __call__(self, fn):
# type: (Callable[P, R]) -> Callable[P, Union[R, Awaitable[R]]]
if iscoroutinefunction(fn):
if TYPE_CHECKING:

@overload
def __call__(self, fn):
# type: (Callable[P, Awaitable[Any]]) -> Callable[P, Awaitable[Any]]
# Unfortunately, mypy does not give us any reliable way to type check the
# return value of an Awaitable (i.e. async function) for this overload,
# since calling iscouroutinefunction narrows the type to Callable[P, Awaitable[Any]].
...

@wraps(fn)
async def inner(*args: "P.args", **kwargs: "P.kwargs"):
# type: (...) -> R
with self: # type: ignore[attr-defined]
return await fn(*args, **kwargs)
@overload
def __call__(self, fn):
# type: (Callable[P, R]) -> Callable[P, R]
...

def __call__(
self,
fn, # type: Union[Callable[P, R], Callable[P, Awaitable[Any]]]
):
# type: (...) -> Union[Callable[P, R], Callable[P, Awaitable[Any]]]
if iscoroutinefunction(fn):
return self._async_wrapper(fn)

else:
if TYPE_CHECKING:
fn = cast("Callable[P, R]", fn)
return self._sync_wrapper(fn)

def _async_wrapper(self, fn):
# type: (Callable[P, Awaitable[Any]]) -> Callable[P, Awaitable[Any]]
@wraps(fn)
async def inner(*args: "P.args", **kwargs: "P.kwargs"):
# type: (...) -> R
with self: # type: ignore[attr-defined]
return await fn(*args, **kwargs)

return inner

@wraps(fn)
def inner(*args: "P.args", **kwargs: "P.kwargs"):
# type: (...) -> R
with self: # type: ignore[attr-defined]
return fn(*args, **kwargs)
def _sync_wrapper(self, fn):
# type: (Callable[P, R]) -> Callable[P, R]
@wraps(fn)
def inner(*args: "P.args", **kwargs: "P.kwargs"):
# type: (...) -> R
with self: # type: ignore[attr-defined]
return fn(*args, **kwargs)

return inner

0 comments on commit 4729d53

Please sign in to comment.