Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/secrets/base_secrets.py: 34%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
19import warnings
20from abc import ABC
21from typing import TYPE_CHECKING
23from airflow.exceptions import RemovedInAirflow3Warning
25if TYPE_CHECKING:
26 from airflow.models.connection import Connection
29class BaseSecretsBackend(ABC):
30 """Abstract base class to retrieve Connection object given a conn_id or Variable given a key."""
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.
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}"
43 def get_conn_value(self, conn_id: str) -> str | None:
44 """
45 Retrieve from Secrets Backend a string value representing the Connection object.
47 If the client your secrets backend uses already returns a python dict, you should override
48 ``get_connection`` instead.
50 :param conn_id: connection id
51 """
52 raise NotImplementedError
54 def deserialize_connection(self, conn_id: str, value: str) -> Connection:
55 """
56 Given a serialized representation of the airflow Connection, return an instance.
58 Looks at first character to determine how to deserialize.
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
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)
72 def get_conn_uri(self, conn_id: str) -> str | None:
73 """
74 Get conn_uri from Secrets Backend.
76 This method is deprecated and will be removed in a future release; implement ``get_conn_value``
77 instead.
79 :param conn_id: connection id
80 """
81 raise NotImplementedError()
83 def get_connection(self, conn_id: str) -> Connection | None:
84 """
85 Return connection object with a given ``conn_id``.
87 Tries ``get_conn_value`` first and if not implemented, tries ``get_conn_uri``
89 :param conn_id: connection id
90 """
91 value = None
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 )
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 )
115 if value:
116 return self.deserialize_connection(conn_id=conn_id, value=value)
117 else:
118 return None
120 def get_connections(self, conn_id: str) -> list[Connection]:
121 """
122 Return connection object with a given ``conn_id``.
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 []
137 def get_variable(self, key: str) -> str | None:
138 """
139 Return value for Airflow Variable.
141 :param key: Variable Key
142 :return: Variable Value
143 """
144 raise NotImplementedError()
146 def get_config(self, key: str) -> str | None:
147 """
148 Return value for Airflow Config Key.
150 :param key: Config Key
151 :return: Config Value
152 """
153 return None