building-multi-ui-apps

构建多界面应用

前置条件

此技能扩展了 myai 的 architecting-changes。请先加载该技能。假定已通过 myai 引导程序加载 using-my-skills 和 engineering-principles。

有关通用架构原则（可复用核心、精简适配器、组合根、依赖注入），请参阅 myai 的 architecting-changes。此技能仅介绍 Python 特有的多界面模式：PySide6 GUI + typer CLI + FastAPI 共享领域核心、入口点路由、平台抽象和组合根装配。

UI 是一个插件。先构建可复用核心，再让每个界面都成为围绕核心的精简适配器。添加新界面（CLI、GUI、API）不应改变业务逻辑。

---

架构

Presentation Layer (top)
├── Qt GUI (PySide6)    - consumes domain, handles display
├── CLI (typer)          - consumes domain, handles terminal I/O
└── API (FastAPI)        - consumes domain, handles HTTP (if needed)
        |
        v
Domain Layer (middle)
├── Managers             - orchestrate operations
├── Models               - dataclasses, TypedDicts
└── Services             - business rules, pure logic
        |
        v
Utility Layer (bottom)
├── Helpers              - stateless functions
├── Wrappers             - typed third-party interfaces
└── Platform             - OS-specific implementations

**依赖只能向下流动。**领域层绝不导入表示层。

---

先构建可复用核心

• 先构建一个可组合的领域 API，再围绕它添加 CLI/GUI/API 适配器。
• 今天发布的第一个界面可能是唯一的界面；即便如此，也要让业务规则远离命令、路由和小组件。
• 表示层负责解析输入、调用核心并格式化输出。
• 框架专用的请求对象、CLI 上下文和小组件应留在边界层。

---

入口点模式

保持简单：使用一个 Typer 多命令应用，每次调用都要指定命令。启动 GUI 的命令只是普通的 Typer 命令，与 CLI 命令列在同一列表中。无需根命令/默认命令，也无需维护模式检测逻辑。

这样可以实现：

• myapp -h / myapp --help -> 列出所有命令的帮助信息
• myapp gui FILE [--debug] -> GUI 命令（显示窗口）
• myapp run foo / myapp config get key -> CLI 命令

有意不设默认/根命令：让不带参数的 myapp 执行特殊操作（例如打开 GUI），就必须使用模式检测启发式逻辑或 Typer/Click 默认命令变通方案（继承 typer.core.TyperGroup、修改 sys.argv），这些方案很脆弱，也容易受版本变化影响。如果需要 GUI 快捷方式，请注册显式的 gui 命令并记录说明——指定命令的代价是，不带参数的 myapp 会显示帮助信息，而不是启动 GUI。

# __main__.py
from __future__ import annotations

import typer

from myapp.bootstrap import create_services
from myapp.version import VERSION

app = typer.Typer(
    add_completion=False,
    no_args_is_help=True,
    context_settings={"help_option_names": ["-h", "--help"]},
)


@app.command()
def gui(path: str, debug: bool = False) -> None:
    """Open the GUI with this file."""
    services = create_services(debug=debug)

    from myapp.gui import run_gui  # noqa: PLC0415 - lazy so CLI runs stay Qt-free

    run_gui(services=services, path=path, debug=debug)


@app.command()
def run(item: str) -> None:
    """A plain CLI command."""
    services = create_services(debug=False)
    typer.echo(f"run {item}")


@app.command()
def version() -> None:
    """Print the version — a normal command, no --version magic needed."""
    typer.echo(VERSION)


if __name__ == "__main__":
    app()

说明：

• 在 GUI 命令的函数体内延迟导入 GUI，这样 run/config/version 就不会导入 Qt。open 命令解析参数、构建核心，然后在调用时执行 from myapp.gui import run_gui。
• 将依赖装配放在其他位置：命令解析输入并调用核心 / run_gui(...)；组合根（create_services）仍负责构建对象图并注入。绝不要在 __main__.py 中装配对象。
• GUI 子命令（myapp show-print-dialog FILE）只是更多路由到 GUI 可调用对象的 @app.command；通过 app.add_typer(...) 创建嵌套命令组的方式，与任何 Typer 应用完全相同。
• 帮助和版本信息都是常规命令，因此无需处理 --version 标志（也不会出现 Missing command 边界情况）。

提示：同时导出一个直接启动 GUI 的可执行文件

导出第二个控制台脚本，直接打开 GUI（myapp-gui），这样用户和桌面环境就无需输入 myapp open：

# __main__.py (add alongside `main`)
def main() -> None:
    app()  # full CLI, including the `open` GUI command


def main_gui() -> None:
    """Bare GUI entry point: `myapp-gui` just opens the window."""
    services = create_services(debug=False)

    from myapp.gui import run_gui  # noqa: PLC0415 - lazy: only this binary needs Qt

    run_gui(services=services, path=None, debug=False)

