Pythonのクラス継承を業務で活用する実践パターン集
Pythonのクラス継承は、オブジェクト指向プログラミングの基本的な概念ですが、実際の業務ではどのように使い分けるべきか、明確に理解している開発者は意外と少ないものです。本記事では、教科書的な説明ではなく、実務で実際に遭遇する場面を想定したクラス継承のパターンを紹介します。
1. クラス継承の簡易的な解説
クラス継承とは、既存のクラス（親クラス）の機能を引き継いで、新しいクラス（子クラス）を作成する仕組みです。コードの重複を減らし、保守性を高めるために活用されます。
基本的な構文は以下の通りです：
class ParentClass:
def __init__(self, name):
self.name = name
def greet(self):
return f\"Hello, {self.name}\"
class ChildClass(ParentClass):
def greet(self):
return f\"Hi, {self.name}! Nice to meet you.\"
# 使用例
child = ChildClass(\"Taro\")
print(child.greet()) # Hi, Taro! Nice to meet you.
この例では、ChildClassがParentClassを継承し、greetメソッドをオーバーライドしています。
2. 業務でのユースケース
実務では、以下のような場面でクラス継承が活躍します：
2-1. データベース接続の管理
複数のデータベースを扱う場合、基本的な接続・クエリ実行ロジックを親クラスに集約し、データベースごとの固有処理を子クラスで実装します。
2-2. API通信の抽象化
複数の外部APIと連携する際、共通の通信ロジック（リトライ、タイムアウト、エラー処理）を親クラスで実装し、各API固有のエンドポイント処理を子クラスで行います。
2-3. ログ処理の拡張
基本的なログ出力機能を持つ基底クラスを作成し、用途に応じて異なるフォーマットやフィルタリング機能を子クラスで実装します。
2-4. 定期実行ジョブの管理
定期実行する複数のバッチ処理について、スケジューリングと実行エラー処理を親クラスで統一し、ビジネスロジックを子クラスで実装します。
3. 実装コード：実務レベルの複合例
3-1. データベース接続と操作の継承パターン
以下は、複数のデータベースを扱う場合の実装例です。この例では、基本的な接続・トランザクション管理を親クラスで行い、各データベース固有の処理を子クラスで実装しています。
import sqlite3
import mysql.connector
import logging
from typing import Any, Dict, List, Optional
from contextlib import contextmanager
class DatabaseConnector:
\"\"\"データベース接続の基底クラス\"\"\"
def __init__(self, connection_timeout: int = 30):
self.connection_timeout = connection_timeout
self.connection = None
self.logger = logging.getLogger(self.__class__.__name__)
def connect(self) -> None:
\"\"\"接続を確立する（子クラスで実装）\"\"\"
raise NotImplementedError
def disconnect(self) -> None:
\"\"\"接続を閉じる\"\"\"
if self.connection:
self.connection.close()
self.logger.info(\"Database connection closed\")
@contextmanager
def transaction(self):
\"\"\"トランザクション管理コンテキストマネージャ\"\"\"
try:
yield self.connection.cursor()
self.connection.commit()
self.logger.debug(\"Transaction committed\")
except Exception as e:
self.connection.rollback()
self.logger.error(f\"Transaction failed: {e}\")
raise
def execute_query(self, query: str, params: tuple = ()) -> List[Dict[str, Any]]:
\"\"\"クエリを実行する（子クラスで実装）\"\"\"
raise NotImplementedError
class SQLiteConnector(DatabaseConnector):
\"\"\"SQLiteの接続実装\"\"\"
def __init__(self, db_path: str, connection_timeout: int = 30):
super().__init__(connection_timeout)
self.db_path = db_path
self.connect()
def connect(self) -> None:
\"\"\"SQLiteに接続\"\"\"
try:
self.connection = sqlite3.connect(
self.db_path,
timeout=self.connection_timeout
)
self.connection.row_factory = sqlite3.Row
self.logger.info(f\"Connected to SQLite: {self.db_path}\")
except sqlite3.Error as e:
self.logger.error(f\"SQLite connection failed: {e}\")
raise
def execute_query(self, query: str, params: tuple = ()) -> List[Dict[str, Any]]:
\"\"\"クエリを実行して結果を返す\"\"\"
try:
with self.transaction() as cursor:
cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
except sqlite3.Error as e:
self.logger.error(f\"Query execution failed: {e}\")
raise
class MySQLConnector(DatabaseConnector):
\"\"\"MySQLの接続実装\"\"\"
def __init__(self, host: str, user: str, password: str,
database: str, connection_timeout: int = 30):
self.host = host
self.user = user
self.password = password
self.database = database
super().__init__(connection_timeout)
def connect(self) -> None:
\"\"\"MySQLに接続\"\"\"
try:
self.connection = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database,
connection_timeout=self.connection_timeout
)
self.logger.info(f\"Connected to MySQL: {self.host}\")
except mysql.connector.Error as e:
self.logger.error(f\"MySQL connection failed: {e}\")
raise
def execute_query(self, query: str, params: tuple = ()) -> List[Dict[str, Any]]:
\"\"\"クエリを実行して結果を返す\"\"\"
try:
with self.transaction() as cursor:
cursor.execute(query, params)
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
except mysql.connector.Error as e:
self.logger.error(f\"Query execution failed: {e}\")
raise
# 使用例
def main():
# SQLiteの場合
sqlite_db = SQLiteConnector(\"./data.db\")
results = sqlite_db.execute_query(
\"SELECT * FROM users WHERE age > ?\",
(30,)
)
print(\"SQLite Results:\", results)
sqlite_db.disconnect()
# MySQLの場合（接続情報は実際の値に置き換え）
# mysql_db = MySQLConnector(
# host=\"localhost\",
# user=\"root\",
# password=\"password\",
# database=\"mydb\"
# )
# results = mysql_db.execute_query(\"SELECT * FROM users\")
# mysql_db.disconnect()
if __name__ == \"__main__\":
main()
3-2. API通信の抽象化パターン
複数の外部APIと連携する場合の実装例です。リトライロジックとエラーハンドリングを親クラスで統一しています。
import requests
import time
import logging
from typing import Dict, Any, Optional
from abc import ABC, abstractmethod
from enum import Enum
class HTTPMethod(Enum):
GET = \"GET\"
POST = \"POST\"
PUT = \"PUT\"
DELETE = \"DELETE\"
class APIConnector(ABC):
\"\"\"API通信の基底クラス\"\"\"
MAX_RETRIES = 3
RETRY_DELAY = 2 # 秒
TIMEOUT = 10
def __init__(self, api_key: str = \"\", base_url: str = \"\"):
self.api_key = api_key
self.base_url = base_url
self.logger = logging.getLogger(self.__class__.__name__)
self.session = requests.Session()
@abstractmethod
def get_endpoint(self, resource: str) -> str:
\"\"\"エンドポイントを構築する（子クラスで実装）\"\"\"
pass
@abstractmethod
def get_headers(self) -> Dict[str, str]:
\"\"\"リクエストヘッダを取得する（子クラスで実装）\"\"\"
pass
def request(self, method: HTTPMethod, resource: str,
data: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
\"\"\"リトライ機能付きでリクエストを実行\"\"\"
url = self.get_endpoint(resource)
headers = self.get_headers()
for attempt in range(1, self.MAX_RETRIES + 1):
try:
response = self.session.request(
method=method.value,
url=url,
headers=headers,
json=data,
params=params,
timeout=self.TIMEOUT
)
response.raise_for_status()
self.logger.debug(f\"Request succeeded: {method.value} {url}\")
return response.json() if response.text else {}
except requests.exceptions.RequestException as e:
self.logger.warning(
f\"Request failed (attempt {attempt}/{self.MAX_RETRIES}): {e}\"
)
if attempt < self.MAX_RETRIES:
time.sleep(self.RETRY_DELAY)
else:
self.logger.error(f\"Request failed after {self.MAX_RETRIES} retries\")
raise
def get(self, resource: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
\"\"\"GETリクエスト\"\"\"
return self.request(HTTPMethod.GET, resource, params=params)
def post(self, resource: str, data: Dict[str, Any]) -> Dict[str, Any]:
\"\"\"POSTリクエスト\"\"\"
return self.request(HTTPMethod.POST, resource, data=data)
def close(self) -> None:
\"\"\"セッションを閉じる\"\"\"
self.session.close()
class GitHubAPI(APIConnector):
\"\"\"GitHub APIの実装\"\"\"
def __init__(self, api_key: str):
super().__init__(
api_key=api_key,
base_url=\"https://api.github.com\"
)
def get_endpoint(self, resource: str) -> str:
\"\"\"GitHub APIのエンドポイントを構築\"\"\"
return f\"{self.base_url}{resource}\"
def get_headers(self) -> Dict[str, str]:
\"\"\"GitHubのリクエストヘッダ\"\"\"
return {
\"Authorization\": f\"token {self.api_key}\",
\"Accept\": \"application/vnd.github.v3+json\"
}
def get_user_repos(self, username: str) -> Dict[str, Any]:
\"\"\"ユーザのリポジトリ一覧を取得\"\"\"
return self.get(f\"/users/{username}/repos\")
class SlackAPI(APIConnector):
\"\"\"Slack APIの実装\"\"\"
def __init__(self, api_key: str):
super().__init__(
api_key=api_key,
base_url=\"https://slack.com/api\"
)
def get_endpoint(self, resource: str) -> str:
\"\"\"Slack APIのエンドポイントを構築\"\"\"
return f\"{self.base_url}{resource}\"
def get_headers(self) -> Dict[str, str]:
\"\"\"Slackのリクエストヘッダ\"\"\"
return {
\"Authorization\": f\"Bearer {self.api_key}\",
\"Content-Type\": \"application/json\"
}
def send_message(self, channel: str, text: str) -> Dict[str, Any]:
\"\"\"メッセージを送信\"\"\"
return self.post(\"/chat.postMessage\", {
\"channel\": channel,
\"text\": text
})
# 使用例
def main():
github = GitHubAPI(api_key=\"your_github_token\")
try:
repos = github.get_user_repos(\"octocat\")
print(\"GitHub Repos:\", repos)
except requests.exceptions.RequestException as e:
print(f\"Error: {e}\")
finally:
github.close()
if __name__ == \"__main__\":
main()
4. よくある応用パターン
4-1. 複数継承を用いたミックスイン（Mixin）パターン
複数の機能を組み合わせたい場合、ミックスインパターンが活躍します。実務では、ロギング機能とキャッシング機能を同時に備えたクラスを作る場合などに使用します。
from datetime import datetime, timedelta
from typing import Any, Optional
import json
class LoggerMixin:
\"\"\"ロギング機能を提供するミックスイン\"\"\"
def log_operation(self, operation: str, details: Dict[str, Any]) -> None:
\"\"\"操作ログを記録\"\"\"
timestamp = datetime.now().isoformat()
log_entry = {
\"timestamp\": timestamp,
\"operation\": operation,
\"details\": details
}
print(f\"[LOG] {json.dumps(log_entry)}\")
class CacheMixin:
\"\"\"キャッシング機能を提供するミックスイン\"\"\"
def __init__(self, cache_ttl: int = 300):
self.cache = {}
self.cache_ttl = cache_ttl
self.cache_timestamps = {}
def get_from_cache(self, key: str) -> Optional[Any]:
\"\"\"キャッシュから値を取得\"\"\"
if key not in self.cache:
return None
cached_time = self.cache_timestamps.get(key)
if cached_time and datetime.now() - cached_time > timedelta(seconds=self.cache_ttl):
del self.cache[key]
del self.cache_timestamps[key]
return None
return self.cache[key]
def set_cache(self, key: str, value: Any) -> None:
\"\"\"キャッシュに値を設定\"\"\"
self.cache[key] = value
self.cache_timestamps[key] = datetime.now()
class DataService(LoggerMixin, CacheMixin):
\"\"\"ログとキャッシュ機能を兼ね備えたデータサービス\"\"\"
def __init__(self):
CacheMixin.__init__(self, cache_ttl=600)
def fetch_user_data(self, user_id: int) -> Dict[str, Any]:
\"\"\"ユーザデータを取得（キャッシュ機能付き）\"\"\"
cache_key = f\"user_{user_id}\"
cached_data = self.get_from_cache(cache_key)
if cached_data:
self.log_operation(\"fetch_user_data\", {
\"user_id\": user_id,
\"source\": \"cache\"
})
return cached_data
# 実際のデータベースから取得
user_data = {\"id\": user_id, \"name\": \"Taro\", \"email\": \"taro@example.com\"}
self.set_cache(cache_key, user_data)
self.log_operation(\"fetch_user_data\", {
\"user_id\": user_id,
\"source\": \"database\"
})
return user_data
# 使用例
service = DataService()
user_data = service.fetch_user_data(1)
print(user_data)
4-2. 抽象基底クラス（ABC）パターン
インターフェースを強制したい場合に活躍します。
from abc import ABC, abstractmethod
class ReportGenerator(ABC):
\"\"\"レポート生成の基底クラス\"\"\"
def __init__(self, title: str):
self.title = title
self.content = \"\"
@abstractmethod
def generate(self) -> str:
\"\"\"レポートを生成する\"\"\"
pass
def save(self, filename: str) -> None:
\"\"\"レポートをファイルに保存\"\"\"
content = self.generate()
with open(filename, \"w\", encoding=\"utf-8\") as f:
f.write(content)
print(f\"Report saved: {filename}\")
class PDFReportGenerator(ReportGenerator):
\"\"\"PDF形式のレポート生成\"\"\"
def generate(self) -> str:
return f\"PDF Report: {self.title}\\nContent here...\"
class HTMLReportGenerator(ReportGenerator):
\"\"\"HTML形式のレポート生成\"\"\"
def generate(self) -> str:
return f\"{self.title}
Content here...
\"
# 使用例
pdf_report = PDFReportGenerator(\"Monthly Report\")
pdf_report.save(\"report.pdf\")
html_report = HTMLReportGenerator(\"Monthly Report\")
html_report.save(\"report.html\")
5. 注意点とアンチパターン
5-1. 深すぎい継承階層は避ける
継承を重ねすぎると、コードの追跡が難しくなります。一般的には2～3段階の継承が目安です。
# 悪い例：深すぎる継承階層
class A: pass
class B(A): pass
class C(B): pass
class D(C): pass
class E(D): pass
# 良い例：適度な継承階層
class Base: pass
class Service(Base): pass
class UserService(Service): pass
5-2. 継承より組成を優先する
「is-a」の関係でない場合は、継承ではなく組成（composition）を使うべきです。
# 悪い例：継承を乱用
class User(Database):
pass
# 良い例：組成を使用
class User:
def __init__(self, db):
self.db = db
5-3. super()を正しく使う
親クラスのメソッドを呼び出すときは、直接クラス名を指定するのではなくsuper()を使いましょう。
class Parent:
def method(self):
return \"Parent\"
class Child(Parent):
def method(self):
# 良い例
parent_result = super().method()
return f\"{parent_result} + Child\"
child = Child()
print(child.method()) # Parent + Child
5-4. 継承のチェーンで値の初期化を忘れない
複数クラスの継承を行う場合、各クラスの__init__メソッドが正しく呼ばれていることを確認しましょう。
class LoggerMixin:
def __init__(self):
self.logs = []
def add_log(self, message: str):
self.logs.append(message)
class Database:
def __init__(self, connection_string: str):
self.connection = connection_string
class UserRepository(LoggerMixin, Database):
def __init__(self, connection_string: str):
LoggerMixin.__init__(self)
Database.__init__(self, connection_string)
self.users = []
def add_user(self, user):
self.add_log(f\"User added: {user}\")
self.users.append(user)
# 使用例
repo = UserRepository(\"localhost:5432\")
repo.add_user({\"id\": 1, \"name\": \"Taro\"})
print(repo.logs) # ['User added: {\'id\': 1, \'name\': \'Taro\'}']
6. 実務で役立つベストプラクティス
6-1. 型ヒント（Type Hints）を活用する
型ヒントを使うことで、コードの可読性と保守性が向上します。
from typing import TypeVar, Generic, List
T = TypeVar('T')
class Repository(Generic[T]):
def __init__(self, items: List[T] = None):
self.items = items or []
def add(self, item: T) -> None:
self.items.append(item)
def get_all(self) -> List[T]:
return self.items
class User:
def __init__(self, id: int, name: str):
self.id = id
self.name = name
# 使用例
user_repo: Repository[User] = Repository()
user_repo.add(User(1, \"Taro\"))
users: List[User] = user_repo.get_all()
6-2. docstringを充実させる
クラスとメソッドのドキュメント文字列を充実させることで、他の開発者が使いやすくなります。
class ConfigManager:
\"\"\"アプリケーション設定を管理するクラス
Attributes:
config (dict): 設定を保持する辞書
Example:
>>> config = ConfigManager(\"config.yaml\")
>>> api_key = config.get(\"api_key\")
\"\"\"
def __init__(self, config_file: str):
\"\"\"設定ファイルから設定を読み込む
Args:
config_file (str): 設定ファイルのパス
Raises:
FileNotFoundError: 設定ファイルが見つからない場合
\"\"\"
self.config = self._load_config(config_file)
def get(self, key: str, default: Any = None) -> Any:
\"\"\"設定値を取得
Args:
key (str): 設定キー
default (Any): キーが見つからない場合のデフォルト値
Returns:
Any: 設定値またはデフォルト値
\"\"\"
return self.config.get(key, default)
7. まとめ
Pythonのクラス継承は、適切に使用すれば業務コードの品質と保守性を大きく向上させることができます。本記事で紹介した内容をまとめると以下の通りです：
- データベース接続やAPI通信など、共通ロジックを親クラスで統一することで、コードの重複を減らし、バグを防ぐことができます。
- ミックスインパターンを活用する
- 抽象基底クラス（ABC）を使用する
- 深すぎる継承階層は避ける
- 型ヒントとdocstringを充実させる
実務では、完璧なクラス設計よりも、シンプルで保守しやすいコード設計が重要です。本記事で紹介したパターンを参考に、プロジェクトの要件に合わせた継承設計を心がけてください。
