Repository navigation
Expand file tree
/
Copy pathdecorators.py
More file actions
91 lines (68 loc) · 2.51 KB
/
Copy pathdecorators.py
File metadata and controls
91 lines (68 loc) · 2.51 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import inspect
import logging
import time
from functools import wraps
from typing import Any, List
from dispatch.metrics import provider as metrics_provider
from .extensions import sentry_sdk, configure_extensions
from .database import SessionLocal
log = logging.getLogger(__name__)
def fullname(o):
module = inspect.getmodule(o)
return f"{module.__name__}.{o.__qualname__}"
def background_task(func):
"""Decorator that sets up the a background task function
with a database session and exception tracking.
As background tasks run in their own threads, it does not attempt
to propagate errors.
"""
@wraps(func)
def wrapper(*args, **kwargs):
db_session = SessionLocal()
kwargs["db_session"] = db_session
try:
metrics_provider.counter(f"function.call.counter", tags={"function": fullname(func)})
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed_time = time.perf_counter() - start
metrics_provider.timer(
f"function.elapsed.time", value=elapsed_time, tags={"function": fullname(func)}
)
return result
except Exception as e:
import traceback
configure_extensions()
print(traceback.format_exc())
sentry_sdk.capture_exception(e)
finally:
db_session.close()
return wrapper
def timer(func: Any):
"""Timing decorator that sends a timing metric."""
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed_time = time.perf_counter() - start
metrics_provider.timer(
f"function.elapsed.time", value=elapsed_time, tags={"function": fullname(func)}
)
return result
return wrapper
def counter(func: Any):
"""Counting decorator that sends a counting metric."""
@wraps(func)
def wrapper(*args, **kwargs):
metrics_provider.counter(f"function.call.counter", tags={"function": fullname(func)})
return func(*args, **kwargs)
return wrapper
def apply(decorator: Any, exclude: List[str] = None):
"""Class decorator that applies specified decorator to all class methods."""
if not exclude:
exclude = []
def decorate(cls):
for attr in cls.__dict__:
if callable(getattr(cls, attr)) and attr not in exclude:
setattr(cls, attr, decorator(getattr(cls, attr)))
return cls
return decorate