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

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 textwrap 

20 

21from fastapi import Depends, HTTPException, status 

22 

23from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrText 

24from airflow.api_fastapi.common.router import AirflowRouter 

25from airflow.api_fastapi.common.types import Mimetype 

26from airflow.api_fastapi.core_api.datamodels.config import ( 

27 Config, 

28 ConfigOption, 

29 ConfigSection, 

30) 

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

32from airflow.api_fastapi.core_api.security import requires_access_configuration 

33from airflow.api_fastapi.core_api.services.public.config import ( 

34 _check_expose_config, 

35 _is_per_key_sensitive_option, 

36 _mask_per_key_sensitive_options, 

37 _response_based_on_accept, 

38) 

39from airflow.configuration import conf 

40 

41text_example_response_for_get_config_value = { 

42 Mimetype.TEXT: { 

43 "schema": { 

44 "type": "string", 

45 "example": textwrap.dedent( 

46 """\ 

47 [core] 

48 dags_folder = /opt/airflow/dags 

49 base_log_folder = /opt/airflow/logs 

50 """ 

51 ), 

52 } 

53 } 

54} 

55 

56text_example_response_for_get_config = { 

57 Mimetype.TEXT: { 

58 "schema": { 

59 "type": "string", 

60 "example": textwrap.dedent( 

61 """\ 

62 [core] 

63 dags_folder = /opt/airflow/dags 

64 base_log_folder = /opt/airflow/logs 

65 

66 [smtp] 

67 smtp_host = localhost 

68 smtp_mail_from = airflow@example.com 

69 """ 

70 ), 

71 }, 

72 } 

73} 

74config_router = AirflowRouter(tags=["Config"], prefix="/config") 

75 

76 

77@config_router.get( 

78 "", 

79 responses={ 

80 **create_openapi_http_exception_doc( 

81 [ 

82 status.HTTP_404_NOT_FOUND, 

83 status.HTTP_406_NOT_ACCEPTABLE, 

84 ] 

85 ), 

86 status.HTTP_200_OK: { 

87 "description": "Successful Response", 

88 "content": text_example_response_for_get_config, 

89 }, 

90 }, 

91 response_model=Config, 

92 dependencies=[Depends(requires_access_configuration("GET"))], 

93) 

94def get_config( 

95 accept: HeaderAcceptJsonOrText, 

96 section: str | None = None, 

97): 

98 display_sensitive = _check_expose_config() 

99 

100 if section and not conf.has_section(section): 

101 raise HTTPException( 

102 status_code=status.HTTP_404_NOT_FOUND, 

103 detail=f"Section {section} not found.", 

104 ) 

105 conf_dict = conf.as_dict(display_source=False, display_sensitive=display_sensitive) 

106 if not display_sensitive: 

107 _mask_per_key_sensitive_options(conf_dict) 

108 

109 if section: 

110 conf_section_value = conf_dict[section] 

111 conf_dict.clear() 

112 conf_dict[section] = conf_section_value 

113 

114 config = Config( 

115 sections=[ 

116 ConfigSection( 

117 name=section, options=[ConfigOption(key=key, value=value) for key, value in options.items()] 

118 ) 

119 for section, options in conf_dict.items() 

120 ] 

121 ) 

122 return _response_based_on_accept(accept, config) 

123 

124 

125@config_router.get( 

126 "/section/{section}/option/{option}", 

127 responses={ 

128 **create_openapi_http_exception_doc( 

129 [ 

130 status.HTTP_404_NOT_FOUND, 

131 status.HTTP_406_NOT_ACCEPTABLE, 

132 ] 

133 ), 

134 status.HTTP_200_OK: { 

135 "description": "Successful Response", 

136 "content": text_example_response_for_get_config_value, 

137 }, 

138 }, 

139 response_model=Config, 

140 dependencies=[Depends(requires_access_configuration("GET"))], 

141) 

142def get_config_value( 

143 section: str, 

144 option: str, 

145 accept: HeaderAcceptJsonOrText, 

146): 

147 _check_expose_config() 

148 

149 if not conf.has_option(section, option): 

150 raise HTTPException( 

151 status_code=status.HTTP_404_NOT_FOUND, 

152 detail=f"Option [{section}/{option}] not found.", 

153 ) 

154 

155 section_l, option_l = section.lower(), option.lower() 

156 if conf.is_sensitive_option(section_l, option_l) or _is_per_key_sensitive_option(section_l, option_l): 

157 value = "< hidden >" 

158 else: 

159 value = conf.get(section, option) 

160 

161 config = Config(sections=[ConfigSection(name=section, options=[ConfigOption(key=option, value=value)])]) 

162 return _response_based_on_accept(accept, config)