背景
ハッカソンでの成果物の延長を実装する際に、適用できそうだったので勉強した。
開発したものはシンプルで、slackの自分の発言を拾ってきて、一覧にするというもの。
一覧にした後、AIに分析させて自分の評価面談準備を楽したいという魂胆で作った。
この延長戦として、slackにとどまらず、google chatやgithubなどさまざまなコミュニケーションツールを対象にデータを取得、同様に一覧化したいと思った。
その際に、別々のツールであるが、最終的に集約されるフォーマットにアダプターパターンが適用できるとのことだった。知らなかったので、学習しながら、設計と実装の方針をAIに整理してもらった。
イメージ
構成要素はシンプルで
- 抽象クラスのAdapterクラス
- 各サービスを当てはめた具象クラスのAdapterクラス
- フォーマットに沿ったアウトプットを定義するActivityクラス
- 一連のアダプタを実体化させるFactoryクラス
具象クラスのAdapterのアウトプットをActivityで統一し、Factoryでそれぞれのadapterを実体化させて、共通の読み取り処理を実行させれば向こう側のコンセントのインターフェースを認識することなく、一つのアウトプットに整理できる。
アダプターパターン設計書
1. 概要
複数のデータソース（Slack、GitHub、Google Chat）から取得したデータを、統一されたインターフェースで扱うためにアダプターパターンを適用します。
2. アダプターパターンの適用理由
2.1 課題
- Slack、GitHub、Google Chat はそれぞれ異なるAPIを持つ
- データ構造やフィールド名が異なる
- 各データソースの差異を吸収する必要がある
2.2 解決策
- 共通のインターフェース（
DataSourceAdapter）を定義 - 各データソース用のアダプタークラスを実装
- クライアントコードは共通インターフェースのみを扱う
3. クラス設計
3.1 共通インターフェース
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
@dataclass
class Activity:
"""統一されたアクティビティデータ"""
timestamp: datetime # 作成日時
source: str # データソース（slack/github/google_chat）
channel_name: str # チャンネル名・リポジトリ名・スペース名
user_name: str # ユーザー名
text: str # 発言内容・コメント・メッセージ
url: str # URLリンク
def to_dict(self) -> dict:
"""CSV出力用の辞書に変換"""
return {
"ts": self.timestamp.isoformat(),
"source": self.source,
"channel_name": self.channel_name,
"user_name": self.user_name,
"text": self.text,
"url": self.url,
}
class DataSourceAdapter(ABC):
"""データソースアダプターの基底クラス"""
@abstractmethod
def fetch_activities(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
target_user: Optional[str] = None,
**kwargs
) -> List[Activity]:
"""
アクティビティを取得
Args:
start_date: 開始日（YYYY-MM-DD）
end_date: 終了日（YYYY-MM-DD）
target_user: 対象ユーザー名
**kwargs: データソース固有のパラメータ
Returns:
Activityオブジェクトのリスト
"""
pass
@abstractmethod
def get_source_name(self) -> str:
"""データソース名を取得"""
pass
3.2 Slack アダプター
from src.slack.client import SlackClient
from src.slack.messages import fetch_channel_messages
class SlackAdapter(DataSourceAdapter):
"""Slack データソースアダプター"""
def __init__(self, token: str):
self.client = SlackClient(token)
def fetch_activities(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
target_user: Optional[str] = None,
channels: Optional[List[str]] = None,
) -> List[Activity]:
"""Slackメッセージを取得してActivityに変換"""
# Slackメッセージを取得
slack_messages = []
for channel in channels or self._get_all_channels():
messages = fetch_channel_messages(
client=self.client,
channel_id=channel.id,
channel_name=channel.name,
oldest=self._date_to_timestamp(start_date),
latest=self._date_to_timestamp(end_date),
target_user=target_user,
)
slack_messages.extend(messages)
# Activityオブジェクトに変換
return [
Activity(
timestamp=msg.timestamp,
source="slack",
channel_name=msg.channel_name,
user_name=msg.user_name,
text=msg.text,
url=msg.url,
)
for msg in slack_messages
]
def get_source_name(self) -> str:
return "slack"
def _get_all_channels(self):
"""全チャンネルを取得"""
return self.client.get_channels()
def _date_to_timestamp(self, date_str: Optional[str]) -> Optional[str]:
"""日付文字列をUnixタイムスタンプに変換"""
if not date_str:
return None
dt = datetime.strptime(date_str, "%Y-%m-%d")
return str(int(dt.timestamp()))
3.3 GitHub アダプター
from src.github.client import GitHubClient
from src.github.activities import fetch_repository_activities
class GitHubAdapter(DataSourceAdapter):
"""GitHub データソースアダプター"""
def __init__(self, token: str):
self.client = GitHubClient(token)
def fetch_activities(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
target_user: Optional[str] = None,
repositories: Optional[List[str]] = None,
) -> List[Activity]:
"""GitHubアクティビティを取得してActivityに変換"""
# GitHubアクティビティを取得
github_activities = fetch_repository_activities(
client=self.client,
repositories=repositories or [],
start_date=start_date,
end_date=end_date,
target_user=target_user,
)
# Activityオブジェクトに変換
return [
Activity(
timestamp=activity.timestamp,
source="github",
channel_name=activity.repo_name,
user_name=activity.user_name,
text=activity.text,
url=activity.url,
)
for activity in github_activities
]
def get_source_name(self) -> str:
return "github"
3.4 Google Chat アダプター
from src.google_chat.client import GoogleChatClient
from src.google_chat.messages import fetch_space_messages
class GoogleChatAdapter(DataSourceAdapter):
"""Google Chat データソースアダプター"""
def __init__(self, credentials_path: str):
self.client = GoogleChatClient(credentials_path)
def fetch_activities(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
target_user: Optional[str] = None,
spaces: Optional[List[str]] = None,
) -> List[Activity]:
"""Google Chatメッセージを取得してActivityに変換"""
# Google Chatメッセージを取得
chat_messages = []
for space in spaces or self._get_all_spaces():
messages = fetch_space_messages(
client=self.client,
space_id=space.id,
space_name=space.name,
start_date=start_date,
end_date=end_date,
target_user=target_user,
)
chat_messages.extend(messages)
# Activityオブジェクトに変換
return [
Activity(
timestamp=msg.timestamp,
source="google_chat",
channel_name=msg.space_name,
user_name=msg.user_name,
text=msg.text,
url=msg.url,
)
for msg in chat_messages
]
def get_source_name(self) -> str:
return "google_chat"
def _get_all_spaces(self):
"""全スペースを取得"""
return self.client.get_spaces()
4. アダプターファクトリー
複数のアダプターを一元管理するファクトリークラス：
from typing import List
import os
from dotenv import load_dotenv
class DataSourceAdapterFactory:
"""データソースアダプターのファクトリー"""
@staticmethod
def create_adapters(sources: List[str] = None) -> List[DataSourceAdapter]:
"""
指定されたデータソースのアダプターを生成
Args:
sources: データソース名のリスト（["slack", "github", "google_chat"]）
Noneの場合は環境変数から判断
Returns:
DataSourceAdapterのリスト
"""
load_dotenv()
adapters = []
# デフォルトは環境変数から判断
if sources is None:
sources = []
if os.getenv("SLACK_BOT_TOKEN"):
sources.append("slack")
if os.getenv("GITHUB_TOKEN"):
sources.append("github")
if os.getenv("GOOGLE_CHAT_CREDENTIALS"):
sources.append("google_chat")
# 各アダプターを生成
for source in sources:
if source == "slack":
token = os.getenv("SLACK_BOT_TOKEN")
if token:
adapters.append(SlackAdapter(token))
elif source == "github":
token = os.getenv("GITHUB_TOKEN")
if token:
adapters.append(GitHubAdapter(token))
elif source == "google_chat":
creds = os.getenv("GOOGLE_CHAT_CREDENTIALS")
if creds:
adapters.append(GoogleChatAdapter(creds))
return adapters
5. 統合取得サービス
アダプターを使ってデータを統合的に取得：
import csv
from pathlib import Path
from typing import List, Optional
class ActivityService:
"""複数データソースからアクティビティを統合取得"""
def __init__(self, adapters: List[DataSourceAdapter]):
self.adapters = adapters
def fetch_all_activities(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
target_user: Optional[str] = None,
**kwargs
) -> List[Activity]:
"""
全データソースからアクティビティを取得
Args:
start_date: 開始日（YYYY-MM-DD）
end_date: 終了日（YYYY-MM-DD）
target_user: 対象ユーザー名
**kwargs: データソース固有のパラメータ
Returns:
全データソースのActivityを統合したリスト
"""
all_activities = []
for adapter in self.adapters:
print(f"Fetching activities from {adapter.get_source_name()}...")
activities = adapter.fetch_activities(
start_date=start_date,
end_date=end_date,
target_user=target_user,
**kwargs
)
all_activities.extend(activities)
print(f" -> {len(activities)} activities fetched")
# タイムスタンプでソート
all_activities.sort(key=lambda x: x.timestamp)
return all_activities
def export_to_csv(
self,
activities: List[Activity],
output_path: str = "output/all_activities.csv"
) -> None:
"""
アクティビティをCSVに出力
Args:
activities: Activityオブジェクトのリスト
output_path: 出力先CSVファイルパス
"""
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w", newline="", encoding="utf-8") as f:
fieldnames = ["ts", "source", "channel_name", "user_name", "text", "url"]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for activity in activities:
writer.writerow(activity.to_dict())
print(f"\nExported {len(activities)} activities to {output_path}")
6. 使用例
# メインスクリプト
def main():
# アダプターを生成（環境変数から自動判定）
adapters = DataSourceAdapterFactory.create_adapters()
# または明示的に指定
# adapters = DataSourceAdapterFactory.create_adapters(["slack", "github"])
# サービスを初期化
service = ActivityService(adapters)
# 全データソースからアクティビティを取得
activities = service.fetch_all_activities(
start_date="2025-12-01",
end_date="2025-12-31",
target_user="username",
# データソース固有のパラメータ
channels=["team-g"], # Slack用
repositories=["owner/repo"], # GitHub用
spaces=["space-123"], # Google Chat用
)
# CSVに出力
service.export_to_csv(activities, "output/all_activities.csv")
if __name__ == "__main__":
main()
7. ディレクトリ構成
src/
├── adapters/
│ ├── __init__.py
│ ├── base.py # DataSourceAdapter, Activity
│ ├── slack_adapter.py # SlackAdapter
│ ├── github_adapter.py # GitHubAdapter
│ ├── google_chat_adapter.py # GoogleChatAdapter
│ └── factory.py # DataSourceAdapterFactory
├── services/
│ ├── __init__.py
│ └── activity_service.py # ActivityService
├── slack/
│ ├── __init__.py
│ ├── client.py
│ ├── messages.py
│ └── models.py
├── github/
│ ├── __init__.py
│ ├── client.py
│ ├── activities.py
│ └── models.py
└── google_chat/
├── __init__.py
├── client.py
├── messages.py
└── models.py
8. アダプターパターンの利点
8.1 拡張性
- 新しいデータソース（例: Discord、Teams）を追加する際、新しいアダプタークラスを作成するだけ
- 既存のコードを変更する必要がない
8.2 保守性
- 各データソースのロジックが独立している
- バグ修正や変更が他のデータソースに影響しない
8.3 テスト容易性
- モックアダプターを作成してテストが簡単
- 各アダプターを個別にテストできる
8.4 統一されたインターフェース
- クライアントコードはデータソースの違いを意識する必要がない
- 全データソースを同じ方法で扱える
感想
実際に開発した延長で勉強することができて面白かった。
adapterという命名が適切なのかは疑問だが、向こう側の処理を認識せずに、しかも拡張しやすい形だと思った。
活用していきたい。