Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/airflow/utils/code_utils.py: 21%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17from __future__ import annotations 

18 

19import functools 

20import inspect 

21import os 

22from pathlib import Path 

23from typing import Any 

24 

25 

26def get_python_source(x: Any) -> str | None: 

27 """Get Python source (or not), preventing exceptions.""" 

28 if isinstance(x, str): 

29 return x 

30 

31 if x is None: 

32 return None 

33 

34 source_code = None 

35 

36 if isinstance(x, functools.partial): 

37 source_code = inspect.getsource(x.func) 

38 

39 if source_code is None: 

40 try: 

41 source_code = inspect.getsource(x) 

42 except TypeError: 

43 pass 

44 

45 if source_code is None: 

46 try: 

47 source_code = inspect.getsource(x.__call__) 

48 except (TypeError, AttributeError): 

49 pass 

50 

51 if source_code is None: 

52 source_code = f"No source code available for {type(x)}" 

53 return source_code 

54 

55 

56def prepare_code_snippet(file_path: str, line_no: int, context_lines_count: int = 5) -> str: 

57 """ 

58 Prepare code snippet with line numbers and a specific line marked. 

59 

60 :param file_path: File name 

61 :param line_no: Line number 

62 :param context_lines_count: The number of lines that will be cut before and after. 

63 :return: str 

64 """ 

65 code_lines = Path(file_path).read_text().splitlines() 

66 # Prepend line number 

67 code_lines = [ 

68 f">{lno:3} | {line}" if line_no == lno else f"{lno:4} | {line}" 

69 for lno, line in enumerate(code_lines, 1) 

70 ] 

71 # # Cut out the snippet 

72 start_line_no = max(0, line_no - context_lines_count - 1) 

73 end_line_no = line_no + context_lines_count 

74 code_lines = code_lines[start_line_no:end_line_no] 

75 # Join lines 

76 code = "\n".join(code_lines) 

77 return code 

78 

79 

80def get_terminal_formatter(**opts): 

81 """Return the best formatter available in the current terminal.""" 

82 if "256" in os.environ.get("TERM", ""): 

83 from pygments.formatters.terminal256 import Terminal256Formatter 

84 

85 formatter = Terminal256Formatter(**opts) 

86 else: 

87 from pygments.formatters.terminal import TerminalFormatter 

88 

89 formatter = TerminalFormatter(**opts) 

90 return formatter