(
writer: Callable[[Path], None],
*,
reload_callback: ReloadCallback,
env_path: str | Path = ENV_PATH,
)
| 591 | |
| 592 | |
| 593 | async def _apply_env_change( |
| 594 | writer: Callable[[Path], None], |
| 595 | *, |
| 596 | reload_callback: ReloadCallback, |
| 597 | env_path: str | Path = ENV_PATH, |
| 598 | ) -> None: |
| 599 | path = Path(env_path) |
| 600 | had_existing_file = path.exists() |
| 601 | previous_content = read_env_content(path) if had_existing_file else "" |
| 602 | |
| 603 | try: |
| 604 | writer(path) |
| 605 | await reload_callback() |
| 606 | except Exception: |
| 607 | if had_existing_file: |
| 608 | path.write_text(previous_content, encoding="utf-8") |
| 609 | elif path.exists(): |
| 610 | path.unlink() |
| 611 | |
| 612 | try: |
| 613 | await reload_callback() |
| 614 | except Exception as restore_exc: |
| 615 | logger.warning(f"⚠️ 回滚配置后重新加载失败: {restore_exc}") |
| 616 | raise |
| 617 | |
| 618 | |
| 619 | async def save_form_config( |
no test coverage detected