Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/func_inspect.py: 11%

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

191 statements  

1""" 

2My own variation on function-specific inspect-like features. 

3""" 

4 

5# Author: Gael Varoquaux <gael dot varoquaux at normalesup dot org> 

6# Copyright (c) 2009 Gael Varoquaux 

7# License: BSD Style, 3 clauses. 

8 

9import collections 

10import inspect 

11import os 

12import re 

13import warnings 

14from itertools import islice 

15from tokenize import open as open_py_source 

16 

17from .hashing import hash 

18from .logger import pformat 

19 

20full_argspec_fields = ( 

21 "args varargs varkw defaults kwonlyargs kwonlydefaults annotations" 

22) 

23full_argspec_type = collections.namedtuple("FullArgSpec", full_argspec_fields) 

24 

25 

26def get_func_code(func): 

27 """Attempts to retrieve a reliable function code hash. 

28 

29 The reason we don't use inspect.getsource is that it caches the 

30 source, whereas we want this to be modified on the fly when the 

31 function is modified. 

32 

33 Returns 

34 ------- 

35 func_code: string 

36 The function code 

37 source_file: string 

38 The path to the file in which the function is defined. 

39 first_line: int 

40 The first line of the code in the source file. 

41 

42 Notes 

43 ------ 

44 This function does a bit more magic than inspect, and is thus 

45 more robust. 

46 """ 

47 source_file = None 

48 try: 

49 code = func.__code__ 

50 source_file = code.co_filename 

51 if not os.path.exists(source_file): 

52 # Use inspect for lambda functions and functions defined in an 

53 # interactive shell, or in doctests 

54 source_code = "".join(inspect.getsourcelines(func)[0]) 

55 line_no = 1 

56 if source_file.startswith("<doctest "): 

57 source_file, line_no = re.match( 

58 r"\<doctest (.*\.rst)\[(.*)\]\>", source_file 

59 ).groups() 

60 line_no = int(line_no) 

61 source_file = "<doctest %s>" % source_file 

62 return source_code, source_file, line_no 

63 # Try to retrieve the source code. 

64 with open_py_source(source_file) as source_file_obj: 

65 first_line = code.co_firstlineno 

66 # All the lines after the function definition: 

67 source_lines = list(islice(source_file_obj, first_line - 1, None)) 

68 return "".join(inspect.getblock(source_lines)), source_file, first_line 

69 except: # noqa: E722 

70 # If the source code fails, we use the hash. This is fragile and 

71 # might change from one session to another. 

72 if hasattr(func, "__code__"): 

73 # Python 3.X 

74 return hash(func.__code__), source_file, -1 

75 else: 

76 # Weird objects like numpy ufunc don't have __code__ 

77 # This is fragile, as quite often the id of the object is 

78 # in the repr, so it might not persist across sessions, 

79 # however it will work for ufuncs. 

80 return repr(func), source_file, -1 

81 

82 

83def _clean_win_chars(string): 

84 """Windows cannot encode some characters in filename.""" 

85 import urllib 

86 

87 if hasattr(urllib, "quote"): 

88 quote = urllib.quote 

89 else: 

90 # In Python 3, quote is elsewhere 

91 import urllib.parse 

92 

93 quote = urllib.parse.quote 

94 for char in ("<", ">", "!", ":", "\\"): 

95 string = string.replace(char, quote(char)) 

96 return string 

97 

98 

99def get_func_name(func, resolv_alias=True, win_characters=True): 

100 """Return the function import path (as a list of module names), and 

101 a name for the function. 

102 

103 Parameters 

104 ---------- 

105 func: callable 

106 The func to inspect 

107 resolv_alias: boolean, optional 

108 If true, possible local aliases are indicated. 

109 win_characters: boolean, optional 

110 If true, substitute special characters using urllib.quote 

111 This is useful in Windows, as it cannot encode some filenames 

112 """ 

113 if hasattr(func, "__module__"): 

114 module = func.__module__ 

115 else: 

116 try: 

117 module = inspect.getmodule(func) 

118 except TypeError: 

119 if hasattr(func, "__class__"): 

120 module = func.__class__.__module__ 

121 else: 

122 module = "unknown" 

123 if module is None: 

124 # Happens in doctests, eg 

125 module = "" 

126 if module == "__main__": 

127 try: 

128 filename = os.path.abspath(inspect.getsourcefile(func)) 

129 except: # noqa: E722 

130 filename = None 

131 if filename is not None: 

132 # mangling of full path to filename 

133 parts = filename.split(os.sep) 

134 if parts[-1].startswith("<ipython-input"): 

135 # We're in a IPython (or notebook) session. parts[-1] comes 

136 # from func.__code__.co_filename and is of the form 

137 # <ipython-input-N-XYZ>, where: 

138 # - N is the cell number where the function was defined 

139 # - XYZ is a hash representing the function's code (and name). 

140 # It will be consistent across sessions and kernel restarts, 

141 # and will change if the function's code/name changes 

142 # We remove N so that cache is properly hit if the cell where 

143 # the func is defined is re-exectuted. 

