Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/secrets/base_secrets.py: 34%

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 warnings 

20from abc import ABC 

21from typing import TYPE_CHECKING 

22 

23from airflow.exceptions import RemovedInAirflow3Warning 

24 

25if TYPE_CHECKING: 

26 from airflow.models.connection import Connection 

27 

28 

29class BaseSecretsBackend(ABC): 

30 """Abstract base class to retrieve Connection object given a conn_id or Variable given a key.""" 

31 

32 @staticmethod 

33 def build_path(path_prefix: str, secret_id: str, sep: str = "/") -> str: 

34 """ 

35 Given conn_id, build path for Secrets Backend. 

36 

37 :param path_prefix: Prefix of the path to get secret 

38 :param secret_id: Secret id 

39 :param sep: separator used to concatenate connections_prefix and conn_id. Default: "/" 

40 """ 

41 return f"{path_prefix}{sep}{secret_id}" 

42 

43 def get_conn_value(self, conn_id: str) -> str | None: 

44 """ 

45 Retrieve from Secrets Backend a string value representing the Connection object. 

46 

47 If the client your secrets backend uses already returns a python dict, you should override 

48 ``get_connection`` instead. 

49 

50 :param conn_id: connection id 

51 """ 

52 raise NotImplementedError 

53 

54 def deserialize_connection(self, conn_id: str, value: str) -> Connection: 

55 """ 

56 Given a serialized representation of the airflow Connection, return an instance. 

57 Looks at first character to determine how to deserialize. 

58 

59 :param conn_id: connection id 

60 :param value: the serialized representation of the Connection object 

61 :return: the deserialized Connection 

62 """ 

63 from airflow.models.connection import Connection 

64 

65 value = value.strip() 

66 if value[0] == "{": 

67 return Connection.from_json(conn_id=conn_id, value=value) 

68 else: 

69 return Connection(conn_id=conn_id, uri=value) 

70 

71 def get_conn_uri(self, conn_id: str) -> str | None: 

72 """ 

73 Get conn_uri from Secrets Backend. 

74 

75 This method is deprecated and will be removed in a future release; implement ``get_conn_value`` 

76 instead. 

77 

78 :param conn_id: connection id 

79 """ 

80 raise NotImplementedError() 

81 

82 def get_connection(self, conn_id: str) -> Connection | None: 

83 """ 

84 Return connection object with a given ``conn_id``. 

85 

86 Tries ``get_conn_value`` first and if not implemented, tries ``get_conn_uri`` 

87 

88 :param conn_id: connection id 

89 """ 

90 value = None 

91 

92 not_implemented_get_conn_value = False 

93 # TODO: after removal of ``get_conn_uri`` we should not catch NotImplementedError here 

94 try: 

95 value = self.get_conn_value(conn_id=conn_id) 

96 except NotImplementedError: 

97 not_implemented_get_conn_value = True 

98 warnings.warn( 

99 "Method `get_conn_uri` is deprecated. Please use `get_conn_value`.", 

100 RemovedInAirflow3Warning, 

101 stacklevel=2, 

102 ) 

103 

104 if not_implemented_get_conn_value: 

105 try: 

106 value = self.get_conn_uri(conn_id=conn_id) 

107 except NotImplementedError: 

108 raise NotImplementedError( 

109 f"Secrets backend {self.__class__.__name__} neither implements " 

110 "`get_conn_value` nor `get_conn_uri`. Method `get_conn_uri` is " 

111 "deprecated and will be removed in a future release. Please implement `get_conn_value`." 

112 ) 

113 

114 if value: 

115 return self.deserialize_connection(conn_id=conn_id, value=value) 

116 else: 

117 return None 

118 

119 def get_connections(self, conn_id: str) -> list[Connection]: 

120 """ 

121 Return connection object with a given ``conn_id``. 

122 

123 :param conn_id: connection id 

124 """ 

125 warnings.warn( 

126 "This method is deprecated. Please use " 

127 "`airflow.secrets.base_secrets.BaseSecretsBackend.get_connection`.", 

128 RemovedInAirflow3Warning, 

129 stacklevel=2, 

130 ) 

131 conn = self.get_connection(conn_id=conn_id) 

132 if conn: 

133 return [conn] 

134 return [] 

135 

136 def get_variable(self, key: str) -> str | None: 

137 """ 

138 Return value for Airflow Variable. 

139 

140 :param key: Variable Key 

141 :return: Variable Value 

142 """ 

143 raise NotImplementedError() 

144 

145 def get_config(self, key: str) -> str | None: 

146 """ 

147 Return value for Airflow Config Key. 

148 

149 :param key: Config Key 

150 :return: Config Value 

151 """ 

152 return None