# pyproject.toml
[project.scripts]
myapp = "myapp.__main__:main"        # full CLI (contains the `open` GUI command)
myapp-gui = "myapp.__main__:main_gui"  # dedicated GUI entry, for launchers/desktop

在 .desktop 文件、文件管理器的“打开方式”处理程序、类似 Launchpad/Spotlight 的启动器，以及打包的桌面条目中使用 myapp-gui——这些集成需要的是一个直接打开 GUI 的可执行文件。CLI myapp（包含 open/show-print-dialog 命令）仍用于脚本和终端操作。

---

共享逻辑

GUI 和 CLI 都使用同一个管理器：

# CLI
def cmd_create(name: str) -> int:
    result = manager.create_profile(name)
    if result.is_err:
        print(f"Error: {result.unwrap_err()}", file=sys.stderr)
        return 1
    print(f"Created: {result.unwrap().name}")
    return 0

# GUI
def on_create_clicked(self) -> None:
    result = self._manager.create_profile(name)
    if result.is_err:
        self._show_error(result.unwrap_err())
        return
    self._refresh_list()

---

交互式／长时间运行的操作

上面的 Shared Logic 示例是一个一次性命令：() -> Result[T, E] ——调用核心逻辑，获得结果，完成。CLI 和 GUI 都可以轻松处理（CLI：print(result)；GUI：刷新视图）。

交互式操作则不同。它们会运行一段时间，持续发出进度／错误信息，还可能在运行过程中向用户提问（例如：“这个职位要求我们在提交前回答一个问题——你怎么回答？”）。GUI（Qt MVVM）可以通过向外发送信号、向内显示对话框来轻松处理。CLI 也可以，但前提是我们不围绕信号来设计核心逻辑。

核心原则：MVVM（信号、@Slot、绑定）是 Qt 适配器实现操作的方式，绝不是 核心逻辑的形态。核心逻辑不得导入 Qt，不得了解信号，也不得了解 typer。应通过一个精简的交互接口来表达操作——一个用于发送事件／错误的输出通道，以及一个用于提问的输入通道——并让每个适配器将其映射到自己的惯用方式。

交互接口（核心层，与框架无关）

# core/interaction.py  — no Qt, no typer imports
from dataclasses import dataclass
from typing import Protocol

@dataclass(frozen=True)
class ProgressEvent:
    vacancy_id: str
    message: str

@dataclass(frozen=True)
class Question:
    vacancy_id: str
    text: str

class ReplyFlowIO(Protocol):
    """The only thing the core knows about its user."""
    async def progress(self, event: ProgressEvent) -> None: ...
    async def error(self, error: str) -> None: ...
    async def ask(self, question: Question) -> str: ...  # returns the user's answer

async def run_reply_flow(vacancies: list[Vacancy], io: ReplyFlowIO) -> ReplyResult:
    """The interactive flow. UI-agnostic: only negotiates through `io`."""
    for v in vacancies:
        await io.progress(ProgressEvent(v.id, f"checking {v.title}"))
        answer: str | None = None
        if v.requires_answer:
            answer = await io.ask(Question(v.id, v.text))
            if not answer:
                await io.progress(ProgressEvent(v.id, "skipped"))
                continue
        result = await submit_reply(v, answer)
        if result.is_err:
            await io.error(result.unwrap_err())
            continue
        await io.progress(ProgressEvent(v.id, "replied"))
    return finalize_reply(vacancies)

整个流程是一个基于该协议的普通 async 函数。核心逻辑决定做什么，而不是如何显示事件或如何回答问题。

CLI 适配器——运行时打印并提示

# cli.py
import asyncio
import typer

class CliReplyIO:
    async def progress(self, event: ProgressEvent) -> None:
        typer.echo(f"[{event.vacancy_id}] {event.message}")

    async def error(self, error: str) -> None:
        typer.echo(f"error: {error}", err=True)

    async def ask(self, question: Question) -> str:
        return typer.prompt(question.text)

@app.command()
def reply(vacancies: list[Path]) -> int:
    """One command that happens to be interactive underneath."""
    result = asyncio.run(run_reply_flow(load_vacancies(vacancies), CliReplyIO()))
    typer.echo(f"done: {result}")
    return 0

从 shell 的角度看，myapp reply ... 仍是一个会阻塞的命令：它会向终端输出进度行，在职位需要回答时等待用户输入，然后退出。这正是 CLI 的“运作方式”——借助交互接口，我们无需污染核心逻辑，就能实现这一点。

Qt MVVM 适配器——由 ViewModel 实现该接口

