Repository: home-assistant/core
File: homeassistant/helpers/config_entry_oauth2_flow.py (Python)

Code before:
            .update_query(self.extra_authorize_data)
        )

    async def async_resolve_external_data(self, external_data: Any) -> dict:
        """Resolve the authorization code to tokens."""
        request_data: dict = {
            "grant_type": "authorization_code",
            "code": external_data["code"],
            "redirect_uri": external_data["state"]["redirect_uri"],
        }
        request_data.update(self.extra_token_resolve_data)
        return await self._token_request(request_data)

    async def _async_refresh_token(self, token: dict) -> dict:
        """Refresh tokens."""
        try:
            new_token = await self._token_request(
                {
                    "grant_type": "refresh_token",
                    "client_id": self.client_id,
                    "refresh_token": token["refresh_token"],
                }
            )
        except ClientResponseError:
            _LOGGER.warning(
                "Refresh token failed. Please catch 'ConfigEntryRefreshTokenFailed'. This will be added in version 2026.x.x and result in a breaking change",
            )
            raise  # Re-raise the exception to keep it non-breaking

        return {**token, **new_token}

    async def _token_request(self, data: dict) -> dict:
        """Make a token request."""
        session = async_get_clientsession(self.hass)

        data["client_id"] = self.client_id
        if self.client_secret:
            data["client_secret"] = self.client_secret

        _LOGGER.debug("Sending token request to %s", self.token_url)

        resp = await session.post(self.token_url, data=data)
        if resp.status >= 400:
            try:
                error_response = await resp.json()
            except (ClientError, JSONDecodeError):
                error_response = {}
            error_code = error_response.get("error", "unknown")
            error_description = error_response.get("error_description", "unknown error")
            _LOGGER.error(
                "Token request for %s failed (%s): %s",

Review comment (suggestion):
I think we'll need to add some more logic to handle the deprecation period properly.

1. We only want to log this warning once during the Home Assistant or at least config entry lifetime.
2. We want to allow integrations to start using the new exception now. The integrations that use the new exception should not get a warning. I'd investigate inheriting from `ClientResponseError` in the new exception to keep it backwards compatible and then raise the new exception already now.

Code after:
                {
                    "response_type": "code",
                    "client_id": self.client_id,
                    "redirect_uri": redirect_uri,
                    "state": _encode_jwt(
                        self.hass, {"flow_id": flow_id, "redirect_uri": redirect_uri}
                    ),
                }
            )
            .update_query(self.extra_authorize_data)
        )

    async def async_resolve_external_data(self, external_data: Any) -> dict:
        """Resolve the authorization code to tokens."""
        request_data: dict = {
            "grant_type": "authorization_code",
            "code": external_data["code"],
            "redirect_uri": external_data["state"]["redirect_uri"],
        }
        request_data.update(self.extra_token_resolve_data)
        return await self._token_request(request_data)

    async def _async_refresh_token(self, token: dict) -> dict:
        """Refresh a token."""

        new_token = await self._token_request(
            {
                "grant_type": "refresh_token",
                "client_id": self.client_id,
                "refresh_token": token["refresh_token"],
            }
        )

        return {**token, **new_token}

    async def _token_request(self, data: dict) -> dict:
        """Make a token request.

        Raises OAuth2TokenRequestError on token request failure.
        """
        session = async_get_clientsession(self.hass)

        data["client_id"] = self.client_id
        if self.client_secret:
            data["client_secret"] = self.client_secret

        _LOGGER.debug("Sending token request to %s", self.token_url)

        try:
            resp = await session.post(self.token_url, data=data)
            if resp.status >= 400: