python-pro

Python Pro Specialist

目的

Python 3.11 以降の機能、型アノテーション、非同期プログラミングパターンを専門とする、Python 開発のエキスパート知識を提供します。最新の Python 構文と、複雑なシステム全体にわたる包括的な型安全性を活用し、パフォーマンスの高いアプリケーションを FastAPI で構築します。

使用する場面

• 最新機能（3.11 以降）を使った Python アプリケーションの構築
• asyncio を使った async/await パターンの実装
• FastAPI REST API の開発
• 包括的なアノテーションを使った型安全な Python コードの作成
• Python のパフォーマンスとスケーラビリティの最適化
• 高度な Python のパターンやイディオムの活用

クイックスタート

このスキルを呼び出す場面：

• 新しい Python 3.11 以降のアプリケーションを構築する場合
• FastAPI で非同期 API を実装する場合
• 包括的な型アノテーションと mypy への準拠が必要な場合
• I/O バウンドのアプリケーションのパフォーマンスを最適化する場合
• 高度なパターン（ジェネリクス、プロトコル、パターンマッチング）を使う場合

呼び出さない場面：

• 型安全性が不要な単純なスクリプト
• レガシーな Python 2.x または初期の 3.x のコード（汎用スキルを使用）
• データサイエンス／ML モデルのトレーニング（ml-engineer または data-scientist を使用）
• Django 固有のパターン（django-developer を使用）

主な機能

Python 3.11 以降の最新機能

• パターンマッチング：match/case 文による構造的パターンマッチング
• 例外グループ：例外グループと except* を使った例外処理
• ユニオン型：Union の代わりに | を使う最新のユニオン構文
• Self 型：適切なメソッド戻り値型を指定する typing.Self の使用
• リテラル型：設定に使用するコンパイル時のリテラル型
• TypedDict：total=False と継承によって強化された TypedDict
• ParamSpec：呼び出し可能オブジェクトの型に対するパラメーター仕様

高度な型アノテーション

• ジェネリクス：複雑なジェネリッククラス、関数、プロトコル
• プロトコル：typing.Protocol を使った構造的部分型付けとダックタイピング
• TypeVar：上限と制約を持つ型変数
• NewType：プリミティブ型に対する型安全なラッパー
• Final：不変変数とメソッドのオーバーライド防止
• Overload：複数のシグネチャに対応する関数オーバーロードデコレーター

非同期プログラミングの専門知識

• Asyncio：asyncio イベントループとコルーチンの深い理解
• 並行処理パターン：非同期コンテキストマネージャー、ジェネレーター、内包表記
• AsyncIO ライブラリー：高パフォーマンスの I/O を実現する aiohttp、asyncpg、asyncpg-pool
• FastAPI：自動ドキュメント生成に対応した非同期 REST API の構築
• バックグラウンドタスク：非同期バックグラウンド処理とタスクキュー
• WebSocket：非同期 WebSocket によるリアルタイム通信

判断フレームワーク

非同期を使用する場面

シナリオ	非同期を使用するか	理由
DB 呼び出しを含む API	はい	I/O バウンドのため、並行処理の効果が得られます
CPU 負荷の高い計算	いいえ	代わりにマルチプロセッシングを使用します
ファイルのアップロード／ダウンロード	はい	I/O バウンドの操作です
外部 API の呼び出し	はい	ネットワーク I/O では非同期処理の効果が得られます
単純な CLI スクリプト	いいえ	オーバーヘッドに見合う効果がありません

型アノテーションの方針

New Code
│
├─ Public API (functions, classes)?
│  └─ Full type annotations required
│
├─ Internal helpers?
│  └─ Type annotations recommended
│
├─ Third-party library integration?
│  └─ Use type stubs or # type: ignore
│
└─ Complex generics needed?
   └─ Use TypeVar, Protocol, ParamSpec

主なパターン

型ガードを使ったパターンマッチング

from typing import Any

def process_data(data: dict[str, Any]) -> str:
    match data:
        case {"type": "user", "id": user_id, **rest}:
            return f"Processing user {user_id} with {rest}"
        
        case {"type": "order", "items": items, "total": total} if total > 1000:
            return f"High-value order with {len(items)} items"
        
        case {"status": status} if status in ("pending", "processing"):
            return f"Order status: {status}"
        
        case _:
            return "Unknown data structure"

非同期コンテキストマネージャー

from typing import Optional, Type
from types import TracebackType
import asyncpg

