EvoSpikeNet SDK 開発者ガイド
Copyright: 2026 Moonlight Technologies Inc.
Author: Masahiro Aoki
最終更新日: 2026年1月15日
概要
この開発者ガイドは、EvoSpikeNet SDKの内部構造、拡張方法、ベストプラクティスについて説明します。SDKの開発、テスト、デプロイに関わる開発者を対象としています。
SDKアーキテクチャ
主要コンポーネント
evospikenet/
├── sdk.py # メインAPIクライアント
├── sdk_jupyter.py # Jupyter Notebook統合
├── __init__.py # パッケージ初期化
└── ...
クラス階層
EvoSpikeNetAPIClient (メインクライアント)
├── 基本HTTP機能
├── エラーハンドリング
├── 統計収集
└── 拡張機能
JupyterAPIClient (Jupyter統合)
├── EvoSpikeNetAPIClient (継承)
├── HTML表示機能
├── インタラクティブ機能
└── マジックコマンド
WebSocketClient (リアルタイム通信)
├── 接続管理
├── メッセージ送受信
└── イベントハンドリング
開発環境のセットアップ
必須ツール
# Python開発環境
pip install -e ".[dev]"
# テスト依存関係
pip install pytest pytest-cov pytest-asyncio
# ドキュメント生成
pip install sphinx sphinx-rtd-theme
# コード品質
pip install black isort flake8 mypy
開発ワークフロー
# コードフォーマット
black evospikenet/
isort evospikenet/
# 型チェック
mypy evospikenet/
# リント
flake8 evospikenet/
# テスト実行
pytest
# カバレッジレポート
pytest --cov=evospikenet --cov-report=html
SDK拡張のガイドライン
新しいAPIエンドポイントの追加
- エンドポイントURLの定義
class EvoSpikeNetAPIClient:
def __init__(self, ...):
# 既存の初期化
self.new_endpoint_url = f"{self.base_url}/api/new_endpoint"
- メソッドの実装
def new_api_method(self, param1: str, param2: int = 10) -> Dict[str, Any]:
"""
新しいAPIメソッドの説明。
Args:
param1: パラメータ1の説明
param2: パラメータ2の説明（デフォルト値付き）
Returns:
APIレスポンスの辞書
Raises:
EvoSpikeNetAPIError: APIエラーの場合
"""
payload = {
"param1": param1,
"param2": param2
}
return cast(Dict[str, Any], self._make_request("post", self.new_endpoint_url, json=payload))
- 型ヒントの使用
from typing import Dict, Any, Optional, List, Union
def method_with_complex_types(
self,
required_param: str,
optional_param: Optional[int] = None,
list_param: List[str] = None,
union_param: Union[str, int] = "default"
) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
# 実装
pass
エラーハンドリングの拡張
class CustomAPIError(EvoSpikeNetAPIError):
"""カスタムAPIエラークラス"""
def __init__(self, error_info: ErrorInfo, custom_field: str = None):
super().__init__(error_info)
self.custom_field = custom_field
# 使用例
try:
result = client.new_method()
except CustomAPIError as e:
print(f"Custom error: {e.custom_field}")
# 特別な処理
非同期メソッドの追加
import aiohttp
async def new_async_method(self, param: str) -> Dict[str, Any]:
"""
非同期バージョンの新しいメソッド。
Args:
param: メソッドパラメータ
Returns:
非同期レスポンス
"""
payload = {"param": param}
async with aiohttp.ClientSession() as session:
async with session.post(
self.new_endpoint_url,
json=payload,
headers=self.session.headers,
timeout=aiohttp.ClientTimeout(total=self.timeout)
) as response:
response.raise_for_status()
return await response.json()
テスト戦略
テスト構造
tests/
├── unit/ # 単体テスト
│ ├── test_sdk.py
│ └── test_jupyter.py
├── integration/ # 統合テスト
│ └── test_api_integration.py
├── fixtures/ # テストフィクスチャ
└── conftest.py # pytest設定
単体テストの例
import pytest
from unittest.mock import Mock, patch
<!-- from evospikenet.sdk import EvoSpikeNetAPIClient, EvoSpikeNetAPIError -->
class TestEvoSpikeNetAPIClient:
"""EvoSpikeNetAPIClientのテストクラス"""
@pytest.fixture
def client(self):
"""テスト用クライアントフィクスチャ"""
return EvoSpikeNetAPIClient(base_url="http://test-server")
@patch('requests.Session.request')
def test_generate_success(self, mock_request, client):
"""正常系のテキスト生成テスト"""
# モックレスポンス設定
mock_response = Mock()
mock_response.json.return_value = {
"generated_text": "Test response",
"prompt": "Test prompt"
}
mock_request.return_value = mock_response
# テスト実行
result = client.generate("Test prompt")
# アサーション
assert result["generated_text"] == "Test response"
assert result["prompt"] == "Test prompt"
mock_request.assert_called_once()
@patch('requests.Session.request')
def test_generate_http_error(self, mock_request, client):
"""HTTPエラーのテスト"""
# エラーレスポンス設定
mock_response = Mock()
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError("404 Not Found")
mock_response.status_code = 404
mock_response.json.return_value = {"detail": "Not found"}
mock_request.return_value = mock_response
# テスト実行と例外確認
with pytest.raises(EvoSpikeNetAPIError) as exc_info:
client.generate("Test prompt")
assert exc_info.value.error_info.status_code == 404
assert "Not found" in exc_info.value.error_info.message
def test_stats_tracking(self, client):
"""統計追跡のテスト"""
# 初期状態確認
stats = client.get_stats()
assert stats["requests"] == 0
# 統計リセット
client.reset_stats()
stats = client.get_stats()
assert stats["requests"] == 0
統合テストの例
import pytest
<!-- TODO: update or remove - import faileNetAPIClient -->
class TestAPIIntegration:
"""API統合テスト"""
@pytest.fixture(scope="class")
def live_client(self):
"""ライブAPIサーバー接続用フィクスチャ"""
client = EvoSpikeNetAPIClient()
# サーバーが利用可能か確認
if not client.wait_for_server(timeout=10):
pytest.skip("API server not available")
return client
def test_health_check_integration(self, live_client):
"""ヘルスチェックの統合テスト"""
health = live_client.health_check()
assert "status" in health
assert health["status"] in ["healthy", "unhealthy"]
def test_generate_integration(self, live_client):
"""テキスト生成の統合テスト"""
prompt = "Hello, world!"
result = live_client.generate(prompt, max_length=50)
assert "generated_text" in result
assert isinstance(result["generated_text"], str)
assert len(result["generated_text"]) > 0
def test_batch_generate_integration(self, live_client):
"""バッチ生成の統合テスト"""
prompts = ["Test 1", "Test 2", "Test 3"]
results = live_client.batch_generate(prompts, max_length=30)
assert len(results) == len(prompts)
for result in results:
if "generated_text" in result:
assert isinstance(result["generated_text"], str)
非同期テストの例
import pytest
import asyncio
from unittest.mock import AsyncMock, patch
<!-- モジュール 'evospikenet' が見つかりません。パッケージ内の移動/名前変更を確認してください -->
<!-ions:
"""非同期操作のテスト"""
@pytest.fixture
def client(self):
return EvoSpikeNetAPIClient()
@pytest.mark.asyncio
@patch('aiohttp.ClientSession.post')
async def test_generate_async_success(self, mock_post, client):
"""非同期生成の成功テスト"""
# モックレスポンス設定
mock_response = AsyncMock()
mock_response.json.return_value = {"generated_text": "Async response"}
mock_response.raise_for_status.return_value = None
mock_post.return_value.__aenter__.return_value = mock_response
# テスト実行
result = await client.generate_async("Test prompt")
# アサーション
assert result["generated_text"] == "Async response"
@pytest.mark.asyncio
async def test_concurrent_generation(self, client):
"""並行生成のテスト"""
prompts = ["Prompt 1", "Prompt 2", "Prompt 3"]
# 実際のAPIを呼び出す場合のタイムアウト設定
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(
asyncio.gather(*[client.generate_async(p) for p in prompts]),
timeout=1.0 # 短いタイムアウトでテスト
)
パフォーマンス最適化
接続プーリング
# セッションの再利用
class EvoSpikeNetAPIClient:
def __init__(self, ...):
self.session = requests.Session()
# 接続プーリング設定
adapter = requests.adapters.HTTPAdapter(
pool_connections=10,
pool_maxsize=20,
max_retries=3
)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
レスポンスキャッシュ
from functools import lru_cache
import hashlib
class EvoSpikeNetAPIClient:
def __init__(self, ...):
self.cache = {}
def _get_cache_key(self, method: str, url: str, **kwargs) -> str:
"""キャッシュキーの生成"""
key_data = f"{method}:{url}:{str(sorted(kwargs.items()))}"
return hashlib.md5(key_data.encode()).hexdigest()
@lru_cache(maxsize=100)
def _cached_request(self, cache_key: str, method: str, url: str, **kwargs):
"""キャッシュ付きリクエスト"""
return self._make_request(method, url, **kwargs)
ストリーミング処理
def download_large_artifact(self, artifact_id: str, destination_path: str):
"""大容量アーティファクトのストリーミングダウンロード"""
url = f"{self.base_url}/api/artifacts/{artifact_id}/download"
with self.session.get(url, stream=True, timeout=300) as response:
response.raise_for_status()
with open(destination_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
セキュリティ考慮事項
APIキーの管理
import os
from typing import Optional
class EvoSpikeNetAPIClient:
def __init__(self, api_key: Optional[str] = None, ...):
# 環境変数からのAPIキー取得
self.api_key = api_key or os.getenv('EVOSPIKENET_API_KEY')
if not self.api_key:
raise ValueError("API key must be provided or set in EVOSPIKENET_API_KEY environment variable")
# ヘッダー設定
self.session.headers.update({
'X-API-Key': self.api_key,
'User-Agent': 'EvoSpikeNet-SDK/2.0'
})
証明書の検証
import ssl
class EvoSpikeNetAPIClient:
def __init__(self, verify_ssl: bool = True, ca_cert_path: Optional[str] = None, ...):
self.session.verify = verify_ssl
if ca_cert_path:
self.session.verify = ca_cert_path
elif verify_ssl:
# システム証明書を使用
pass
else:
# 警告を表示
import warnings
warnings.warn("SSL verification is disabled. This is insecure for production use.")
タイムアウト設定
class EvoSpikeNetAPIClient:
def __init__(self, timeout: int = 60, ...):
self.timeout = timeout
# 異なる操作に対するタイムアウト設定
self.timeouts = {
'health': 5,
'generate': timeout,
'upload': 300, # アップロードは長め
'download': 300, # ダウンロードは長め
}
ドキュメント生成
Sphinxドキュメント
# docs/conf.py
extensions = [
'sphinx.ext.autodoc',
'sphinx.ext.napoleon',
'sphinx.ext.viewcode',
'sphinx.ext.intersphinx',
]
# APIドキュメント生成
# sphinx-apidoc -f -o docs/source evospikenet
型ヒントの活用
from typing import Dict, Any, Optional, List, Union
def well_documented_method(
self,
required_param: str,
optional_param: Optional[int] = None
) -> Dict[str, Any]:
"""
型ヒント付きのメソッド。
Args:
required_param: 必須パラメータの説明
optional_param: オプションのパラメータ説明
Returns:
戻り値の説明
Raises:
EvoSpikeNetAPIError: エラーの説明
Example:
>>> result = client.well_documented_method("test")
>>> print(result)
{'status': 'success'}
"""
pass
リリースプロセス
バージョン管理
# evospikenet/__init__.py
__version__ = "2.0.0"
# バージョン取得メソッド
def get_version() -> str:
"""SDKバージョンを取得"""
return __version__
リリースチェックリスト
- [ ] 全てのテストが通る
- [ ] カバレッジが80%以上
- [ ] 型チェックが通る
- [ ] リントが通る
- [ ] ドキュメントが更新されている
- [ ] 変更履歴が更新されている
- [ ] バージョン番号が更新されている
- [ ] 統合テストが通る
パッケージング
# setup.py
from setuptools import setup, find_packages
setup(
name="evospikenet-sdk",
version="2.0.0",
packages=find_packages(),
install_requires=[
"requests>=2.25.0",
"typing-extensions>=4.0.0",
],
extras_require={
"jupyter": ["ipython", "jupyter"],
"async": ["aiohttp"],
"websocket": ["websockets"],
"all": ["ipython", "jupyter", "aiohttp", "websockets"],
},
python_requires=">=3.8",
)
トラブルシューティング
一般的な開発時問題
-
Importエラー
# パスが正しいか確認 python -c "import sys; print(sys.path)" # 開発インストールが正しいか確認 pip show evospikenet -
テスト失敗
# 詳細なテスト出力 pytest -v -s # 特定のテストのみ実行 pytest tests/unit/test_sdk.py::TestEvoSpikeNetAPIClient::test_generate_success -
型チェックエラー
# mypyの詳細出力 mypy evospikenet/ --show-error-codes --show-traceback
パフォーマンス問題の診断
import time
import cProfile
def profile_method():
"""メソッドのパフォーマンスプロファイリング"""
client = EvoSpikeNetAPIClient()
start_time = time.time()
cProfile.runctx('client.generate("Test prompt")', globals(), locals(), 'profile.out')
end_time = time.time()
print(f"Execution time: {end_time - start_time:.3f}s")
# プロファイル結果分析
import pstats
stats = pstats.Stats('profile.out')
stats.sort_stats('cumulative').print_stats(10)
貢献ガイドライン
コーディング標準
- PEP 8準拠: Blackとisortを使用
- 型ヒント: 全てのメソッドに型ヒントを付ける
- ドキュメント: docstringはGoogleスタイル
- テスト: 新しい機能にはテストを追加
- コミット: 意味のあるコミットメッセージ
プルリクエストテンプレート
## 概要
何を実装/修正したか
## 変更点
- 変更1
- 変更2
## テスト
- [ ] 単体テスト追加
- [ ] 統合テスト確認
- [ ] 手動テスト実施
## 確認事項
- [ ] ドキュメント更新
- [ ] 変更履歴更新
- [ ] 後方互換性維持
このガイドはSDKの開発を進める上で必要な情報を提供します。質問があれば、開発チームまでお問い合わせください。