MVVM 在 Qt 适配器中发挥作用。ViewModel 实现相同的 ReplyFlowIO；它是核心逻辑与 QML 之间唯一的桥梁。进度／错误会变成 VM 状态和信号，供 QML 绑定；ask() 会将控制权交给 QML 对话框，并通过一个待处理的 asyncio.Future 等待回答。

# gui/reply_viewmodel.py
from __future__ import annotations

import asyncio
from PySide6.QtCore import QObject, Signal, Slot

from core.interaction import ProgressEvent, Question, ReplyFlowIO

class ReplyViewModel(QObject, ReplyFlowIO):
    progress_message = Signal(str)     # -> QML updates a status line / list row
    error_message = Signal(str)        # -> QML shows an error block
    question_requested = Signal(str)   # -> QML opens a prompt dialog with `text`
    reply_finished = Signal(object)    # -> QML knows the flow ended

    def __init__(self) -> None:
        super().__init__()
        self._pending: asyncio.Future[str] | None = None

    # --- ReplyFlowIO: turn core events into VM state + signals ---
    async def progress(self, event: ProgressEvent) -> None:
        self.progress_message.emit(f"[{event.vacancy_id}] {event.message}")

    async def error(self, error: str) -> None:
        self.error_message.emit(error)

    async def ask(self, question: Question) -> str:
        # Defer to the view: emit, present a dialog in QML,
        # and sleep until QML calls back with the answer.
        self._pending = asyncio.get_running_loop().create_future()
        self.question_requested.emit(question.text)
        return await self._pending

    # --- Called by QML when the prompt dialog is answered ---
    @Slot(str)
    def submit_answer(self, text: str) -> None:
        if self._pending is not None and not self._pending.done():
            self._pending.set_result(text)

在 QML 中，question_requested 信号会打开对话框，用户确认后会调用 submit_answer——对话框纯粹负责视图，异步流程则会在收到返回的字符串后继续。线程说明：这些回调在 GUI 线程上运行（qasync 循环也在此运行），因此 set_result 可以安全地恢复协程。如果流程执行真正的阻塞式 I/O，则应在工作线程／ThreadPoolExecutor 中运行，并通过信号将结果转回——但核心接口的形态不受影响。

按约定异步

交互接口有意采用 async。GUI 必须 await 对话框，因此需要协程。CLI 实现相同的异步协议，只需通过 asyncio.run(...) 运行即可——CLI 保留阻塞行为，GUI 保留非阻塞事件循环，两者共用一个核心函数。同步的一次性命令继续使用 Result；交互式操作则通过 …IO 接口实现。

**总结：**基本命令返回 Result。交互式操作通过注入的 …IO 接口协商（向外发送事件，向内接收问题）。MVVM 只是 Qt 适配器实现该接口的方式，绝不是核心逻辑的形态。

扩展这一模式

以上示例是起步形态，而非上限。在实际的 Qt 互操作中，会出现三个缺口，而且它们都集中在 ask()/控制桥接上——核心边界保持稳定。

1. 用户发起取消（用户 → 核心方向）。 如图所示，流程是单向的：用户只会在收到提问时作答。真实的 GUI 用户希望能在运行中途停止。适用时优先使用内置原语，不适用时再退而使用临时标志：

• 内置异步取消——将流程作为 asyncio.Task 运行，并让 VM 的“停止”按钮调用 task.cancel()；CancelledError 会在下一个 await 处触发，并逐层展开执行清理。适用于操作确实是异步的情况（每个阻塞步骤都是一次 await），且你希望立即强制停止。此方式较为突然——核心无法在安全边界处拦截——因此适合“现在就停止整个任务”，而不适合“完成当前项目后暂停”。
• 协作式标志 / asyncio.Event——核心在步骤之间检查 await should_abort()（或 evt.is_set()），并返回干净的 aborted 结果。适用于优雅停止或检查点式停止（完成当前职位后再停止），或者即时中断可能不安全的情况。

选择哪种方式取决于工作如何运行：Task.cancel() 仅在同一事件循环中存在 await 点时才有效。如果阻塞工作运行在 ThreadPoolExecutor 中，正在运行的线程无法取消——应使用在阻塞调用之间检查的协作式标志，或在进程级别取消。

2. 并发 / 动态提问——为 future 设置键。 示例只保留一个 self._pending: Future[str]，如果第一个 ask 尚未解决时又触发第二个，它就会被覆盖。实际应用会并发处理任务，或运行下一个问题取决于上一个答案的向导。应按问题 ID 为 future 设置键（或将它们排队），这样每个 ask 都会等待自己的 future；并将整个 Question 传给 QML，而不是只传一个字符串：

self._pending: dict[str, asyncio.Future[str]] = {}

