Custom secret backend issue

[dor-schreiber]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.9.1

What happened?

Can't use a custom secret backend. I want to make a backend that reads a yaml file to override connections From the GCP secret manager.

Getting error:

Module "plugins.custom_secrets_backend" does not define a "LocalSecretBackend" attribute/class
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
File ~/.local/lib/python3.11/site-packages/airflow/utils/module_loading.py:42, in import_string(dotted_path)
     41 try:
---> 42     return getattr(module, class_name)
     43 except AttributeError:

AttributeError: partially initialized module 'plugins.custom_secrets_backend' has no attribute 'LocalSecretBackend' (most likely due to a circular import)

During handling of the above exception, another exception occurred:

ImportError                               Traceback (most recent call last)
File ~/.local/lib/python3.11/site-packages/airflow/configuration.py:1211, in AirflowConfigParser.getimport(self, section, key, **kwargs)
   1210 try:
-> 1211     return import_string(full_qualified_path)
   1212 except ImportError as e:

File ~/.local/lib/python3.11/site-packages/airflow/utils/module_loading.py:44, in import_string(dotted_path)
     43 except AttributeError:
---> 44     raise ImportError(f'Module "{module_path}" does not define a "{class_name}" attribute/class')

ImportError: Module "plugins.custom_secrets_backend" does not define a "LocalSecretBackend" attribute/class

During handling of the above exception, another exception occurred:

AirflowConfigException                    Traceback (most recent call last)
Cell In[1], line 1
----> 1 from plugins.custom_secrets_backend import LocalSecretBackend

File /opt/airflow/plugins/custom_secrets_backend.py:1
----> 1 import airflow.providers.google.cloud.secrets.secret_manager
      2 import yaml
      3 from airflow.models.connection import Connection

File ~/.local/lib/python3.11/site-packages/airflow/__init__.py:52
     35     warnings.warn(
     36         "Airflow currently can be run on POSIX-compliant Operating Systems. For development, "
     37         "it is regularly tested on fairly modern Linux Distros and recent versions of macOS. "
   (...)
     42         stacklevel=1,
     43     )
     45 # The configuration module initializes and validates the conf object as a side effect the first
     46 # time it is imported. If it is not imported before importing the settings module, the conf
     47 # object will then be initted/validated as a side effect of it being imported in settings,
   (...)
     50 # those functions likely import settings).
     51 # configuration is therefore initted early here, simply by importing it.
---> 52 from airflow import configuration, settings
     54 __all__ = [
     55     "__version__",
     56     "DAG",
     57     "Dataset",
     58     "XComArg",
     59 ]
     61 # Make `airflow` a namespace package, supporting installing
     62 # airflow.providers.* in different locations (i.e. one in site, and one in user
     63 # lib.)

File ~/.local/lib/python3.11/site-packages/airflow/configuration.py:2347
   2344 WEBSERVER_CONFIG = ""  # Set by initialize_config
   2346 conf: AirflowConfigParser = initialize_config()
-> 2347 secrets_backend_list = initialize_secrets_backends()
   2348 conf.validate()

File ~/.local/lib/python3.11/site-packages/airflow/configuration.py:2255, in initialize_secrets_backends()
   2247 """
   2248 Initialize secrets backend.
   2249
   2250 * import secrets backend classes
   2251 * instantiate them and return them in a list
   2252 """
   2253 backend_list = []
-> 2255 custom_secret_backend = get_custom_secret_backend()
   2257 if custom_secret_backend is not None:
   2258     backend_list.append(custom_secret_backend)

File ~/.local/lib/python3.11/site-packages/airflow/configuration.py:2225, in get_custom_secret_backend()
   2223 def get_custom_secret_backend() -> BaseSecretsBackend | None:
   2224     """Get Secret Backend if defined in airflow.cfg."""
-> 2225     secrets_backend_cls = conf.getimport(section="secrets", key="backend")
   2227     if not secrets_backend_cls:
   2228         return None

File ~/.local/lib/python3.11/site-packages/airflow/configuration.py:1214, in AirflowConfigParser.getimport(self, section, key, **kwargs)
   1212 except ImportError as e:
   1213     log.warning(e)
-> 1214     raise AirflowConfigException(
   1215         f'The object could not be loaded. Please check "{key}" key in "{section}" section. '
   1216         f'Current value: "{full_qualified_path}".'
   1217     )

AirflowConfigException: The object could not be loaded. Please check "backend" key in "secrets" section. Current value: "plugins.custom_secrets_backend.LocalSecretBackend".
2024-11-07 17:41:21 airflow.exceptions.AirflowConfigException: The object could not be loaded. Please check "backend" key in "secrets" section. Current value: "plugins.custom_secrets_backend.CloudSecretManagerBackend".

