Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/utils/code_utils.py: 21%
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
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# 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
19import functools
20import inspect
21import os
22from pathlib import Path
23from typing import Any
26def get_python_source(x: Any) -> str | None:
27 """Get Python source (or not), preventing exceptions."""
28 if isinstance(x, str):
29 return x
31 if x is None:
32 return None
34 source_code = None
36 if isinstance(x, functools.partial):
37 source_code = inspect.getsource(x.func)
39 if source_code is None:
40 try:
41 source_code = inspect.getsource(x)
42 except TypeError:
43 pass
45 if source_code is None:
46 try:
47 source_code = inspect.getsource(x.__call__)
48 except (TypeError, AttributeError):
49 pass
51 if source_code is None:
52 source_code = f"No source code available for {type(x)}"
53 return source_code
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.
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
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
85 formatter = Terminal256Formatter(**opts)
86 else:
87 from pygments.formatters.terminal import TerminalFormatter
89 formatter = TerminalFormatter(**opts)
90 return formatter