writing-python-code

Python コードの作成

前提条件

このスキルは myai の engineering-principles を拡張します。最初にそれを読み込んでください。using-my-skills と engineering-principles は、myai のブートストラップによってすでに読み込まれているものとします。

一般的なエンジニアリング哲学（成功への近道、早期の失敗、制御フローとしてのエラー処理、テスト哲学、アーキテクチャの分離）については、myai の engineering-principles を参照してください。このスキルで扱うのは、Python 固有のツール、型システムの設定、パターンのみです。

プロジェクトローカルのツールは、システムにインストールされたバイナリではなく uv を通じて実行します：uv run python、uv run pytest、uv run ruff、uv run basedpyright、uv run poe、uv run pre-commit。通常の検証では、uv run poe lint_full と uv run poe test の組み合わせを使用します。

---

型システム

basedpyright の設定

[tool.pyright]
pythonVersion = "3.14"
typeCheckingMode = "strict"
reportAny = "error"
reportImportCycles = "error"
reportImplicitStringConcatenation = "none"
reportUnusedCallResult = "none"
reportUnnecessaryIsInstance = "none"

追加の厳格オプション（中規模から大規模のプロジェクト向け）：

reportExplicitAny = "error"
reportUnnecessaryTypeIgnoreComment = "error"
reportMissingModuleSource = "error"
reportPrivateUsage = "error"
reportOptionalMemberAccess = "error"
reportOptionalCall = "error"
reportAttributeAccessIssue = "error"

使用禁止のパターン

使用禁止	代わりに使用
Any	完全に禁止 — 実際の型を定義します
object（制限なし）	境界部分でのみ許可されます（下記参照）
typing.cast()	isinstance、TypeIs、パターンマッチング
根拠のない # type: ignore	# type: ignore[specific-code] # rationale: <reason>
ビジネスロジック内の生の dict	msgspec.Struct、dataclass、または TypedDict
暗黙的な戻り値の型	すべての関数に明示的な注釈を付けます

object が許可されるのは次の場合のみです： TypeIs/TypeGuard のパラメーター、*args: object、Coroutine[object, None, T]、PySide6 の Signal(object)。それ以外では、Protocol、TypeVar、ユニオン型、または型付き構造体を使用してください。

型のパターン

外部データ → msgspec.Struct：

import msgspec

class UserConfig(msgspec.Struct):
    name: str
    port: int
    debug: bool = False

# JSON → typed object (validates at decode time)
config = msgspec.json.decode(raw_bytes, type=UserConfig)

# Dict/YAML → typed object
config = msgspec.convert(raw_dict, type=UserConfig)

辞書との互換性が必要な場合（例：**unpacking、辞書を期待する API）には、TypedDict も引き続き有効です。

FastAPI の HTTP エッジ → トランスポート境界では pydantic DTO を使用します。直ちにフレームワークに依存しない型付き構造体に変換し、フレームワークモデルをドメインモデルにしないでください。

ドメインオブジェクト → dataclass：

@dataclass(frozen=True, slots=True)
class Profile:
    name: str
    version: str
    active: bool = True

ダックタイピング → Protocol：

class Renderable(Protocol):
    def render(self) -> str: ...

定数 → Final：

import platformdirs

MAX_RETRIES: Final = 3
CONFIG_PATH: Final[Path] = Path(platformdirs.user_config_dir("app"))

ライブラリ境界での Any の扱い

0. 型付きデシリアライズ（外部データ向け）：

msgspec.json.decode(data, type=MyStruct) は JSON/API レスポンスの Any を解消します。戻り値の型は Any ではなく MyStruct です。境界でのデータのデシリアライズが目的であれば、ラッパーを検討する前にこれを使用してください。

1. 型付きラッパー（ライブラリ API では推奨）：

class WhisperModelWrapper:
    def __init__(self, model_size: str, device: str = "auto") -> None:
        from faster_whisper import WhisperModel as _WhisperModel
        self._model = _WhisperModel(model_size, device=device)

    def transcribe(self, audio: np.ndarray, language: str | None = None) -> TranscriptionResult:
        segments_gen, info = self._model.transcribe(audio, language=language)
        return TranscriptionResult(
            text="".join(s.text for s in segments_gen),
            language=str(info.language),
        )

ruff でラッパーの使用を強制します：

[tool.ruff.lint.flake8-tidy-imports.banned-api]
"faster_whisper" = { msg = "Use src/wrappers/whisper_wrapper instead" }

2. 型スタブを src/stubs/ に配置します：

# src/stubs/some_library.pyi
def some_function(arg: str) -> list[int]: ...

basedpyright の設定で stubPath = "src/stubs" を指定します。

