class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle OAuth2 authentication."""
reauth_entry: ConfigEntry | None = None
async def async_step_reauth(self, user_input=None):
"""在 API 认证错误时执行 reauth。"""
self.reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(self, user_input=None):
"""通知用户需要 reauth 的对话框。"""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({}),
)
return await self.async_step_user()
async def async_oauth_create_entry(self, data: dict) -> dict:
"""为 reauth 创建 oauth config entry 或更新现有 entry。"""
if self.reauth_entry:
# 仅当 entry 被更新时才 reload
return self.async_update_reload_and_abort(
self.reauth_entry,
data=data,
reload_even_if_entry_is_unchanged=False,
)
return await super().async_oauth_create_entry(data)