144 # The XYZ hash should avoid collisions between functions with 

145 # the same name, both within the same notebook but also across 

146 # notebooks 

147 split = parts[-1].split("-") 

148 parts[-1] = "-".join(split[:2] + split[3:]) 

149 elif len(parts) > 2 and parts[-2].startswith("ipykernel_"): 

150 # In a notebook session (ipykernel). Filename seems to be 'xyz' 

151 # of above. parts[-2] has the structure ipykernel_XXXXXX where 

152 # XXXXXX is a six-digit number identifying the current run (?). 

153 # If we split it off, the function again has the same 

154 # identifier across runs. 

155 parts[-2] = "ipykernel" 

156 filename = "-".join(parts) 

157 if filename.endswith(".py"): 

158 filename = filename[:-3] 

159 module = module + "-" + filename 

160 module = module.split(".") 

161 if hasattr(func, "func_name"): 

162 name = func.func_name 

163 elif hasattr(func, "__name__"): 

164 name = func.__name__ 

165 else: 

166 name = "unknown" 

167 # Hack to detect functions not defined at the module-level 

168 if resolv_alias: 

169 # TODO: Maybe add a warning here? 

170 if hasattr(func, "func_globals") and name in func.func_globals: 

171 if func.func_globals[name] is not func: 

172 name = "%s-alias" % name 

173 if hasattr(func, "__qualname__") and func.__qualname__ != name: 

174 # Extend the module name in case of nested functions to avoid 

175 # (module, name) collisions 

176 module.extend(func.__qualname__.split(".")[:-1]) 

177 if inspect.ismethod(func): 

178 # We need to add the name of the class 

179 if hasattr(func, "im_class"): 

180 klass = func.im_class 

181 module.append(klass.__name__) 

182 if os.name == "nt" and win_characters: 

183 # Windows can't encode certain characters in filenames 

184 name = _clean_win_chars(name) 

185 module = [_clean_win_chars(s) for s in module] 

186 return module, name 

187 

188 

189def _signature_str(function_name, arg_sig): 

190 """Helper function to output a function signature""" 

191 return "{}{}".format(function_name, arg_sig) 

192 

193 

194def _function_called_str(function_name, args, kwargs): 

195 """Helper function to output a function call""" 

196 template_str = "{0}({1}, {2})" 

197 

198 args_str = repr(args)[1:-1] 

199 kwargs_str = ", ".join("%s=%s" % (k, v) for k, v in kwargs.items()) 

200 return template_str.format(function_name, args_str, kwargs_str) 

201 

202 

203def filter_args(func, ignore_lst, args=(), kwargs=dict()): 

204 """Filters the given args and kwargs using a list of arguments to 

205 ignore, and a function specification. 

206 

207 Parameters 

208 ---------- 

209 func: callable 

210 Function giving the argument specification 

211 ignore_lst: list of strings 

212 List of arguments to ignore (either a name of an argument 

213 in the function spec, or '*', or '**') 

214 *args: list 

215 Positional arguments passed to the function. 

216 **kwargs: dict 

217 Keyword arguments passed to the function 

218 

219 Returns 

220 ------- 

221 filtered_args: list 

222 List of filtered positional and keyword arguments. 

223 """ 

224 args = list(args) 

225 if isinstance(ignore_lst, str): 

226 # Catch a common mistake 

227 raise ValueError( 

228 "ignore_lst must be a list of parameters to ignore " 

229 f"{ignore_lst} (type {type(ignore_lst)}) was given" 

230 ) 

231 # Special case for functools.partial objects 

232 if not inspect.ismethod(func) and not inspect.isfunction(func): 

233 if ignore_lst: 

234 warnings.warn( 

235 f"Cannot inspect object {func}, ignore list will not work.", 

236 stacklevel=2, 

237 ) 

238 return {"*": args, "**": kwargs} 

239 arg_sig = inspect.signature(func) 

240 arg_names = [] 

241 arg_kwonlyargs = [] 

242 arg_varargs = None 

243 arg_varkw = None 

244 arg_dict = dict() 

245 n_positional_only = 0 

246 for param in arg_sig.parameters.values(): 

247 if param.kind is param.POSITIONAL_OR_KEYWORD: 

248 arg_names.append(param.name) 

249 elif param.kind is param.POSITIONAL_ONLY: 

250 arg_names.append(param.name) 

251 n_positional_only += 1 

252 elif param.kind is param.KEYWORD_ONLY: 

253 arg_kwonlyargs.append(param.name) 

254 elif param.kind is param.VAR_POSITIONAL: 

255 arg_varargs = param.name 

256 elif param.kind is param.VAR_KEYWORD: 

257 arg_varkw = param.name 

258 if param.default is not param.empty: 

259 arg_dict[param.name] = param.default 

260 if inspect.ismethod(func): 

261 # First argument is 'self', it has been removed by Python 

262 # we need to add it back: 

263 args = [ 

264 func.__self__, 

265 ] + args 

266 # func is an instance method, inspect.signature(func) does not 

267 # include self, we need to fetch it from the class method, i.e 

