Skip to content

Use functools.wraps instead of decorator #766

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions .coveragerc
Original file line number Diff line number Diff line change
@@ -1,8 +1,6 @@
[run]
branch = True
source = prometheus_client
omit =
prometheus_client/decorator.py

[paths]
source =
Expand Down
2 changes: 0 additions & 2 deletions NOTICE
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
Prometheus instrumentation library for Python applications
Copyright 2015 The Prometheus Authors

This product bundles decorator 4.0.10 which is available under a "2-clause BSD"
license. For details, see prometheus_client/decorator.py.
5 changes: 1 addition & 4 deletions mypy.ini
Original file line number Diff line number Diff line change
@@ -1,7 +1,4 @@
[mypy]
exclude = prometheus_client/decorator.py|prometheus_client/twisted|tests/test_twisted.py
exclude = prometheus_client/twisted|tests/test_twisted.py
implicit_reexport = False
disallow_incomplete_defs = True

[mypy-prometheus_client.decorator]
follow_imports = skip
44 changes: 25 additions & 19 deletions prometheus_client/context_managers.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,21 @@
import functools
import sys
from timeit import default_timer
from types import TracebackType
from typing import (
Any, Callable, Literal, Optional, Tuple, Type, TYPE_CHECKING, TypeVar,
Union,
Callable, Literal, Optional, Tuple, Type, TYPE_CHECKING, TypeVar, Union,
)

from .decorator import decorate
if sys.version_info >= (3, 10):
from typing import ParamSpec
else:
from typing_extensions import ParamSpec

if TYPE_CHECKING:
from . import Counter
F = TypeVar("F", bound=Callable[..., Any])

TParam = ParamSpec("TParam")
TResult = TypeVar("TResult")


class ExceptionCounter:
Expand All @@ -25,12 +31,12 @@ def __exit__(self, typ: Optional[Type[BaseException]], value: Optional[BaseExcep
self._counter.inc()
return False

def __call__(self, f: "F") -> "F":
def wrapped(func, *args, **kwargs):
def __call__(self, f: Callable[TParam, TResult]) -> Callable[TParam, TResult]:
@functools.wraps(f)
def wrapped(*args: TParam.args, **kwargs: TParam.kwargs) -> TResult:
with self:
return func(*args, **kwargs)

return decorate(f, wrapped)
return f(*args, **kwargs)
return wrapped # type: ignore
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like there are two options for now: ignore or cast.

PS In python 3.10 it was resolved via https://docs.python.org/3/library/typing.html#typing.ParamSpec.



class InprogressTracker:
Expand All @@ -43,12 +49,12 @@ def __enter__(self):
def __exit__(self, typ, value, traceback):
self._gauge.dec()

def __call__(self, f: "F") -> "F":
def wrapped(func, *args, **kwargs):
def __call__(self, f: Callable[TParam, TResult]) -> Callable[TParam, TResult]:
@functools.wraps(f)
def wrapped(*args: TParam.args, **kwargs: TParam.kwargs) -> TResult:
with self:
return func(*args, **kwargs)

return decorate(f, wrapped)
return f(*args, **kwargs)
return wrapped # type: ignore


class Timer:
Expand All @@ -72,11 +78,11 @@ def __exit__(self, typ, value, traceback):
def labels(self, *args, **kw):
self._metric = self._metric.labels(*args, **kw)

def __call__(self, f: "F") -> "F":
def wrapped(func, *args, **kwargs):
def __call__(self, f: Callable[TParam, TResult]) -> Callable[TParam, TResult]:
@functools.wraps(f)
def wrapped(*args: TParam.args, **kwargs: TParam.kwargs) -> TResult:
# Obtaining new instance of timer every time
# ensures thread safety and reentrancy.
with self._new_timer():
return func(*args, **kwargs)

return decorate(f, wrapped)
return f(*args, **kwargs)
return wrapped # type: ignore
Loading