Repository: apache/airflow
File: providers/hashicorp/src/airflow/providers/hashicorp/secrets/vault.py (Python)

Code before:

    # Make sure connection is imported this way for type checking, otherwise when importing
    # the backend it will get a circular dependency and fail
    if TYPE_CHECKING:
        from airflow.models.connection import Connection

    def get_connection(self, conn_id: str) -> Connection | None:
        """
        Get connection from Vault as secret.

        Prioritize conn_uri if exists, if not fall back to normal Connection creation.

        :return: A Connection object constructed from Vault data
        """
        # The Connection needs to be locally imported because otherwise we get into cyclic import
        # problems when instantiating the backend during configuration
        from airflow.models.connection import Connection

        response = self.get_response(conn_id)
        if not response:
            return None
        try:
            uri = response["conn_uri"]
            return Connection(conn_id, uri=uri)
        except KeyError:
            self.log.warning('Vault connection %s fetched but does not have required key "conn_uri"', conn_id)
            return Connection(conn_id, **response)
             

    def get_variable(self, key: str, team_name: str | None = None) -> str | None:
        """
        Get Airflow Variable.

        :param key: Variable Key
        :param team_name: Team name associated to the task trying to access the variable (if any)
        :return: Variable Value retrieved from the vault
        """
        mount_point, variable_key = self._parse_path(key)
        if self.variables_path is None or variable_key is None:
            return None
        if self.variables_path == "":
            secret_path = variable_key
        else:
            secret_path = self.build_path(self.variables_path, variable_key)
        response = self.vault_client.get_secret(
            secret_path=(mount_point + "/" if mount_point else "") + secret_path
        )
        if not response:
            return None
        try:
            return response["value"]

Review comment (suggestion):
On second thought, this is not the behaviour we want for fetching connections. While variables and configs can only be set via one key `value` (where it makes sense to warn when a key `value` isn't set), connections can be set in two ways: `conn_uri` or the [json-representation keys](https://airflow.apache.org/docs/apache-airflow/3.1.5/howto/connection.html#generating-a-json-connection-representation) (`host`, `login`, etc.). This code would therefore introduce an (undesirable) warning when fetching connections that are configured with the json convention.

I suggest we remove the changes for `get_connection`.

Code after:

    # Make sure connection is imported this way for type checking, otherwise when importing
    # the backend it will get a circular dependency and fail
    if TYPE_CHECKING:
        from airflow.models.connection import Connection

    def get_connection(self, conn_id: str) -> Connection | None:
        """
        Get connection from Vault as secret.

        Prioritize conn_uri if exists, if not fall back to normal Connection creation.

        :return: A Connection object constructed from Vault data
        """
        # The Connection needs to be locally imported because otherwise we get into cyclic import
        # problems when instantiating the backend during configuration
        from airflow.models.connection import Connection

        response = self.get_response(conn_id)
        if response is None:
            return None

        uri = response.get("conn_uri")
        if uri:
            return Connection(conn_id, uri=uri)

        return Connection(conn_id, **response)

    def get_variable(self, key: str, team_name: str | None = None) -> str | None:
        """
        Get Airflow Variable.

        :param key: Variable Key
        :param team_name: Team name associated to the task trying to access the variable (if any)
        :return: Variable Value retrieved from the vault
        """
        mount_point, variable_key = self._parse_path(key)
        if self.variables_path is None or variable_key is None:
            return None
        if self.variables_path == "":
            secret_path = variable_key
        else:
            secret_path = self.build_path(self.variables_path, variable_key)
        response = self.vault_client.get_secret(
            secret_path=(mount_point + "/" if mount_point else "") + secret_path
        )
        if not response:
            return None
        try:
            return response["value"]
        except KeyError: