Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/_shared/secrets_backend/base.py: 36%

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 inspect 

20from abc import ABC 

21from collections.abc import Callable 

22 

23 

24def accepts_kwarg(method: Callable, name: str) -> bool: 

25 """ 

26 Return whether a secrets-backend method accepts the keyword *name*. 

27 

28 Backends override ``get_conn_value`` / ``get_variable`` / ``get_connection`` with 

29 whatever signature was current when they were written, so a keyword added later cannot 

30 be forwarded blindly: it raises ``TypeError`` inside the backend, which callers swallow 

31 per-backend and report as a missing secret. ``team_name`` (added by AIP-67 in 3.2) is 

32 one example. A method is considered to accept *name* if it explicitly declares the 

33 parameter or has a ``**kwargs`` catch-all. 

34 """ 

35 try: 

36 parameters = inspect.signature(method).parameters 

37 except (TypeError, ValueError): 

38 # Un-introspectable callable (e.g. C-implemented): assume the current signature. 

39 return True 

40 return name in parameters or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in parameters.values()) 

41 

42 

43def call_secrets_backend_method(method: Callable, *, team_name: str | None, **kwargs): 

44 """ 

45 Call a secrets-backend lookup ``method``, forwarding ``team_name`` only when supported. 

46 

47 Forward ``team_name`` to backends that accept it (3.2+ overrides) and omit it for 

48 pre-3.2 overrides, so older bundled providers and custom backends keep working -- 

49 in both single-team and multi-team deployments -- without being forced to add the 

50 parameter. A ``TypeError`` raised inside an accepting backend is left to propagate 

51 rather than retried without ``team_name``, which could mask the error and resolve a 

52 team-scoped lookup against the global scope. 

53 """ 

54 if accepts_kwarg(method, "team_name"): 

55 return method(team_name=team_name, **kwargs) 

56 return method(**kwargs) 

57 

58 

59class BaseSecretsBackend(ABC): 

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

61 

62 @staticmethod 

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

64 """ 

65 Given conn_id, build path for Secrets Backend. 

66 

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

68 :param secret_id: Secret id 

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

70 """ 

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

72 

73 def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | None: 

74 """ 

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

76 

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

78 ``get_connection`` instead. 

79 

80 :param conn_id: connection id 

81 :param team_name: Team name associated to the task trying to access the connection (if any) 

82 """ 

83 raise NotImplementedError 

84 

85 def get_variable(self, key: str, team_name: str | None = None) -> str | None: 

86 """ 

87 Return value for Airflow Variable. 

88 

89 :param key: Variable Key 

90 :param team_name: Team name associated to the task trying to access the variable (if any) 

91 :return: Variable Value 

92 """ 

93 raise NotImplementedError() 

94 

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

96 """ 

97 Return value for Airflow Config Key. 

98 

99 :param key: Config Key 

100 :return: Config Value 

101 """ 

102 return None 

103 

104 def _set_connection_class(self, conn_class: type) -> None: 

105 if not isinstance(conn_class, type): 

106 raise TypeError(f"Connection class must be a type/class, got {type(conn_class).__name__}") 

107 self._connection_class = conn_class 

108 

109 def _get_connection_class(self) -> type: 

110 """Get the Connection class to use for deserialization.""" 

111 conn_class = getattr(self, "_connection_class", None) 

112 if conn_class is None: 

113 raise RuntimeError( 

114 "Connection class not set on backend instance. " 

115 "Backends must be instantiated via initialize_secrets_backends() " 

116 "or have _connection_class set manually." 

117 ) 

118 return conn_class 

119 

120 @staticmethod 

121 def _deserialize_connection_value(conn_class: type, conn_id: str, value: str): 

122 value = value.strip() 

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

124 # JSON secrets stored by Airflow 2 backends may lack both "conn_type" 

125 # and "uri" (e.g. {"host": ..., "login": ..., "password": ...}). This is 

126 # valid: conn_type is intentionally optional on the SDK Connection model 

127 # for Airflow 2 -> 3 migration compatibility. 

128 # Check: https://github.com/apache/airflow/pull/61728 

129 # TODO: Remove this compatibility once the minimum supported Airflow 

130 # version in providers is 3.0. 

131 return conn_class.from_json(value=value, conn_id=conn_id) # type: ignore[attr-defined] 

132 

133 # TODO: Only sdk has from_uri defined on it. Is it worthwhile developing the core path or not? 

134 if hasattr(conn_class, "from_uri"): 

135 return conn_class.from_uri(conn_id=conn_id, uri=value) 

136 return conn_class(conn_id=conn_id, uri=value) 

137 

138 def deserialize_connection(self, conn_id: str, value: str): 

139 """ 

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

141 

142 Uses the Connection class set on this class (which should be set to the appropriate Connection class for the execution context). 

143 Uses Connection.from_json() for JSON format, Connection(uri=...) for URI format. 

144 

145 :param conn_id: connection id 

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

147 :return: the deserialized Connection 

148 """ 

149 conn_class = self._get_connection_class() 

150 return self._deserialize_connection_value(conn_class, conn_id, value) 

151 

152 def get_connection(self, conn_id: str, team_name: str | None = None): 

153 """ 

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

155 

156 :param conn_id: connection id 

157 :param team_name: Team name associated to the task trying to access the connection (if any) 

158 :return: Connection object or None 

159 """ 

160 value = call_secrets_backend_method(self.get_conn_value, team_name=team_name, conn_id=conn_id) 

161 if value: 

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

163 return None