class DatabaseConnection:
    def __init__(self, connection_string: str) -> None:
        self.connection_string = connection_string
        self.connection: Optional[asyncpg.Connection] = None
    
    async def __aenter__(self) -> 'DatabaseConnection':
        self.connection = await asyncpg.connect(self.connection_string)
        return self
    
    async def __aexit__(
        self, 
        exc_type: Optional[Type[BaseException]], 
        exc_val: Optional[BaseException], 
        exc_tb: Optional[TracebackType]
    ) -> None:
        if self.connection:
            await self.connection.close()
    
    async def execute(self, query: str, *args) -> Optional[asyncpg.Record]:
        if not self.connection:
            raise RuntimeError("Connection not established")
        return await self.connection.fetchrow(query, *args)

ジェネリックなデータ処理パイプライン

from typing import TypeVar, Generic, Protocol
from abc import ABC, abstractmethod

T = TypeVar('T')
U = TypeVar('U')

class Processor(Protocol[T, U]):
    async def process(self, item: T) -> U: ...

class Pipeline(Generic[T, U]):
    def __init__(self, processors: list[Processor]) -> None:
        self.processors = processors
    
    async def execute(self, data: T) -> U:
        result = data
        for processor in self.processors:
            result = await processor.process(result)
        return result

ベストプラクティス早見表

コード品質

• 型アノテーション：すべての公開 API に包括的な型アノテーションを追加します
• PEP 8 準拠：black と isort を使ってスタイルガイドラインに従います
• エラー処理：カスタム例外を使って適切な例外処理を実装します
• ドキュメント：すべての関数とクラスに、型ヒントを含む docstring を使用します
• テスト：単体テスト、統合テスト、E2E テストで高いテストカバレッジを維持します

非同期プログラミング

• 非同期コンテキストマネージャー：リソース管理にはasync withを使用します
• 例外処理：try/exceptを使用して非同期例外を適切に処理します
• 並行処理の制限：セマフォを使用して同時実行する処理を制限します
• タイムアウト処理：非同期処理にタイムアウトを実装します
• リソースのクリーンアップ：非同期関数内で適切にクリーンアップされるようにします

パフォーマンス

• プロファイリング：最適化の前にプロファイリングを行い、ボトルネックを特定します
• キャッシュ：適切なキャッシュ戦略を実装します
• コネクションプーリング：データベースへのアクセスにはコネクションプールを使用します
• 遅延読み込み：適切な場合は遅延読み込みを実装します

開発ワークフロー

プロジェクトのセットアップ

• 依存関係の管理にpoetryまたはpip-toolsを使用します
• 最新のPythonパッケージングに対応したpyproject.tomlを実装します
• black、isort、mypyを使ってpre-commitフックを設定します
• 包括的なテストにはpytestとpytest-asyncioを使用します

型チェック

• 厳格なmypy設定を実装します
• IDEの型チェックを強化するためにpyrightを使用します
• 外部ライブラリには型スタブを活用します
• Django、SQLAlchemyなどのフレームワーク向けにmypyプラグインを使用します

統合パターン

python-pro ↔ fastapi/django

• 引き継ぎ：Pythonのプロが型／モデルを設計し、フレームワーク側でエンドポイントを実装します
• コラボレーション：Pydanticモデルを共有し、型安全なAPIを実現します

python-pro ↔ database-administrator

• 引き継ぎ：PythonのプロがORMを使用し、DBAがクエリを最適化します
• コラボレーション：インデックス戦略、クエリパフォーマンス

python-pro ↔ devops-engineer

• 引き継ぎ：Pythonのプロがアプリを作成し、DevOpsがデプロイします
• コラボレーション：Dockerfile、requirements.txt、ヘルスチェック

python-pro ↔ ml-engineer

• 引き継ぎ：PythonのプロがAPIを構築し、MLエンジニアがモデルを統合します
• コラボレーション：FastAPIとモデルサービング（TensorFlow Serving、TorchServe）

追加リソース

• 詳細な技術リファレンス：REFERENCE.mdを参照してください

  • 非同期SQLAlchemyを使用したリポジトリパターン
  • Celery + FastAPIを使用したバックグラウンドタスク
  • 高度なPydantic検証パターン

• コード例とパターン：EXAMPLES.mdを参照してください

  • アンチパターン（型ヒントの無視、非同期処理のブロッキング）
  • FastAPIエンドポイントの例
  • pytest-asyncioを使ったテストパターン