Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/func_inspect.py: 14%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2My own variation on function-specific inspect-like features.
3"""
5# Author: Gael Varoquaux <gael dot varoquaux at normalesup dot org>
6# Copyright (c) 2009 Gael Varoquaux
7# License: BSD Style, 3 clauses.
9import collections
10import hashlib
11import inspect
12import os
13import re
14import types
15import warnings
16from itertools import islice
17from tokenize import open as open_py_source
19from .logger import pformat
21full_argspec_fields = (
22 "args varargs varkw defaults kwonlyargs kwonlydefaults annotations"
23)
24full_argspec_type = collections.namedtuple("FullArgSpec", full_argspec_fields)
27def _code_fingerprint(code):
28 """Digest the contents of a code object.
30 ``hash(code)`` would be simpler, but it hashes the strings it is built from
31 and those are salted by PYTHONHASHSEED. This value is written to
32 func_code.py and compared by *other* processes, which then conclude the
33 function changed and wipe its cache directory (#1694).
35 The fields below mirror CPython's own ``code_hash()``:
36 https://github.com/python/cpython/blob/5918085bb6f4a3a48193cacb9bb99b044d4e0452/Objects/codeobject.c#L2608-L2624
37 co_localsplusnames is spelled out as varnames/cellvars/freevars. Only the
38 position information, co_firstlineno and co_linetable, is deliberately left
39 out: joblib compares source *text* and tracks the line number separately,
40 so moving a function without editing it must not invalidate its cache. For
41 the same reason co_filename is not included, unlike ``marshal.dumps``,
42 which would key a notebook function to the cell number it last ran in.
43 """
44 parts = [
45 code.co_name,
46 # co_code de-specializes adaptive bytecode, like _Py_GetBaseCodeUnit
47 code.co_code.hex(),
48 # 3.11+ moved the try/except ranges out of the bytecode and in here
49 getattr(code, "co_exceptiontable", b"").hex(),
50 *code.co_names,
51 *code.co_varnames,
52 *code.co_cellvars,
53 *code.co_freevars,
54 *map(
55 repr,
56 (
57 code.co_argcount,
58 code.co_posonlyargcount,
59 code.co_kwonlyargcount,
60 code.co_flags,
61 ),
62 ),
63 ]
64 for const in code.co_consts:
65 # repr() of a nested code object embeds its address
66 if isinstance(const, types.CodeType):
67 parts.append(_code_fingerprint(const))
68 else:
69 parts.append(repr(const))
70 digest = hashlib.new("md5", usedforsecurity=False)
71 digest.update("\n".join(parts).encode("utf-8", errors="replace"))
72 return digest.hexdigest()
75def get_func_code(func):
76 """Attempts to retrieve a reliable function code hash.
78 The reason we don't use inspect.getsource is that it caches the
79 source, whereas we want this to be modified on the fly when the
80 function is modified.
82 Returns
83 -------
84 func_code: string
85 The function code
86 source_file: string
87 The path to the file in which the function is defined.
88 first_line: int
89 The first line of the code in the source file.
91 Notes
92 ------
93 This function does a bit more magic than inspect, and is thus
94 more robust.
95 """
96 source_file = None
97 try:
98 code = func.__code__
99 source_file = code.co_filename
100 if not os.path.exists(source_file):
101 # Use inspect for lambda functions and functions defined in an
102 # interactive shell, or in doctests
103 source_code = "".join(inspect.getsourcelines(func)[0])
104 line_no = 1
105 if source_file.startswith("<doctest "):
106 source_file, line_no = re.match(
107 r"\<doctest (.*\.rst)\[(.*)\]\>", source_file
108 ).groups()
109 line_no = int(line_no)
110 source_file = "<doctest %s>" % source_file
111 return source_code, source_file, line_no
112 # Try to retrieve the source code.
113 with open_py_source(source_file) as source_file_obj:
114 first_line = code.co_firstlineno
115 # All the lines after the function definition:
116 source_lines = list(islice(source_file_obj, first_line - 1, None))
117 return "".join(inspect.getblock(source_lines)), source_file, first_line
118 except: # noqa: E722
119 # If the source code fails, we use the hash. This is fragile and
120 # might change from one session to another.
121 if hasattr(func, "__code__"):
122 # Python 3.X
123 return _code_fingerprint(func.__code__), source_file, -1
124 else:
125 # Weird objects like numpy ufunc don't have __code__
126 # This is fragile, as quite often the id of the object is
127 # in the repr, so it might not persist across sessions,
128 # however it will work for ufuncs.
129 return repr(func), source_file, -1
132def _clean_win_chars(string):
133 """Windows cannot encode some characters in filename."""
134 import urllib
136 if hasattr(urllib, "quote"):
137 quote = urllib.quote
138 else:
139 # In Python 3, quote is elsewhere
140 import urllib.parse
142 quote = urllib.parse.quote
143 for char in ("<", ">", "!", ":", "\\"):
144 string = string.replace(char, quote(char))
145 return string
148def get_func_name(func, resolv_alias=True, win_characters=True):
149 """Return the function import path (as a list of module names), and
150 a name for the function.
152 Parameters
153 ----------
154 func: callable
155 The func to inspect
156 resolv_alias: boolean, optional
157 If true, possible local aliases are indicated.
158 win_characters: boolean, optional
159 If true, substitute special characters using urllib.quote
160 This is useful in Windows, as it cannot encode some filenames
161 """
162 if hasattr(func, "__module__"):
163 module = func.__module__
164 else:
165 try:
166 module = inspect.getmodule(func)
167 except TypeError:
168 if hasattr(func, "__class__"):
169 module = func.__class__.__module__
170 else:
171 module = "unknown"
172 if module is None:
173 # Happens in doctests, eg
174 module = ""
175 if module == "__main__":
176 try:
177 filename = os.path.abspath(inspect.getsourcefile(func))
178 except: # noqa: E722
179 filename = None
180 if filename is not None:
181 # mangling of full path to filename
182 parts = filename.split(os.sep)
183 if parts[-1].startswith("<ipython-input"):
184 # We're in a IPython (or notebook) session. parts[-1] comes
185 # from func.__code__.co_filename and is of the form
186 # <ipython-input-N-XYZ>, where:
187 # - N is the cell number where the function was defined
188 # - XYZ is a hash representing the function's code (and name).
189 # It will be consistent across sessions and kernel restarts,
190 # and will change if the function's code/name changes
191 # We remove N so that cache is properly hit if the cell where
192 # the func is defined is re-exectuted.
193 # The XYZ hash should avoid collisions between functions with
194 # the same name, both within the same notebook but also across
195 # notebooks
196 split = parts[-1].split("-")
197 parts[-1] = "-".join(split[:2] + split[3:])
198 elif len(parts) > 2 and parts[-2].startswith("ipykernel_"):
199 # In a notebook session (ipykernel). Filename seems to be 'xyz'
200 # of above. parts[-2] has the structure ipykernel_XXXXXX where
201 # XXXXXX is a six-digit number identifying the current run (?).
202 # If we split it off, the function again has the same
203 # identifier across runs.
204 parts[-2] = "ipykernel"
205 filename = "-".join(parts)
206 if filename.endswith(".py"):
207 filename = filename[:-3]
208 module = module + "-" + filename
209 module = module.split(".")
210 if hasattr(func, "func_name"):
211 name = func.func_name
212 elif hasattr(func, "__name__"):
213 name = func.__name__
214 else:
215 name = "unknown"
216 # Hack to detect functions not defined at the module-level
217 if resolv_alias:
218 # TODO: Maybe add a warning here?
219 if hasattr(func, "func_globals") and name in func.func_globals:
220 if func.func_globals[name] is not func:
221 name = "%s-alias" % name
222 if hasattr(func, "__qualname__") and func.__qualname__ != name:
223 # Extend the module name in case of nested functions to avoid
224 # (module, name) collisions
225 module.extend(func.__qualname__.split(".")[:-1])
226 if inspect.ismethod(func):
227 # We need to add the name of the class
228 if hasattr(func, "im_class"):
229 klass = func.im_class
230 module.append(klass.__name__)
231 if os.name == "nt" and win_characters:
232 # Windows can't encode certain characters in filenames
233 name = _clean_win_chars(name)
234 module = [_clean_win_chars(s) for s in module]
235 return module, name
238def _signature_str(function_name, arg_sig):
239 """Helper function to output a function signature"""
240 return "{}{}".format(function_name, arg_sig)
243def _function_called_str(function_name, args, kwargs):
244 """Helper function to output a function call"""
245 template_str = "{0}({1}, {2})"
247 args_str = repr(args)[1:-1]
248 kwargs_str = ", ".join("%s=%s" % (k, v) for k, v in kwargs.items())
249 return template_str.format(function_name, args_str, kwargs_str)
252def filter_args(func, ignore_lst, args=(), kwargs=dict()):
253 """Filters the given args and kwargs using a list of arguments to
254 ignore, and a function specification.
256 Parameters
257 ----------
258 func: callable
259 Function giving the argument specification
260 ignore_lst: list of strings
261 List of arguments to ignore (either a name of an argument
262 in the function spec, or '*', or '**')
263 *args: list
264 Positional arguments passed to the function.
265 **kwargs: dict
266 Keyword arguments passed to the function
268 Returns
269 -------
270 filtered_args: list
271 List of filtered positional and keyword arguments.
272 """
273 args = list(args)
274 if isinstance(ignore_lst, str):
275 # Catch a common mistake
276 raise ValueError(
277 "ignore_lst must be a list of parameters to ignore "
278 f"{ignore_lst} (type {type(ignore_lst)}) was given"
279 )
280 # Special case for functools.partial objects
281 if not inspect.ismethod(func) and not inspect.isfunction(func):
282 if ignore_lst:
283 warnings.warn(
284 f"Cannot inspect object {func}, ignore list will not work.",
285 stacklevel=2,
286 )
287 return {"*": args, "**": kwargs}
288 arg_sig = inspect.signature(func)
289 arg_names = []
290 arg_kwonlyargs = []
291 arg_varargs = None
292 arg_varkw = None
293 arg_dict = dict()
294 for param in arg_sig.parameters.values():
295 if param.kind is param.POSITIONAL_OR_KEYWORD:
296 arg_names.append(param.name)
297 elif param.kind is param.KEYWORD_ONLY:
298 arg_kwonlyargs.append(param.name)
299 elif param.kind is param.VAR_POSITIONAL:
300 arg_varargs = param.name
301 elif param.kind is param.VAR_KEYWORD:
302 arg_varkw = param.name
303 if param.default is not param.empty:
304 arg_dict[param.name] = param.default
305 if inspect.ismethod(func):
306 # First argument is 'self', it has been removed by Python
307 # we need to add it back:
308 args = [
309 func.__self__,
310 ] + args
311 # func is an instance method, inspect.signature(func) does not
312 # include self, we need to fetch it from the class method, i.e
313 # func.__func__
314 class_method_sig = inspect.signature(func.__func__)
315 self_name = next(iter(class_method_sig.parameters))
316 arg_names = [self_name] + arg_names
317 # XXX: Maybe I need an inspect.isbuiltin to detect C-level methods, such
318 # as on ndarrays.
320 _, name = get_func_name(func, resolv_alias=False)
322 # Check for positional argument of the function, that can be
323 # passed either as positional or keyword arguments in the call
324 for arg_position, arg_name in enumerate(arg_names):
325 if arg_position < len(args):
326 # given as positional
327 arg_dict[arg_name] = args[arg_position]
328 if arg_name in kwargs:
329 raise ValueError(
330 f"Argument {arg_name} was given both as positional and as keyword"
331 f" for {_signature_str(name, arg_sig)}:\n"
332 f" {_function_called_str(name, args, kwargs)} was called."
333 )
334 elif arg_name in kwargs:
335 # given as keyword
336 arg_dict[arg_name] = kwargs[arg_name]
337 elif arg_name not in arg_dict:
338 # Missing argument
339 raise ValueError(
340 f"Wrong number of arguments for {_signature_str(name, arg_sig)}:\n"
341 f" {_function_called_str(name, args, kwargs)} was called."
342 )
344 # If more positional arguments are given, they correspond
345 # to *args, store them in vargs
346 if len(args) > len(arg_names):
347 if arg_varargs is None:
348 raise ValueError(
349 f"Too many arguments for {_signature_str(name, arg_sig)}:\n"
350 f" {_function_called_str(name, args, kwargs)} was called."
351 )
352 arg_dict["*"] = args[len(arg_names) :]
353 elif arg_varargs is not None:
354 arg_dict["*"] = []
356 # Check keyword only arguments
357 for arg_name in arg_kwonlyargs:
358 if arg_name in kwargs:
359 arg_dict[arg_name] = kwargs[arg_name]
360 elif arg_name not in arg_dict:
361 # required keyword only argument that is missing,
362 # raise an error
363 raise ValueError(
364 f"Wrong number of arguments for {_signature_str(name, arg_sig)}:\n"
365 f" {_function_called_str(name, args, kwargs)} was called."
366 )
368 # If more keyword arguments are given, store them in varkwargs
369 varkwargs = {k: v for k, v in kwargs.items() if k not in arg_dict}
370 if arg_varkw is not None:
371 arg_dict["**"] = varkwargs
373 elif varkwargs:
374 raise ValueError(
375 f"Too many keyword arguments for {_signature_str(name, arg_sig)}:\n"
376 f" {_function_called_str(name, args, kwargs)} was called."
377 )
379 # Now remove the arguments to be ignored
380 for item in ignore_lst:
381 if item in arg_dict:
382 arg_dict.pop(item)
383 else:
384 raise ValueError(
385 f"Ignore list: argument '{item}' is not defined for "
386 f"function {_signature_str(name, arg_sig)}"
387 )
388 # XXX: Return a sorted list of pairs?
389 return arg_dict
392def _format_arg(arg):
393 formatted_arg = pformat(arg, indent=2)
394 if len(formatted_arg) > 1500:
395 formatted_arg = "%s..." % formatted_arg[:700]
396 return formatted_arg
399def format_signature(func, *args, **kwargs):
400 # XXX: Should this use inspect.formatargvalues/formatargspec?
401 module, name = get_func_name(func)
402 module = [m for m in module if m]
403 if module:
404 module.append(name)
405 module_path = ".".join(module)
406 else:
407 module_path = name
408 arg_str = list()
409 previous_length = 0
410 for arg in args:
411 formatted_arg = _format_arg(arg)
412 if previous_length > 80:
413 formatted_arg = "\n%s" % formatted_arg
414 previous_length = len(formatted_arg)
415 arg_str.append(formatted_arg)
416 arg_str.extend(["%s=%s" % (v, _format_arg(i)) for v, i in kwargs.items()])
417 arg_str = ", ".join(arg_str)
419 signature = "%s(%s)" % (name, arg_str)
420 return module_path, signature
423def format_call(func, args, kwargs, object_name="Memory"):
424 """Returns a nicely formatted statement displaying the function
425 call with the given arguments.
426 """
427 path, signature = format_signature(func, *args, **kwargs)
428 msg = "%s\n[%s] Calling %s...\n%s" % (80 * "_", object_name, path, signature)
429 return msg
430 # XXX: Not using logging framework
431 # self.debug(msg)