async def ask(self, question: Question) -> str:
    fut = asyncio.get_running_loop().create_future()
    self._pending[question.id] = fut
    self.question_requested.emit(question)
    return await fut

@Slot(str, str)
def submit_answer(self, question_id: str, text: str) -> None:
    fut = self._pending.pop(question_id, None)
    if fut is not None and not fut.done():
        fut.set_result(text)

---

平台抽象

适用于必须在多个平台上运行的应用。默认采用跨平台方式（Windows/macOS/Linux）；有意针对特定操作系统的项目则属于例外。每个后端都使用 platformdirs（user_config_dir、user_data_dir、user_log_dir）获取其操作系统对应的目录。

from abc import ABC, abstractmethod

class PlatformBackend(ABC):
    @abstractmethod
    async def start_instance(self, profile: Profile, binary: Path) -> Result[int, str]: ...

    @abstractmethod
    def get_data_dir(self) -> Path: ...

    @abstractmethod
    def get_config_dir(self) -> Path: ...

class LinuxBackend(PlatformBackend):
    async def start_instance(self, profile: Profile, binary: Path) -> Result[int, str]:
        env = {
            "XDG_CONFIG_HOME": str(profile.path / "config"),
            "XDG_DATA_HOME": str(profile.path / "data"),
        }
        process = await asyncio.create_subprocess_exec(
            str(binary), "-many", "-workdir", str(profile.path),
            env={**os.environ, **env},
        )
        return Ok(process.pid) if process.pid else Err("Failed to start")

不要这样做——在平台抽象层中直接通过条件语句调用特定平台的代码：

# ❌ WRONG: NotificationsManager directly branches on platform
class NotificationsManager:
    def send(self, message: str) -> None:
        if sys.platform == "linux":
            linux_backend.run(message)          # direct call, no interface
        elif sys.platform == "darwin":
            macos_backend.notify(message)       # direct call, no interface
        else:
            windows_backend.toast(message)      # direct call, no interface

现在管理器了解每个平台。新增操作系统意味着要编辑业务逻辑。平台代码必须隐藏在接口/协议/抽象类之后；管理器只调用抽象层。

在启动时选择后端：

def get_backend() -> PlatformBackend:
    match sys.platform:
        case "linux":
            return LinuxBackend()
        case _:
            raise NotImplementedError(f"Unsupported platform: {sys.platform}")

---

依赖注入——组合根

通过构造函数参数传递依赖。在单个组合根函数中完成所有组件的装配。不使用 DI 库——它们会破坏 basedpyright 严格模式，或引入不必要的间接层。

模式

# app/bootstrap.py
def create_domain(config: AppConfig) -> SessionManager:
    """Composition root — the ONLY place dependencies are wired."""
    db = DatabaseWrapper(config.db_path)
    api = ApiClientWrapper(config.api_url, config.api_key)
    auth = AuthService(api_client=api)
    sync = SyncService(db=db, api_client=api)
    return SessionManager(auth=auth, sync=sync)

# GUI entry point
def main_gui() -> None:
    config = load_config()
    session = create_domain(config)
    window = MainWindow(session=session)
    ...

# CLI entry point
def main_cli() -> None:
    config = load_config()
    session = create_domain(config)
    cli_app = build_typer_app(session=session)
    cli_app()

规则

• **领域类绝不自行实例化基础设施。**依赖通过构造函数传入。
• **每个应用只有一个组合根。**这是了解对象图的唯一位置。
• 对可能需要测试替身的依赖使用基于协议类型的接口。
• **测试：**直接使用伪对象构造——无需设置容器：

def test_sync_handles_conflict() -> None:
    db = FakeDatabaseWrapper()
    api = FakeApiClient(responses=[CONFLICT_RESPONSE])
    sync = SyncService(db=db, api_client=api)
    result = sync.pull_changes()
    assert result.is_err

---

其他表示层

可以添加 FastAPI 作为另一个使用相同领域层的表示层：

@router.post("/profiles")
async def create_profile(req: CreateProfileRequest) -> ProfileResponse:
    result = manager.create_profile(req.name)
    if result.is_err:
        raise HTTPException(400, result.unwrap_err())
    return ProfileResponse.from_domain(result.unwrap())

在特定情况下也可以使用其他表示层：TUI、可导出的 Python API

---

相关 myai 技能

• architecting-changes — 父级技能。语言无关的架构决策框架：可复用核心、精简适配器、组合根。
• engineering-principles — 语言无关的理念：架构分离、将 UI 作为插件。
• architecting-python-changes — Python 专用的架构路由器。
• building-qt-apps — Python 专用的 PySide6 GUI 模式。
• building-backends（myai）— 添加 API 层时使用的后端架构模式。
• writing-python-code — 共享领域核心的 Python 专用编码规则。