What you think should happen instead?

No response

How to reproduce

Code for custom backend:
File name custom_secrets_backend under plugins folder

import yaml
from airflow.models.connection import Connection
from airflow.providers.google.cloud.secrets.secret_manager import (
    CloudSecretManagerBackend,
)


class LocalSecretBackend(CloudSecretManagerBackend):

    def __init__(self, conn_path: str = None, env_path: str = None, **kwargs):
        super().__init__(**kwargs)
        self.env_path = env_path
        self.env_values = self._read_env_file()
        self.conn_path = conn_path
        self.conn_values = self._read_conn_file()

    def get_connection(self, conn_id: str):
        if conn_id in self.conn_values:
            return self.conn_values[conn_id]
        else:
            return super().get_connection(conn_id=conn_id)

    def get_variable(self, key: str) -> str | None:
        if key in self.env_values:
            return self.env_values[key]
        else:
            return super().get_variable(key=key)

    def _read_env_file(self):
        env_dict = {}
        with open(self.env_path, "r") as file:
            for line in file:
                # Remove comments and skip empty lines
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                # Split key and value
                key, value = line.split("=", 1)
                env_dict[key.strip()] = value.strip()
        return env_dict

    def _read_conn_file(self):
        conn_dict = {}

        # Load YAML file
        with open(self.conn_path, "r") as file:
            config = yaml.safe_load(file)

        # Extract connections list
        connections = config.get("connections", [])

        # Create each connection
        for conn in connections:
            conn_id = conn.get("conn_id")
            conn_type = conn.get("conn_type")
            host = conn.get("host")
            schema = conn.get("schema")
            login = conn.get("login")
            password = conn.get("password")
            port = conn.get("port")
            extra = conn.get("extra")

            conn_dict[conn_id] = Connection(
                conn_id=conn_id,
                conn_type=conn_type,
                host=host,
                schema=schema,
                login=login,
                password=password,
                port=port,
                extra=extra,
            )

        return conn_dict

in airflow.cfg file:

[secrets]
backend = plugins.custom_secrets_backend.LocalSecretBackend
backend_kwargs = {"project_id": "", "connections_prefix":"airflow-connection", "conn_path": "local/local_connections.yaml"}

example yaml:

connections:
  - conn_id: mysql_conn
    conn_type: mysql
    host: 
    schema:
    login:
    password: 
    port:

Operating System

Mac Sonoma 14.4.1

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

Converted it to a discussion, it's not an airflow issue, It is something you have problems with and ask for help - so this should be a discussion in the first place.

Looks lke you have circular import somewhere. There are two ways to deal with circular imports : if TYPE_CHECKING (when your issue is caused by typing) or local imports (if your top level import is importing something that imports your module in turn) - look for many other places where both were applied.

[dor-schreiber]

@potiuk Thanks for the response.
The circular import doesn't make much sense.
I'm only importing:

from airflow.providers.google.cloud.secrets.secret_manager import (
    CloudSecretManagerBackend,
)

So I'm not sure why I get the issue

[potiuk]

Don't know either. It needs a bit digging - circular imports sometimes require drilling down and analysing what is going on - you need to look deeper.

Clearly it is what you are experiencing.

AttributeError: partially initialized module 'plugins.custom_secrets_backend' has no attribute 'LocalSecretBackend' (most likely due to a circular import)

[romantsovmike]

Hi @dor-schreiber
I'm facing the same issue right now.
Were you able to solve that somehow?

Deployment details - Composer 3 in Google Cloud Platform, Airflow version - 2.9.3/2.10.2 (both not working).

[dor-schreiber]

Hi @romantsovmike
I wasn't having much luck with it. Iv'e decided to drop the secret manager option for local development and use LocalFilesystemBackend for now.
Another solution I have is to add support in the CloudSecretManagerBackend for ignoring keys:

class CloudSecretManagerBackend(BaseSecretsBackend, LoggingMixin):
   def __init__(
        self,
       ...
      ignore_keys: Sequence[str] | None = None,
        **kwargs,
    ) -> None:
      ...

    #example for one of the get functions
    def get_variable(self, key: str) -> str | None:
        if self.variables_prefix is None:
            return None

        if key in ignore_keys:
            return None

        return self._get_secret(self.variables_prefix, key)

The main issue with my request is that the flow of the class is that he access the secret manager first always so we don't have the option to override the same key.

@potiuk WDYT about the ignore_keys?