268 # func.__func__ 

269 class_method_sig = inspect.signature(func.__func__) 

270 self_name = next(iter(class_method_sig.parameters)) 

271 arg_names = [self_name] + arg_names 

272 n_positional_only += 1 

273 # XXX: Maybe I need an inspect.isbuiltin to detect C-level methods, such 

274 # as on ndarrays. 

275 

276 _, name = get_func_name(func, resolv_alias=False) 

277 

278 # Check for positional argument of the function, that can be 

279 # passed either as positional or keyword arguments in the call 

280 for arg_position, arg_name in enumerate(arg_names): 

281 if arg_position < len(args): 

282 # given as positional 

283 arg_dict[arg_name] = args[arg_position] 

284 if arg_position >= n_positional_only and arg_name in kwargs: 

285 raise ValueError( 

286 f"Argument {arg_name} was given both as positional and as keyword" 

287 f" for {_signature_str(name, arg_sig)}:\n" 

288 f" {_function_called_str(name, args, kwargs)} was called." 

289 ) 

290 elif arg_position >= n_positional_only and arg_name in kwargs: 

291 # given as keyword if not positional only 

292 arg_dict[arg_name] = kwargs[arg_name] 

293 elif arg_name not in arg_dict: 

294 # Missing argument 

295 raise ValueError( 

296 f"Wrong number of arguments for {_signature_str(name, arg_sig)}:\n" 

297 f" {_function_called_str(name, args, kwargs)} was called." 

298 ) 

299 

300 # If more positional arguments are given, they correspond 

301 # to *args, store them in vargs 

302 if len(args) > len(arg_names): 

303 if arg_varargs is None: 

304 raise ValueError( 

305 f"Too many arguments for {_signature_str(name, arg_sig)}:\n" 

306 f" {_function_called_str(name, args, kwargs)} was called." 

307 ) 

308 arg_dict["*"] = args[len(arg_names) :] 

309 elif arg_varargs is not None: 

310 arg_dict["*"] = [] 

311 

312 # Check keyword only arguments 

313 for arg_name in arg_kwonlyargs: 

314 if arg_name in kwargs: 

315 arg_dict[arg_name] = kwargs[arg_name] 

316 elif arg_name not in arg_dict: 

317 # required keyword only argument that is missing, 

318 # raise an error 

319 raise ValueError( 

320 f"Wrong number of arguments for {_signature_str(name, arg_sig)}:\n" 

321 f" {_function_called_str(name, args, kwargs)} was called." 

322 ) 

323 

324 # If more keyword arguments are given, store them in varkwargs 

325 named_args = {*arg_names[n_positional_only:], *arg_kwonlyargs} 

326 varkwargs = {k: v for k, v in kwargs.items() if k not in named_args} 

327 if arg_varkw is not None: 

328 arg_dict["**"] = varkwargs 

329 elif varkwargs: 

330 raise ValueError( 

331 f"Too many keyword arguments for {_signature_str(name, arg_sig)}:\n" 

332 f" {_function_called_str(name, args, kwargs)} was called." 

333 ) 

334 

335 # Now remove the arguments to be ignored 

336 for item in ignore_lst: 

337 if item in arg_dict: 

338 arg_dict.pop(item) 

339 else: 

340 raise ValueError( 

341 f"Ignore list: argument '{item}' is not defined for " 

342 f"function {_signature_str(name, arg_sig)}" 

343 ) 

344 # XXX: Return a sorted list of pairs? 

345 return arg_dict 

346 

347 

348def _format_arg(arg): 

349 formatted_arg = pformat(arg, indent=2) 

350 if len(formatted_arg) > 1500: 

351 formatted_arg = "%s..." % formatted_arg[:700] 

352 return formatted_arg 

353 

354 

355def format_signature(func, *args, **kwargs): 

356 # XXX: Should this use inspect.formatargvalues/formatargspec? 

357 module, name = get_func_name(func) 

358 module = [m for m in module if m] 

359 if module: 

360 module.append(name) 

361 module_path = ".".join(module) 

362 else: 

363 module_path = name 

364 arg_str = list() 

365 previous_length = 0 

366 for arg in args: 

367 formatted_arg = _format_arg(arg) 

368 if previous_length > 80: 

369 formatted_arg = "\n%s" % formatted_arg 

370 previous_length = len(formatted_arg) 

371 arg_str.append(formatted_arg) 

372 arg_str.extend(["%s=%s" % (v, _format_arg(i)) for v, i in kwargs.items()]) 

373 arg_str = ", ".join(arg_str) 

374 

375 signature = "%s(%s)" % (name, arg_str) 

376 return module_path, signature 

377 

378 

379def format_call(func, args, kwargs, object_name="Memory"): 

380 """Returns a nicely formatted statement displaying the function 

381 call with the given arguments. 

382 """ 

383 path, signature = format_signature(func, *args, **kwargs) 

384 msg = "%s\n[%s] Calling %s...\n%s" % (80 * "_", object_name, path, signature) 

385 return msg 

386 # XXX: Not using logging framework 

387 # self.debug(msg)