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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
import logging
import traceback
import functools
import os
import io
from contextlib import contextmanager
depreclog = logging.getLogger("tomoscan.DEPRECATION")
deprecache = set([])
def deprecated_warning(
type_,
name,
reason=None,
replacement=None,
since_version=None,
only_once=True,
skip_backtrace_count=0,
):
"""
Function to log a deprecation warning
:param type_: Nature of the object to be deprecated:
"Module", "Function", "Class" ...
:param name: Object name.
:param reason: Reason for deprecating this function
(e.g. "feature no longer provided",
:param replacement: Name of replacement function (if the reason for
deprecating was to rename the function)
:param since_version: First *silx* version for which the function was
deprecated (e.g. "0.5.0").
:param only_once: If true, the deprecation warning will only be
generated one time for each different call locations. Default is true.
:param skip_backtrace_count: Amount of last backtrace to ignore when
logging the backtrace
"""
if not depreclog.isEnabledFor(logging.WARNING):
# Avoid computation when it is not logged
return
msg = "%s %s is deprecated"
if since_version is not None:
msg += " since tomoscan version %s" % since_version
msg += "."
if reason is not None:
msg += " Reason: %s." % reason
if replacement is not None:
msg += " Use '%s' instead." % replacement
msg += "\n%s"
limit = 2 + skip_backtrace_count
backtrace = "".join(traceback.format_stack(limit=limit)[0])
backtrace = backtrace.rstrip()
if only_once:
data = (msg, type_, name, backtrace)
if data in deprecache:
return
else:
deprecache.add(data)
depreclog.warning(msg, type_, name, backtrace)
def deprecated(
func=None,
reason=None,
replacement=None,
since_version=None,
only_once=True,
skip_backtrace_count=1,
):
"""
Decorator that deprecates the use of a function
:param reason: Reason for deprecating this function
(e.g. "feature no longer provided",
:param replacement: Name of replacement function (if the reason for
deprecating was to rename the function)
:param since_version: First *silx* version for which the function was
deprecated (e.g. "0.5.0").
:param only_once: If true, the deprecation warning will only be
generated one time. Default is true.
:param skip_backtrace_count: Amount of last backtrace to ignore when
logging the backtrace
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
deprecated_warning(
type_="Function",
name=func.__name__,
reason=reason,
replacement=replacement,
since_version=since_version,
only_once=only_once,
skip_backtrace_count=skip_backtrace_count,
)
return func(*args, **kwargs)
return wrapper
if func is not None:
return decorator(func)
return decorator
def filter_esrf_mounting_points(path: str):
"""
filter path like '/mnt/multipath-shares' or '/gpfs/easy'
"""
parts = path.split(os.sep)
if len(parts) > 4:
if parts[0] == "" and parts[1] in ("gpfs", "mnt") and parts[3] == "data":
new_parts = list(parts[0:1]) + list(parts[3:])
path = os.sep.join(new_parts)
return path
def _deprecate_url_in_signature(url, name="url"):
if url is None:
return
deprecated_warning(
type_="parameter", name=name, reason="about to be removed", since_version="2.3"
)
@contextmanager
def catch_log_messages(logger_name=depreclog.name, level=logging.WARNING):
# Create a buffer to capture log messages
log_capture = io.StringIO()
# Get the logger and add a handler to capture messages
logger = logging.getLogger(logger_name)
handler = logging.StreamHandler(log_capture)
handler.setLevel(level)
logger.addHandler(handler)
try:
yield log_capture
finally:
# Clean up: remove the handler
logger.removeHandler(handler)
|