Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/ui/config.py: 76%

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 

19from json import loads 

20from typing import Any 

21 

22from fastapi import Depends, status 

23 

24from airflow.api_fastapi.common.router import AirflowRouter 

25from airflow.api_fastapi.common.types import UIAlert 

26from airflow.api_fastapi.core_api.datamodels.ui.config import ConfigResponse 

27from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc 

28from airflow.api_fastapi.core_api.security import requires_authenticated 

29from airflow.configuration import conf 

30from airflow.settings import DASHBOARD_UIALERTS 

31from airflow.utils.log.log_reader import TaskLogReader 

32 

33config_router = AirflowRouter(tags=["Config"]) 

34 

35 

36API_CONFIG_KEYS = [ 

37 "enable_swagger_ui", 

38 "hide_paused_dags_by_default", 

39 "fallback_page_limit", 

40 "default_wrap", 

41 "auto_refresh_interval", 

42 "require_confirmation_dag_change", 

43] 

44 

45 

46@config_router.get( 

47 "/config", 

48 responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]), 

49 dependencies=[Depends(requires_authenticated())], 

50) 

51def get_configs() -> ConfigResponse: 

52 """Get configs for UI.""" 

53 config = {key: conf.get("api", key) for key in API_CONFIG_KEYS} 

54 

55 task_log_reader = TaskLogReader() 

56 additional_config: dict[str, Any] = { 

57 "instance_name": conf.get("api", "instance_name", fallback="Airflow"), 

58 "test_connection": conf.get("core", "test_connection", fallback="Disabled"), 

59 # Expose "dashboard_alert" using a list comprehension so UIAlert instances can be expressed dynamically. 

60 "dashboard_alert": [alert for alert in DASHBOARD_UIALERTS if isinstance(alert, UIAlert)], 

61 "show_external_log_redirect": task_log_reader.supports_external_link, 

62 "external_log_name": getattr(task_log_reader.log_handler, "log_name", None), 

63 "theme": loads(conf.get("api", "theme", fallback="{}")) or None, 

64 "multi_team": conf.getboolean("core", "multi_team"), 

65 # Return None when the option isn't explicitly set so the UI hook can apply 

66 # its own fallback (False for clear/rerun, True for backfills). 

67 "rerun_with_latest_version": ( 

68 conf.getboolean("core", "rerun_with_latest_version") 

69 if conf.has_option("core", "rerun_with_latest_version") 

70 else None 

71 ), 

72 } 

73 

74 config.update({key: value for key, value in additional_config.items()}) 

75 

76 return ConfigResponse.model_validate(config)