Python文法の基礎と応用
1. リスト内包表記の活用
Pythonのリスト内包表記は、簡潔で読みやすいコードを書くための強力な機能です。 従来のforループを使用する方法と比較して、より簡潔に記述できます。 また、ジェネレータ式を使用することで、メモリ効率の良いコードを書くことができます。
# 従来のforループを使用する方法
numbers = []
for i in range(1, 11):
if i % 2 == 0:
numbers.append(i * i)
# リスト内包表記を使用する方法
numbers = [i * i for i in range(1, 11) if i % 2 == 0]
# ジェネレータ式を使用する方法（メモリ効率が良い）
numbers = (i * i for i in range(1, 11) if i % 2 == 0)
# 複数のforループを使用した複雑な例
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flattened = [num for row in matrix for num in row] # [1, 2, 3, 4, 5, 6, 7, 8, 9]
# 文字列の配列を結合する例
words = ['Python はどういう言語だと思いますか？', 'Python は速度を求められる場面ではおすすめできない。', 'だが、学習・ツール作成・実験にいたるまで、あらゆる場面で有用である。', 'これは何故か？それは Python のつづりがとてもかわいらしいからだ']
sentence = ' '.join([word.title() for word in words])2. デコレータの実装と使用方法
デコレータは、関数やクラスの動作を変更するための強力な機能です。 関数デコレータとクラスデコレータの両方を実装できます。 また、デコレータに引数を渡すことで、より柔軟な実装が可能です。
import functools
import time
from typing import Callable, TypeVar, ParamSpec
P = ParamSpec("P")
R = TypeVar("R")
def timing_decorator(func: Callable[P, R]) -> Callable[P, R]:
"""関数の実行時間を計測するデコレータ"""
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start_time = time.perf_counter()
result = func(*args, **kwargs)
end_time = time.perf_counter()
print(f"{func.__name__} took {end_time - start_time:.6f} seconds to execute")
return result
return wrapper
def retry(max_attempts: int = 3, delay: float = 1.0) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""指定回数リトライするデコレータ（引数付き）"""
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
last_error = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_attempts - 1:
time.sleep(delay)
raise last_error
return wrapper
return decorator3. 非同期プログラミングの実装例
Pythonの非同期プログラミングは、I/O待ちの多い処理を効率的に実行するために使用されます。 asyncioモジュールを使用することで、コルーチンを定義し、非同期処理を実装できます。 また、async/awaitキーワードを使用することで、より直感的なコードを書くことができます。
import asyncio
from typing import List, Dict, Any
import aiohttp
import json
class AsyncAPIClient:
"""非同期APIクライアントの実装例"""
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.session: aiohttp.ClientSession | None = None
async def __aenter__(self) -> 'AsyncAPIClient':
self.session = aiohttp.ClientSession(
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
if self.session:
await self.session.close()
self.session = None
async def get_items(self, page: int = 1, per_page: int = 10) -> List[Dict[str, Any]]:
if not self.session:
raise RuntimeError("Session is not initialized. Use async with context manager.")
params = {'page': page, 'per_page': per_page}
async with self.session.get(f'{self.base_url}/items', params=params) as response:
response.raise_for_status()
return await response.json()
async def create_item(self, data: Dict[str, Any]) -> Dict[str, Any]:
if not self.session:
raise RuntimeError("Session is not initialized. Use async with context manager.")
async with self.session.post(f'{self.base_url}/items', json=data) as response:
response.raise_for_status()
return await response.json()
async def main():
api_client = AsyncAPIClient('https://api.example.com', 'your-api-key')
async with api_client:
# 並列でAPIリクエストを実行
tasks = [
api_client.get_items(page=i) for i in range(1, 4)
]
results = await asyncio.gather(*tasks)
# 結果を処理
for page, items in enumerate(results, 1):
print(f"Page {page}: {len(items)} items")
# 新しいアイテムを作成
new_item = await api_client.create_item({
'name': 'Test Item',
'description': 'This is a test item',
'price': 1000
})
print(f"Created new item: {json.dumps(new_item, indent=2)}")
if __name__ == '__main__':
asyncio.run(main())4. 型ヒントとProtocolの活用
Pythonの型ヒントを使用することで、コードの可読性が向上し、静的型チェッカーによるエラーの早期発見が可能になります。 特に、Protocolクラスを使用することで、構造的部分型を実現できます。 これにより、より柔軟なインターフェースの定義が可能になります。
from typing import Protocol, TypeVar, List, Dict, Any, runtime_checkable
from dataclasses import dataclass
from datetime import datetime
T = TypeVar('T')
@runtime_checkable
class Repository(Protocol[T]):
"""リポジトリのインターフェース定義"""
async def get(self, id: str) -> T:
"""IDに基づいてエンティティを取得"""
...
async def save(self, entity: T) -> None:
"""エンティティを保存"""
...
async def delete(self, id: str) -> None:
"""IDに基づいてエンティティを削除"""
...
async def find_all(self) -> List[T]:
"""すべてのエンティティを取得"""
...
@dataclass
class User:
"""ユーザーエンティティ"""
id: str
name: str
email: str
created_at: datetime
updated_at: datetime
metadata: Dict[str, Any]
class UserRepository:
"""ユーザーリポジトリの実装"""
def __init__(self, connection_string: str):
self.connection_string = connection_string
async def get(self, id: str) -> User:
# データベースからユーザーを取得する実装
...
async def save(self, user: User) -> None:
# ユーザーをデータベースに保存する実装
...
async def delete(self, id: str) -> None:
# ユーザーをデータベースから削除する実装
...
async def find_all(self) -> List[User]:
# すべてのユーザーを取得する実装
...5. コンテキストマネージャの実装
コンテキストマネージャは、リソースの確保と解放を自動的に行うための機能です。 with文と組み合わせることで、より安全なコードを書くことができます。 また、非同期コンテキストマネージャを実装することで、非同期処理でもリソースの適切な管理が可能です。
from typing import TypeVar, Generic, Optional
from contextlib import contextmanager
import threading
import logging
T = TypeVar('T')
class ResourcePool(Generic[T]):
"""リソースプールの実装例"""
def __init__(self, factory: Callable[[], T], max_size: int = 10):
self.factory = factory
self.max_size = max_size
self.resources: List[T] = []
self.available: List[T] = []
self.lock = threading.Lock()
self.not_empty = threading.Condition(self.lock)
self.logger = logging.getLogger(__name__)
def _create_resource(self) -> T:
"""新しいリソースを作成"""
resource = self.factory()
self.resources.append(resource)
return resource
@contextmanager
def acquire(self, timeout: Optional[float] = None) -> Generator[T, None, None]:
"""リソースを取得するコンテキストマネージャ"""
resource = self._acquire(timeout)
try:
yield resource
finally:
self._release(resource)
def _acquire(self, timeout: Optional[float] = None) -> T:
"""リソースを取得"""
with self.lock:
while not self.available and len(self.resources) >= self.max_size:
if not self.not_empty.wait(timeout):
raise TimeoutError("Failed to acquire resource: pool is full and no resources available")
if self.available:
resource = self.available.pop()
self.logger.debug(f"Acquired existing resource (total: {len(self.resources)}, available: {len(self.available)})")
else:
resource = self._create_resource()
self.logger.debug(f"Created new resource (total: {len(self.resources)}, available: {len(self.available)})")
return resource
def _release(self, resource: T) -> None:
"""リソースを解放"""
with self.lock:
self.available.append(resource)
self.not_empty.notify()
self.logger.debug(f"Released resource (total: {len(self.resources)}, available: {len(self.available)})")
# 使用例
class DatabaseConnection:
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.connected = False
def connect(self) -> None:
self.connected = True
print(f"Connected to database: {self.connection_string}")
def disconnect(self) -> None:
self.connected = False
print(f"Disconnected from database: {self.connection_string}")
def execute(self, query: str) -> None:
if not self.connected:
raise RuntimeError("Not connected to database")
print(f"Executing query: {query}")
# プールの作成
connection_pool = ResourcePool(
lambda: DatabaseConnection("postgresql://localhost:5432/mydb"),
max_size=5
)
# リソースの使用
with connection_pool.acquire() as conn:
conn.connect()
conn.execute("SELECT * FROM users")
# 自動的にプールに返却される6. Python学習のための参考リンク
Pythonを学習する際に参考になる公式ドキュメントやコミュニティサイトを紹介します。 これらのリソースを活用することで、より効率的に学習を進めることができます。