3. 一度限りのケース向けのインライン型絞り込み：

raw_value = untyped_lib.get_value()  # returns Any
assert isinstance(raw_value, str)    # narrows to str

TypeIs ガード

注： msgspec でデコードされたデータはすでに完全に型付けされているため、TypeIs ガードは不要です。型が不明なインメモリーオブジェクトの型を絞り込む場合に使用してください。

from typing import TypeIs, TypedDict, Required

class ValidResponse(TypedDict):
    status: Required[str]
    data: Required[dict[str, str | int | bool | list[str]]]

# Simplified for brevity — production code should use msgspec.convert() for full validation
def is_valid_response(obj: object) -> TypeIs[ValidResponse]:
    return (
        isinstance(obj, dict)
        and isinstance(obj.get("status"), str)
        and isinstance(obj.get("data"), dict)
    )

def process(response: object) -> Result[str, str]:
    if not is_valid_response(response):
        return Err("Invalid response format")
    return Ok(response["data"]["key"])  # type-safe access

型安全のためのパターンマッチング

@dataclass
class Success:
    value: str

@dataclass
class Failure:
    error: str
    code: int

type Outcome = Success | Failure

def handle(outcome: Outcome) -> str:
    match outcome:
        case Success(value=v):
            return f"OK: {v}"
        case Failure(error=e, code=c):
            return f"Error {c}: {e}"

# type: ignore の方針

すべての # type: ignore に、具体的なエラーコードと根拠を記述する必要があります：

# BAD
result = some_call()  # type: ignore

# GOOD
result = some_call()  # type: ignore[no-any-return]  # rationale: lib returns Any, validated below
assert isinstance(result, ExpectedType)

TYPE_CHECKING ガード

循環依存を引き起こすインポートや、型注釈にのみ必要なインポートの場合：

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from src.managers.audio_manager import AudioManager

class TranscriptionService:
    def __init__(self, audio: "AudioManager") -> None: ...

注： TYPE_CHECKING は同じレイヤー内での前方参照に使用します。2つのモジュールが互いにインポートしている場合、それは循環依存です。TYPE_CHECKING で循環を隠すのではなく、アーキテクチャを修正してください（Protocol を抽出する、型を共通モジュールに移動するなど）。

---

エラー処理

制御フローとしてのエラー処理に関する一般的な原則については、myai の engineering-principles を参照してください。 Python 固有の方針：想定される失敗には rusty-results の Result[T, E] を使用します。例外はプログラミングエラー（バグ）に限って使用します。

判断表

状況	使用するもの
ファイルが見つからない、ネットワークエラー、無効な入力	Result[T, E]
ユーザーが無効なデータを提供した	Result[T, E]
サードパーティライブラリで例外が発生した	境界で捕捉 → Result[T, E]
不変条件に違反した（本来起こるべきではない）	raise で例外を送出
無効なプログラム状態（バグ）	raise で例外を送出

パターン

import msgspec
from rusty_results import Result, Ok, Err

class Config(msgspec.Struct):
    name: str

def load_config(path: Path) -> Result[Config, str]:
    if not path.exists():
        return Err(f"Config not found: {path}")
    try:
        return Ok(msgspec.json.decode(path.read_bytes(), type=Config))
    except (msgspec.DecodeError, OSError) as e:
        return Err(f"Failed to load: {e}")

ルール

1. 早期リターン — まずエラーを処理し、成功パスは直線的に保ちます：

   result = load_data(path)
   if result.is_err:
       return Err(f"Cannot proceed: {result.unwrap_err()}")
   data = result.unwrap()
   

2. 3 つのエラー境界：

   • ライブラリ：サードパーティの例外をキャッチし、Result に変換します
   • コンポーネント：各サブシステムは呼び出し元に Result を返します
   • グローバル：UI／CLI の最上位であらゆるエラーをキャッチし、ユーザーにメッセージを表示します

3. エラーを決して握りつぶさない — except: pass を使わず、Result を無視しないでください。すべての Result[T, E] を確認し、最低限、GUI ではエラーをログに記録してトースト／アラートを表示し、CLI ではエラーをログに記録して出力する必要があります。

4. 失敗時のクリーンアップ — 複数ステップの処理が途中で失敗した場合は、部分的な状態をクリーンアップします

5. 複雑なドメインにはカスタムエラー型を使用します：

   @dataclass
   class ConfigError:
       path: Path
       reason: str
       line: int | None = None
   

6. 受け取ったエラー値は適切に処理します：UI 警告の表示、ログへの記録、早期リターン、またはコンテキストを添えた伝播を行います

非同期タスクの境界

asyncio.ensure_future() または create_task() で起動したコルーチンは、ファイア・アンド・フォーゲットの境界です。例外が外に漏れると、誰もその例外を取得できず、UI が中間状態（例：「処理中」のまま）で停止します。

必須のパターン： 安全策として、コルーチン全体を try/except Exception で囲みます：

async def _do_work(self, path: Path) -> None:
    try:
        await self._do_work_inner(path)
    except Exception as exc:
        logger.exception("Unexpected error for %s", path)
        self._set_error_state(path, f"Unexpected error: {exc}")

内側のメソッドでは、想定されるエラー（Result の確認、特定の例外）を処理します。外側のメソッドでは、UI が必ず終了状態に遷移するようにします。

CLI のエラー境界

def main() -> int:
    result = run_app()
    if result.is_err:
        typer.echo(f"Error: {result.unwrap_err()}", err=True)
        return 1
    return 0

---

非同期パターン

使用する場面

非同期を使用する	非同期を使用しない
ファイル I/O	純粋なデータ変換
ネットワークリクエスト	単純な計算
サブプロセスの実行	メモリ内操作
100 ミリ秒を超える処理	高速な検索
並列 I/O 操作	順次実行する純粋なロジック

HTTP リクエスト

async def fetch_data(url: str) -> Result[bytes, str]:
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(url, timeout=30.0)
            resp.raise_for_status()
            return Ok(resp.content)
    except httpx.HTTPError as e:
        return Err(f"HTTP error: {e}")

サブプロセスの実行

タイムアウト、強制終了、出力の確実な復旧などの高度なケースについては、python-reliable-subprocess-handling を参照してください。 asyncio.timeout() 内で communicate() を実行すると、キャンセル時にすでに読み出されたバイト列が破棄されます。また、kill() は子プロセスの子プロセスを終了しません。これらについては、そちらで説明しています。以下のパターンが安全なのは、単純かつ高速で、タイムアウトが不要な場合に限られます。

async def run_command(args: list[str]) -> Result[str, str]:
    try:
        process = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await process.communicate()
        if process.returncode != 0:
            return Err(f"Command failed: {stderr.decode()}")
        return Ok(stdout.decode())
    except FileNotFoundError:
        return Err(f"Command not found: {args[0]}")

並行処理

async def fetch_all(urls: list[str]) -> list[Result[bytes, str]]:
    return await asyncio.gather(*(fetch_data(url) for url in urls))

タイムアウト

async def fetch_with_timeout(url: str) -> Result[bytes, str]:
    try:
        async with asyncio.timeout(10):
            return await fetch_data(url)
    except TimeoutError:
        return Err(f"Timeout fetching {url}")

非同期処理のルール

1. 非同期コンテキスト内で subprocess.run()、time.sleep()、または同期 HTTP を決して呼び出さないでください
2. サブプロセスの呼び出しで shell=True を決して使用しないでください
3. サブプロセスには必ず asyncio.create_subprocess_exec() を使用してください
4. リソース管理（HTTP クライアント、ファイルハンドル）には必ず async with を使用してください
5. 同じ呼び出しチェーン内で同期処理と非同期処理を決して混在させないでください
6. 停止する可能性がある処理では、必ず TimeoutError を処理してください

---

コードスタイル

ルール	値
行の長さ	120
インデント	4 つのスペース
引用符	二重引用符（エスケープを避ける場合のみ一重引用符）
インポート順	標準ライブラリ → サードパーティ → ローカル（ruff で自動ソート）

命名規則

要素	規約	例
モジュール、関数、変数	snake_case	load_config、user_name
クラス、TypedDict	PascalCase	ProfileManager、UserConfig
定数	UPPER_SNAKE	MAX_RETRIES、DEFAULT_PORT
プライベート	_prefix	_internal_state

ドキュメント

公開 API には Google スタイルの docstring を付けます。コメントでは何をするかではなく、なぜそうするかを説明します。

---

クロスプラットフォーム

成熟したプロジェクトは、Windows、macOS、Linux を等しく対象にします。意図的に特定の OS に限定したプロジェクトは例外です。スタンドアロンの小規模なスクリプトは、特定の OS に限定したままでもかまいません。

• パスには pathlib を使用します。ユーザーディレクトリには platformdirs の user_config_dir、user_data_dir、user_cache_dir、user_log_dir、user_state_dir を使用します。
• POSIX 専用のモジュールやシステムコール（os.killpg、pwd、grp、fcntl、resource、signal.SIGKILL）は使用しません。
• OS による小さな違いには、限定的な sys.platform／os.name 分岐を使用します。複雑な OS 固有のサブシステムには、OS ごとに 1 つの実装を持つ Protocol を使い、起動時に選択します（building-multi-ui-apps を参照）。ビジネスロジック内でプラットフォームによる分岐は行いません。
• start_new_session と os.killpg を使ったサブプロセスの終了処理は POSIX 専用です。POSIX を対象とする場合にのみ使用してください。クロスプラットフォームアプリには、OS ごとの終了処理の実装が必要です。

---

前提条件と検証

一般的なフェイルファストの原則については、myai の engineering-principles を参照してください。 Python 固有の方法として、サブシステムのエントリーポイントで msgspec.Struct を使ってデータ形状を、isinstance／TypeIs を使って型の絞り込みを検証し、想定される検証エラーには Result[T,E] を使用します。ビジネスロジックに進む前に、権限、外部依存関係、設定の妥当性、入力範囲を確認してください。

---

アーキテクチャ

一般的なアーキテクチャ分離の原則については、myai の engineering-principles と architecting-changes を参照してください。 Python 固有のレイアウト：

Presentation (Qt GUI / CLI / API)
        |
        v
Domain (Managers, Models, Business Rules)
        |
        v
Utilities (Helpers, Wrappers, Common)

• 依存関係は下方向のみに流れます
• UI はプラグインです。CLI、GUI、API を追加しても、ビジネスロジックを変更する必要はありません
• ドメイン層からプレゼンテーション層をインポートすることはありません
• アーキテクチャ上の判断については、myai の architecting-changes を拡張した architecting-python-changes を読み込みます

---

セキュリティ

• サブプロセスの呼び出しで shell=True を使用しません
• パスを検証します（シンボリックリンクを考慮）：

  def is_safe_path(path: Path, base: Path) -> bool:
      try:
          resolved = path.resolve(strict=True)
          return resolved.is_relative_to(base.resolve(strict=True))
      except (OSError, ValueError):
          return False
  

• シークレットをハードコードしません。環境変数または dotenv を使用します
• システム境界で入力を検証します
• ユーザー入力をサブプロセスのコマンドに埋め込みません
• 失敗時にクリーンアップします。複数の手順からなる操作が途中で失敗した場合は、部分的に残った状態をクリーンアップします

---

パフォーマンス

• まずコードを明確にし、プロファイリングデータがある場合にのみ最適化します
• 頻繁にインスタンス化されるデータクラスでは __slots__ を使用します
• I/O 操作をまとめます（ループ内でファイルを 1 つずつ読み込まないでください）
• コストのかかるリソースは遅延読み込みにします（インポート時ではなく、初回使用時に読み込みます）
• asyncio.gather() で I/O を並行処理します

---

ロギング

import colorlog

def get_logger(name: str) -> logging.Logger:
    handler = colorlog.StreamHandler()
    handler.setFormatter(colorlog.ColoredFormatter(
        "%(log_color)s%(levelname)-8s%(reset)s %(name)s: %(message)s"
    ))
    logger = logging.getLogger(name)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger

使用方法：logger = get_logger(__name__)

---

Jinja2 テンプレート

テキスト出力（HTML、設定ファイル、レポート、Markdown）の生成には、次を使用します：

---

ツール

ツール	用途
uv	パッケージ管理、スクリプト実行
basedpyright	型チェック（厳格）
ruff	厳格な lint とフォーマット
pytest	テスト
rusty-results	Result[T, E] パターン
typer	CLI フレームワーク（推奨）。必ず -h のサポートを設定します（下記を参照）
argparse	標準ライブラリのみを使うスクリプト専用の CLI
PySide6	GUI（システム依存関係なし）
httpx	HTTP（非同期）
msgspec	外部データの検証と解析
platformdirs	OS ごとのユーザー設定／データ／キャッシュ／ログ用ディレクトリ（クロスプラットフォームでのデフォルト）
Jinja2	テキスト出力の生成

ヘルプ用の -h は必ず有効にしてください。typer はデフォルトでは --help のみをサポートします：

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

uv run poe lint_full を継続的に実行してください。最後に実行するだけでは不十分です。basedpyright、Ruff のチェック／フォーマット、カスタムリンターをまとめて確認する主要な手段として扱い、検証には uv run poe test も併用してください。

関連する myai スキル

• engineering-principles — 親スキルです。言語に依存しない理念：成功への道筋、迅速な失敗、制御フローとしてのエラー処理、テストの理念、アーキテクチャの分離。
• python-reliable-subprocess-handling — 上記の非同期サブプロセスのルールをさらに詳しく説明します：タイムアウトに安全な出力のキャプチャ、孤立した子プロセスとパイプ、テストダブル。
• architecting-changes — コードを書く前にアーキテクチャ上の決定を行う場合に使用します。Python 固有の拡張については、architecting-python-changes を読み込みます。
• code-simplification — Python コードをリファクタリングして明確にする場合に使用します。