Python 実践 — Kotlin エンジニアのための高速キャッチアップ

対象読者: Kotlin の経験を生かし、対比しながら Python の実践力を身につけたい人。

Kotlin → Python 対応表

1. データクラスとバリデーション

Kotlin の data class に慣れているなら pydantic が自然：

# Kotlin:
# data class Book(val id: Long, val title: String, val status: ReadingStatus)

from pydantic import BaseModel
from enum import Enum

class ReadingStatus(str, Enum):
    UNREAD = "unread"
    READING = "reading"
    COMPLETED = "completed"

class Book(BaseModel):
    id: int
    title: str
    author: str
    status: ReadingStatus = ReadingStatus.UNREAD
    isbn: str | None = None  # nullable（Kotlinの String?）

# バリデーション付きで自動生成
book = Book(id=1, title="Clean Architecture", author="Robert Martin", status="reading")
print(book.model_dump())  # dictに変換
print(book.model_dump_json())  # JSONに変換

2. 非同期処理（Coroutines → asyncio）

Kotlin の suspend fun + coroutineScope とほぼ同じ概念：

# Kotlin:
# suspend fun fetchBook(id: Long): Book = withContext(Dispatchers.IO) { api.getBook(id) }
# 
# viewModelScope.launch {
#     val books = awaitAll(
#         async { fetchBook(1) },
#         async { fetchBook(2) }
#     )
# }

import asyncio
import httpx

async def fetch_book(client: httpx.AsyncClient, book_id: int) -> dict:
    """Kotlin の suspend fun に相当"""
    response = await client.get(f"https://api.example.com/books/{book_id}")
    response.raise_for_status()
    return response.json()

async def main():
    async with httpx.AsyncClient() as client:
        # Kotlin の awaitAll に相当（並列実行）
        books = await asyncio.gather(
            fetch_book(client, 1),
            fetch_book(client, 2),
            fetch_book(client, 3),
        )
    return books

asyncio.run(main())

3. Repository パターン（Kotlin の Room 経験を活かす）

# Kotlin:
# interface BookRepository {
#     fun getAllBooks(): Flow<List<Book>>
#     suspend fun insertBook(book: Book): Long
# }

from abc import ABC, abstractmethod
from typing import Protocol

# Protocol = Kotlin の interface
class BookRepository(Protocol):
    async def get_all_books(self) -> list[Book]: ...
    async def insert_book(self, book: Book) -> int: ...
    async def search_books(self, query: str) -> list[Book]: ...

# 実装クラス（Room の RepositoryImpl に相当）
import aiosqlite

class SQLiteBookRepository:
    def __init__(self, db_path: str):
        self.db_path = db_path

    async def get_all_books(self) -> list[Book]:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM books ORDER BY updated_at DESC") as cursor:
                rows = await cursor.fetchall()
                return [Book(id=r[0], title=r[1], author=r[2], status=r[3]) for r in rows]

    async def insert_book(self, book: Book) -> int:
        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "INSERT INTO books (title, author, status) VALUES (?, ?, ?)",
                (book.title, book.author, book.status.value)
            )
            await db.commit()
            return cursor.lastrowid

4. boto3 — AWS 操作（Lambda 経験の延長）

import boto3
from botocore.exceptions import ClientError

# Lambda で使っている requests → boto3 に切り替え
ec2 = boto3.client('ec2', region_name='ap-northeast-1')

def get_running_instances() -> list[dict]:
    """実行中のEC2一覧を取得"""
    paginator = ec2.get_paginator('describe_instances')
    instances = []
    
    for page in paginator.paginate(
        Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
    ):
        for reservation in page['Reservations']:
            for inst in reservation['Instances']:
                name = next(
                    (t['Value'] for t in inst.get('Tags', []) if t['Key'] == 'Name'),
                    'No Name'
                )
                instances.append({
                    'id': inst['InstanceId'],
                    'type': inst['InstanceType'],
                    'name': name,
                    'private_ip': inst.get('PrivateIpAddress', 'N/A'),
                })
    return instances

# エラーハンドリング（Kotlin の try-catch と同じ構造）
try:
    instances = get_running_instances()
    for inst in instances:
        print(f"{inst['name']:30} {inst['type']:15} {inst['id']}")
except ClientError as e:
    print(f"AWS Error: {e.response['Error']['Code']}: {e.response['Error']['Message']}")

5. 型ヒント実践（Kotlin の型安全に近づける）

from typing import TypeVar, Generic

T = TypeVar('T')

# Kotlin の Result<T> / sealed class に相当
class Result(Generic[T]):
    def __init__(self, value: T | None, error: Exception | None):
        self._value = value
        self._error = error

    @classmethod
    def success(cls, value: T) -> 'Result[T]':
        return cls(value, None)

    @classmethod
    def failure(cls, error: Exception) -> 'Result[T]':
        return cls(None, error)

    @property
    def is_success(self) -> bool:
        return self._error is None

    def get_or_raise(self) -> T:
        if self._error:
            raise self._error
        return self._value

# 使用例
def fetch_config(key: str) -> Result[str]:
    import os
    value = os.environ.get(key)
    if value is None:
        return Result.failure(KeyError(f"環境変数 {key} が未設定"))
    return Result.success(value)

result = fetch_config("DATABASE_URL")
if result.is_success:
    url = result.get_or_raise()

6. CLI ツール作成（click）

import click

@click.group()
def cli():
    """インフラ管理ツール"""
    pass

@cli.command()
@click.option('--region', default='ap-northeast-1', help='AWSリージョン')
@click.option('--env', type=click.Choice(['prod', 'stg', 'dev']), required=True)
def list_instances(region: str, env: str):
    """EC2インスタンス一覧を表示"""
    click.echo(f"Region: {region}, Env: {env}")
    # boto3 で取得...

@cli.command()
@click.argument('instance_id')
@click.confirmation_option(prompt='本当に停止しますか？')
def stop_instance(instance_id: str):
    """EC2インスタンスを停止"""
    click.echo(f"Stopping {instance_id}...")

if __name__ == '__main__':
    cli()

python tool.py list-instances --env prod
python tool.py stop-instance i-abc1234

🔗 次のステップ

• Anthropic API — Claude API 実践（今後追加予定）
• ../../06_DevOps/CICD/01_GitHub_Actions.md — Python スクリプトを CI/CD で自動実行