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

Code before:
            "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. Raises OAuth2RefreshTokenFailed on failure."""
        try:
            new_token = await self._token_request(
                {
                    "grant_type": "refresh_token",
                    "client_id": self.client_id,
                    "refresh_token": token["refresh_token"],
                }
            )
        except ClientResponseError as err:
            # Store the warning messages in memory during Home Assistant runtime
            # This to prevent spamming for logs who are not ready, yet
            you_are_warned: set[str] = self.hass.data.setdefault(
                DATA_OAUTH2_REFRESHTOKEN_DEPRECATION_WARNING, set()
            )

            # Best what I could to think of is to use the clientID, unique enough to prevent flooding?
            # I don't have config_entry available. If correct, you have a client ID per config entry
            if self.client_id not in you_are_warned:
                _LOGGER.warning(
                    "Refresh token failed. Please catch 'OAuth2RefreshTokenFailed'. This will be added in version 2026.x.x and result in a breaking change"
                )
                you_are_warned.add(self.client_id)

            raise OAuth2RefreshTokenFailed(
                request_info=err.request_info,
                history=err.history,
                status=err.status,
                message=err.message,
                headers=err.headers,
            ) from err

        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)

Review comment (refactor):
I think we can simply use the `report_usage` helper. It will make sure that we don't log more than once per integration if we pass the integration domain.

https://github.com/home-assistant/core/blob/cc30add73ae99d33c747c9a65978daf4604bfecb/homeassistant/helpers/frame.py#L164-L185

Code after:
    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:
                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",
                    self.domain,
                    error_code,