building-multi-ui-apps

マルチ UI アプリの構築

前提条件

このスキルは myai の architecting-changes を拡張します。まずこれを読み込んでください。using-my-skills と engineering-principles は、myai のブートストラップで読み込み済みであることを前提とします。

再利用可能なコア、薄いアダプター、コンポジションルート、依存性注入といった一般的なアーキテクチャ原則については、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 を 1 つ構築し、その後に CLI／GUI／API アダプターを追加します。
• 現時点では最初にリリースしたインターフェースしかなくても、ビジネスルールはコマンド、ルート、ウィジェットに含めないようにします。
• プレゼンテーション層は入力を解析し、コアを呼び出して、出力を整形します。
• フレームワーク固有のリクエストオブジェクト、CLI コンテキスト、ウィジェットはエッジに留めます。

---

エントリーポイントパターン

シンプルに保ちます。すべての呼び出しでコマンドを指定する、Typer のマルチコマンドアプリを 1 つ使います。GUI を起動するコマンドも、CLI コマンドと同じリストに含める通常の Typer コマンドです。ルート／デフォルトコマンドも、モード検出ロジックも不要です。

これにより、次のようになります。

• 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 を直接開く 2 つ目のコンソールスクリプト（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 ファイル、ファイルマネージャーの「アプリケーションから開く」ハンドラー、ランチャー／Spotlight 風のランチャー、パッケージのデスクトップエントリーでは myapp-gui を使用してください。こうした連携では、実行すると GUI が開く単独のバイナリが想定されています。open／show-print-dialog コマンドを含む CLI の myapp は、スクリプトやターミナルでの使用向けとして引き続き使います。

---

ロジックの共有

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

シェルから見ると、myapp reply ... は引き続き単一のブロッキングコマンドです。進捗行をターミナルに表示し、求人への回答が必要な場合はプロンプトで入力を待ってから終了します。これはまさに CLI の「作法」です。インタラクションインターフェースを使えば、コアを汚さずに実現できます。

Qt MVVM アダプター — ViewModel がインターフェースを実装

Qt アダプターでは、MVVM の利点が発揮されます。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 と連携すると、3 つの不足点が現れます。いずれも ask()／制御ブリッジに集中しており、コアの境界は安定したままです。

**1. ユーザーによるキャンセル（ユーザー → コアの方向）。**図示したフローは一方向で、ユーザーは質問されたときにだけ回答します。実際の GUI では、ユーザーは実行途中で停止したいと考えます。適合する場合は組み込みのプリミティブを優先し、適合しない場合に限りアドホックなフラグを使用します。

• 組み込みの非同期キャンセル — フローを asyncio.Task として実行し、VM の「停止」ボタンから task.cancel() を呼び出します。次の await で CancelledError が発生し、クリーンアップ処理を経由して巻き戻されます。処理が本当に非同期である場合（ブロッキング処理がすべて await である場合）や、即座に完全停止したい場合に最適です。これは強制的な停止であり、コアは安全な境界で割り込めないため、「この項目を終えてから一時停止」ではなく、「ジョブ全体を今すぐ停止」に適しています。
• 協調的なフラグ／asyncio.Event — コアがステップ間で await should_abort()（または evt.is_set()）を確認し、正常な aborted 結果を返します。正常に、あるいはチェックポイントを経て停止したい場合（現在の求人を処理してから停止する場合）や、即時中断が危険な場合に適しています。

どちらを選ぶかは処理の実行方法と連動します。Task.cancel() が機能するのは、同じイベントループ内に await ポイントがある場合だけです。ブロッキング処理が ThreadPoolExecutor 上で実行されている場合、実行中のスレッドはキャンセルできません。ブロッキング呼び出しの間に確認する協調的なフラグを使うか、プロセスレベルでキャンセルしてください。

**2. 並行または動的な質問 — Future にキーを付ける。**この例では self._pending: Future[str] を 1 つだけ保持しているため、最初の質問が解決する前に 2 つ目の 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）です。意図的に OS 固有のプロジェクトにするのは例外です。各バックエンドは platformdirs（user_config_dir、user_data_dir、user_log_dir）を使用して、各 OS のディレクトリを扱います。

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

この状態では、マネージャーがすべてのプラットフォームを把握しています。新しい OS を追加するには、ビジネスロジックの編集が必要です。プラットフォーム固有のコードは、必ずインターフェース／プロトコル／抽象クラスの背後に隠し、マネージャーは抽象化レイヤーのみを呼び出すようにします。

起動時にバックエンドを選択します。

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

---

依存性の注入 — コンポジションルート

依存関係はコンストラクターのパラメーターで渡します。すべてを単一のコンポジションルート関数で結線します。DI ライブラリーは使いません。basedpyright の strict モードを破るか、不要な間接層を追加するためです。

パターン

# 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()

ルール

• ドメインクラスが独自にインフラストラクチャをインスタンス化してはいけません。 依存関係はコンストラクターを通じて渡します。
• アプリごとにコンポジションルートを 1 つ設けます。 オブジェクトグラフを把握するための唯一の場所です。
• テスト用の代替実装が必要となる依存関係には、プロトコル型のインターフェースを使います。
• テスト： フェイクを使って直接構築します。コンテナーの設定は不要です。

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 固有のコーディングルールです。