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

Code before:
        )

        self.vault_client = _VaultClient(**client_kwargs)

    def _get_kubernetes_parameters_from_connection(
        self, kubernetes_jwt_path: str | None, kubernetes_role: str | None
    ) -> tuple[str | None, str | None]:
        if not kubernetes_jwt_path:
            kubernetes_jwt_path = self.connection.extra_dejson.get("kubernetes_jwt_path")
            if not kubernetes_jwt_path:
                kubernetes_jwt_path = DEFAULT_KUBERNETES_JWT_PATH
        if not kubernetes_role:
            kubernetes_role = self.connection.extra_dejson.get("kubernetes_role")
        return kubernetes_jwt_path, kubernetes_role

    def _get_jwt_parameters_from_connection(
        self, jwt_role: str | None, jwt_token: str | None, jwt_token_path: str | None
    ) -> tuple[str | None, str | None, str | None]:
        if not jwt_role:
            jwt_role = self.connection.extra_dejson.get("jwt_role")
        if not jwt_token:
            jwt_token = self.connection.extra_dejson.get("jwt_token")
        if not jwt_token_path:
            jwt_token_path = self.connection.extra_dejson.get("jwt_token_path")
            if not jwt_token_path:
                jwt_token_path = DEFAULT_JWT_TOKEN_PATH
        return jwt_role, jwt_token, jwt_token_path

    def _get_gcp_parameters_from_connection(
        self,
        gcp_key_path: str | None,
        gcp_scopes: str | None,
    ) -> tuple[str | None, dict | None, str | None]:
        if not gcp_scopes:
            gcp_scopes = self.connection.extra_dejson.get("gcp_scopes")
        if not gcp_key_path:
            gcp_key_path = self.connection.extra_dejson.get("gcp_key_path")
        string_keyfile_dict = self.connection.extra_dejson.get("gcp_keyfile_dict")
        gcp_keyfile_dict = json.loads(string_keyfile_dict) if string_keyfile_dict else None
        return gcp_key_path, gcp_keyfile_dict, gcp_scopes

    def _get_azure_parameters_from_connection(
        self, azure_resource: str | None, azure_tenant_id: str | None
    ) -> tuple[str | None, str | None]:
        if not azure_tenant_id:
            azure_tenant_id = self.connection.extra_dejson.get("azure_tenant_id")
        if not azure_resource:
            azure_resource = self.connection.extra_dejson.get("azure_resource")
        return azure_resource, azure_tenant_id

    def _get_radius_parameters_from_connection(

Review comment (suggestion):
```suggestion
            jwt_token_path = self.connection.extra_dejson.get("jwt_token_path") or DEFAULT_JWT_TOKEN_PATH
```

Code after:

    def _get_kubernetes_parameters_from_connection(
        self, kubernetes_jwt_path: str | None, kubernetes_role: str | None
    ) -> tuple[str | None, str | None]:
        if not kubernetes_jwt_path:
            kubernetes_jwt_path = self.connection.extra_dejson.get("kubernetes_jwt_path")
            if not kubernetes_jwt_path:
                kubernetes_jwt_path = DEFAULT_KUBERNETES_JWT_PATH
        if not kubernetes_role:
            kubernetes_role = self.connection.extra_dejson.get("kubernetes_role")
        return kubernetes_jwt_path, kubernetes_role

    def _get_jwt_parameters_from_connection(
        self, jwt_role: str | None, jwt_token: str | None, jwt_token_path: str | None
    ) -> tuple[str | None, str | None, str | None]:
        if not jwt_role:
            jwt_role = self.connection.extra_dejson.get("jwt_role")
        if not jwt_token:
            jwt_token = self.connection.extra_dejson.get("jwt_token")
        if not jwt_token_path:
            jwt_token_path = self.connection.extra_dejson.get("jwt_token_path")
        return jwt_role, jwt_token, jwt_token_path

    def _get_gcp_parameters_from_connection(
        self,
        gcp_key_path: str | None,
        gcp_scopes: str | None,
    ) -> tuple[str | None, dict | None, str | None]:
        if not gcp_scopes:
            gcp_scopes = self.connection.extra_dejson.get("gcp_scopes")
        if not gcp_key_path:
            gcp_key_path = self.connection.extra_dejson.get("gcp_key_path")
        string_keyfile_dict = self.connection.extra_dejson.get("gcp_keyfile_dict")
        gcp_keyfile_dict = json.loads(string_keyfile_dict) if string_keyfile_dict else None
        return gcp_key_path, gcp_keyfile_dict, gcp_scopes

    def _get_azure_parameters_from_connection(
        self, azure_resource: str | None, azure_tenant_id: str | None
    ) -> tuple[str | None, str | None]:
        if not azure_tenant_id:
            azure_tenant_id = self.connection.extra_dejson.get("azure_tenant_id")
        if not azure_resource:
            azure_resource = self.connection.extra_dejson.get("azure_resource")
        return azure_resource, azure_tenant_id

    def _get_radius_parameters_from_connection(
        self, radius_host: str | None, radius_port: int | None
    ) -> tuple[str | None, int | None]:
        if not radius_port:
            radius_port_str = self.connection.extra_dejson.get("radius_port")
            if radius_port_str: