|
29 | 29 | from contextlib import contextmanager |
30 | 30 | from pathlib import Path |
31 | 31 | from tempfile import NamedTemporaryFile, _TemporaryFileWrapper |
| 32 | +from typing import Any |
32 | 33 |
|
33 | 34 | AIRFLOW_ROOT_PATH = Path(__file__).parents[3].resolve() |
34 | 35 | AIRFLOW_CORE_ROOT_PATH = AIRFLOW_ROOT_PATH / "airflow-core" |
@@ -118,19 +119,35 @@ def read_airflow_version() -> str: |
118 | 119 | ) |
119 | 120 |
|
120 | 121 |
|
| 122 | +def _read_global_constants_assignment(name: str) -> Any: |
| 123 | + """Read a top-level assignment from global_constants.py.""" |
| 124 | + tree = ast.parse(GLOBAL_CONSTANTS_PATH.read_text()) |
| 125 | + for node in tree.body: |
| 126 | + if isinstance(node, ast.Assign): |
| 127 | + for target in node.targets: |
| 128 | + if isinstance(target, ast.Name) and target.id == name: |
| 129 | + return ast.literal_eval(node.value) |
| 130 | + raise RuntimeError(f"{name} not found in global_constants.py") |
| 131 | + |
| 132 | + |
121 | 133 | def read_allowed_kubernetes_versions() -> list[str]: |
122 | 134 | """Parse ALLOWED_KUBERNETES_VERSIONS from global_constants.py (single source of truth). |
123 | 135 |
|
124 | 136 | Returns versions without the ``v`` prefix, e.g. ``["1.30.13", "1.31.12", ...]``. |
125 | 137 | """ |
126 | | - tree = ast.parse(GLOBAL_CONSTANTS_PATH.read_text()) |
127 | | - for node in tree.body: |
128 | | - if isinstance(node, ast.Assign): |
129 | | - for target in node.targets: |
130 | | - if isinstance(target, ast.Name) and target.id == "ALLOWED_KUBERNETES_VERSIONS": |
131 | | - versions: list[str] = ast.literal_eval(node.value) |
132 | | - return [v.lstrip("v") for v in versions] |
133 | | - raise RuntimeError("ALLOWED_KUBERNETES_VERSIONS not found in global_constants.py") |
| 138 | + versions: list[str] = _read_global_constants_assignment("ALLOWED_KUBERNETES_VERSIONS") |
| 139 | + return [v.lstrip("v") for v in versions] |
| 140 | + |
| 141 | + |
| 142 | +def read_default_python_major_minor_version_for_images() -> str: |
| 143 | + """Parse DEFAULT_PYTHON_MAJOR_MINOR_VERSION_FOR_IMAGES from global_constants.py.""" |
| 144 | + value = _read_global_constants_assignment("DEFAULT_PYTHON_MAJOR_MINOR_VERSION_FOR_IMAGES") |
| 145 | + if not isinstance(value, str): |
| 146 | + raise RuntimeError( |
| 147 | + "DEFAULT_PYTHON_MAJOR_MINOR_VERSION_FOR_IMAGES in global_constants.py " |
| 148 | + f"must be a string, got {type(value).__name__}" |
| 149 | + ) |
| 150 | + return value |
134 | 151 |
|
135 | 152 |
|
136 | 153 | def pre_process_mypy_files(files: list[str]) -> list[str]: |
|
0 commit comments