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 

58 Looks at first character to determine how to deserialize. 

59 

60 :param conn_id: connection id 

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

62 :return: the deserialized Connection 

63 """ 

64 from airflow.models.connection import Connection 

65 

66 value = value.strip() 

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

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

69 else: 

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

71 

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

73 """ 

74 Get conn_uri from Secrets Backend. 

75 

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

77 instead. 

78 

79 :param conn_id: connection id 

80 """ 

81 raise NotImplementedError() 

82 

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

84 """ 

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

86 

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

88 

89 :param conn_id: connection id 

90 """ 

91 value = None 

92 

93 not_implemented_get_conn_value = False 

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

95 try: 

96 value = self.get_conn_value(conn_id=conn_id) 

97 except NotImplementedError: 

98 not_implemented_get_conn_value = True 

99 warnings.warn( 

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

101 RemovedInAirflow3Warning, 

102 stacklevel=2, 

103 ) 

104 

105 if not_implemented_get_conn_value: 

106 try: 

107 value = self.get_conn_uri(conn_id=conn_id) 

108 except NotImplementedError: 

109 raise NotImplementedError( 

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

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

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

113 ) 

114 

115 if value: 

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

117 else: 

118 return None 

119 

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

121 """ 

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

123 

124 :param conn_id: connection id 

125 """ 

126 warnings.warn( 

127 "This method is deprecated. Please use " 

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

129 RemovedInAirflow3Warning, 

130 stacklevel=2, 

131 ) 

132 conn = self.get_connection(conn_id=conn_id) 

133 if conn: 

134 return [conn] 

135 return [] 

136 

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

138 """ 

139 Return value for Airflow Variable. 

140 

141 :param key: Variable Key 

142 :return: Variable Value 

143 """ 

144 raise NotImplementedError() 

145 

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

147 """ 

148 Return value for Airflow Config Key. 

149 

150 :param key: Config Key 

151 :return: Config Value 

152 """ 

153 return None