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

Code before:
        return self.vault_client.get_secret(
            secret_path=(mount_point + "/" if mount_point else "") + secret_path
        )

    # 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"]
        except KeyError:
            self.log.warning('Vault connection %s fetched but does not have required key "conn_uri"', conn_id)
            return Connection(conn_id, **response)

        return Connection(conn_id, uri=uri)

    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:

Review comment (suggestion):
For readability sake I'd place the `return Connection(conn_id, uri=uri)` inside the `try` clause, but the business logic looks ok.

Code after:
        return self.vault_client.get_secret(
            secret_path=(mount_point + "/" if mount_point else "") + secret_path
        )

    # 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