ネットワーククライアント

ErisPulse は、HTTP リクエスト、WebSocket 接続、接続プール管理を統合した統一されたネットワーククライアントを提供しています。モジュールやアダプターは、このクライアントを優先して使用する必要があります。aiohttp / httpx / requests などのサードパーティライブラリを直接インポートしてはいけません。

概要

ネットワーククライアントの主な機能：

• 統一されたインターフェース：get / post / put / delete / patch / request メソッドを提供
• WebSocket クライアント：ws_connect を使用してクライアント WebSocket 接続を確立
• 自動ログ記録：すべてのリクエストに対して自動的にログと統計情報を記録
• ライフサイクル統合：リクエストごとに client.request ライフサイクルイベントがトリガーされ、WS 接続は client.ws.connect イベントをトリガー
• リトライサポート：自動リトライ回数と間隔を設定可能
• タイムアウト制御：接続タイムアウトとリクエストタイムアウトを個別に制御
• 接続プールの再利用：aiohttp.ClientSession に基づく接続プール管理
• 例外体系：aiohttp の例外を ErisPulse の例外 (ClientError 体系) に自動的に変換

速習

HTTP リクエスト

from ErisPulse.Core import client

# GET リクエスト
resp = await client.get("https://httpbin.org/get")
data = await resp.json()
print(resp.status)  # 200

# POST リクエスト
resp = await client.post(
    "https://httpbin.org/post",
    json={"key": "value"},
)
data = await resp.json()

WebSocket 接続

from ErisPulse import client   # 2.10+ からのルートインポート（ErisPulse.Core.client の深いパスも互換性あり）

ws = await client.ws_connect("wss://example.com/ws")

async for text in ws.iter_text():
    await ws.send_text(f"Echo: {text}")

出力接続の自動プール登録（2.10+）

ws_connect で確立された接続はデフォルトでフレームワークの接続プールに登録されます（track=False で無効化可能）： connections.list(owner=...) で確認でき、その接続にブロードキャストが可能です。接続が閉じられたり、リモートが切断されたりすると自動的に登録解除されます。 モジュールのアンロード / アダプターの停止時に、フレームワークがその名下のすべての出力接続を一括で閉じるため、リークを防ぎます。

from ErisPulse import client, connections

ws = await client.ws_connect("wss://example.com/ws")
print(ws.id, ws.owner)   # 登録済み：接続 ID と所有者

# owner コンテキストが利用できない場合（例：ツールスレッドのコールバック）は、明示的に所有者を指定し、アンロード時に自動回収されるようにします：
ws2 = await client.ws_connect("wss://example.com/ws2", owner="MyAdapter")

接続プールとブロードキャストの詳細は 接続プールとブロードキャスト を参照してください。

HttpResponse

すべてのリクエストメソッドは HttpResponse オブジェクトを返します：

from ErisPulse.Core import client

resp = await client.get("https://httpbin.org/get")

resp.status       # int - HTTP ステータスコード (例: 200, 404)
resp.reason       # str | None - ステータスの説明 (例: "OK")
resp.headers      # 応答ヘッダー (大文字小文字を区別しない)
resp.content_type # str | None - Content-Type
resp.url          # 最終的な URL (リダイレクトにより変更される可能性がある)
resp.raw          # ベースの生の応答オブジェクト (現在は aiohttp.ClientResponse)

# 応答本文の読み取り
body = await resp.read()       # bytes
text = await resp.text()       # str
data = await resp.json()       # JSON を解析
text = await resp.text("gbk")  # 指定されたエンコーディング

リクエストメソッド

GET

from ErisPulse.Core import client

resp = await client.get(
    "https://api.example.com/users",
    params={"page": "1", "limit": "10"},
    headers={"Authorization": "Bearer token"},
)

POST

from ErisPulse.Core import client

# JSON リクエスト本文
resp = await client.post(
    "https://api.example.com/users",
    json={"name": "Alice", "age": 30},
)

# フォームリクエスト本文
resp = await client.post(
    "https://api.example.com/login",
    data={"username": "admin", "password": "123"},
)

# 原始データ
resp = await client.post(
    "https://api.example.com/upload",
    data=b"raw bytes",
    headers={"Content-Type": "application/octet-stream"},
)

# ファイルアップロード (files パラメータを使用、aiohttp をインポートする必要はない)
# 形式: {フィールド名: ファイルオブジェクト/bytes/(filename, file)/(filename, file, content_type)}
resp = await client.post(
    "https://api.example.com/upload",
    data={"description": "プロフィール画像"},            # 任意: 通常のフォームフィールドも同時に送信可能
    files={
        "file": ("photo.png", open("photo.png", "rb"), "image/png"),
    },
)

# 簡単な書き方: ファイルオブジェクトを直接渡す
resp = await client.post(
    "https://api.example.com/upload",
    files={"file": open("photo.png", "rb")},
)

# メモリ内のデータを直接アップロード (ファイルに落とす必要はない)
import io

resp = await client.post(
    "https://api.example.com/upload",
    files={"file": ("data.txt", io.BytesIO(b"file content"), "text/plain")},
)

PUT / DELETE / PATCH

from ErisPulse.Core import client

resp = await client.put("https://api.example.com/users/1", json={"name": "Bob"})
resp = await client.delete("https://api.example.com/users/1")
resp = await client.patch("https://api.example.com/users/1", json={"age": 31})

一般的な request

from ErisPulse.Core import client

resp = await client.request(
    "OPTIONS",
    "https://api.example.com/resource",
    headers={"Origin": "https://example.com"},
)

パラメータの説明

HTTP リクエストパラメータ

ws_connect パラメータ

タイムアウトとリトライ

from ErisPulse.Core import Client

# タイムアウトをカスタムしたクライアントを作成
client = Client(
    timeout=60,           # リクエスト全体のタイムアウト 60秒
    connect_timeout=5,    # 接続のタイムアウト 5秒
    max_retries=3,        # 失敗時に自動リトライ 3回
    retry_delay=2,        # リトライ間隔 2秒
)

# 単回のリクエストでタイムアウトを上書き
resp = await client.get("https://slow-api.example.com/data", timeout=120)

デフォルトヘッダーのカスタマイズ

client = Client(
    headers={
        "Authorization": "Bearer token",
        "X-App-Id": "my-app",
    },
    user_agent="MyBot/1.0",
)

リクエスト統計

from ErisPulse.Core import client

# 統計情報の確認
stats = client.stats
# {"total_requests": 42, "total_errors": 1, "total_bytes_sent": 0, "total_bytes_received": 0}

# 統計情報のリセット
client.reset_stats()

ライフサイクルイベント

HTTP リクエストイベント

リクエスト完了後に client.request イベントがトリガーされ、監視に利用できます：

from ErisPulse.Core import lifecycle

@lifecycle.on("client.request")
async def on_request(event_data):
    print(f"{event_data['method']} {event_data['url']} -> {event_data['status']} ({event_data['elapsed']}s)")

WebSocket 接続イベント

WebSocket 接続確立後に client.ws.connect イベントがトリガーされます：

from ErisPulse.Core import lifecycle

@lifecycle.on("client.ws.connect")
async def on_ws_connect(event_data):
    print(f"WS 接続: {event_data['url']}")

コンテキストマネージャー

# コンテキストマネージャーとして使用し、セッションを自動的に閉じる
async with Client(timeout=30) as client:
    resp = await client.get("https://httpbin.org/get")
    data = await resp.json()

WebSocket クライアント

client.ws_connect() を使用して WebSocket クライアント接続を確立し、ClientWebSocket オブジェクトを返します。クライアントとサーバーの WebSocket は同じ WebSocketConnectionBase 基底クラスを共有し、send/receive/iter のインターフェースは完全に一致します。

基本的な使い方

from ErisPulse.Core import client

ws = await client.ws_connect("wss://example.com/ws", heartbeat=30)

await ws.send_text("Hello")
await ws.send_bytes(b"\x00\x01\x02")
await ws.send_json({"type": "ping"})

メッセージの受信

高レベルメソッド（推奨）

メッセージの種類を自動的にフィルタリングし、切断時に WebSocketDisconnect をスローします：

from ErisPulse.Core import client
from ErisPulse.Core.Bases.errors import WebSocketDisconnect

ws = await client.ws_connect("wss://example.com/ws")

# 単一のメッセージ受信
text = await ws.receive_text()    # str
data = await ws.receive_bytes()   # bytes
obj = await ws.receive_json()     # dict / list

# メッセージのイテレーション受信 (切断時に自動的に停止)
async for text in ws.iter_text():
    print(text)

async for data in ws.iter_bytes():
    print(data)

async for obj in ws.iter_json():
    print(obj)

低レベルメソッド

receive() と iter_messages() を使用して、メッセージの種類を識別し、TEXT / BINARY / CLOSE / ERROR を区別します：

from ErisPulse.Core import client
from ErisPulse.Core.Bases.websocket import WSMessage

ws = await client.ws_connect("wss://example.com/ws")

# 単一のメッセージ受信
msg = await ws.receive()
# msg.type  -> WSMessage.TEXT / WSMessage.BINARY / WSMessage.CLOSE / WSMessage.ERROR
# msg.data  -> str | bytes | None

# メッセージのイテレーション受信 (CLOSE/ERROR の時点で自動的に停止)
async for msg in ws.iter_messages():
    if msg.type == WSMessage.TEXT:
        print(f"テキスト: {msg.data}")
    elif msg.type == WSMessage.BINARY:
        print(f"バイナリ: {len(msg.data)} bytes")

WSMessage

WSMessage は、下層ライブラリに依存しない統一された WebSocket メッセージタイプです：

ClientWebSocket の属性

ライフサイクルフック

サービス側 WebSocketConnection と同様に、on_disconnect と on_error のコールバックをサポートします：

from ErisPulse.Core import client

ws = await client.ws_connect("wss://example.com/ws")

@ws.on_disconnect
async def handle_disconnect(ws, reason="unknown"):
    print(f"接続が切断されました: {reason}")

@ws.on_error
async def handle_error(ws, error=""):
    print(f"接続エラー: {error}")

接続の閉じ

await ws.close(code=1000, reason="正常な切断")

例外体系

ErisPulse は、統一された例外階層を定義しており、sdk.client から発生するリクエストは、aiohttp の例外を自動的に ErisPulse の例外に変換します。

後方互換性：aiohttp.ClientSession を直接使用する旧モジュール/アダプターは完全に影響を受けません。例外の変換は、sdk.client からリクエストを発生させた場合にのみ有効で、aiohttp を直接使用するコードは、aiohttp.ClientError などの元の例外をキャッチし続けます。両方の方法を共存させることができます。

例外階層

ErisPulseError
├── ClientError                  # HTTP/WS クライアントリクエスト例外の基底クラス
│   ├── ClientConnectionError    # 接続失敗 (DNS 解析失敗、接続拒否、ネットワーク不可)
│   ├── ClientTimeoutError       # 接続タイムアウトまたはリクエストタイムアウト
│   └── HTTPStatusError          # HTTP 4xx/5xx ステータスコードエラー
└── WebSocketError               # WebSocket 例外の基底クラス
    └── WebSocketDisconnect      # WebSocket 接続切断 (クライアントとサーバー共通)

例外のキャッチ

from ErisPulse.Core import client
from ErisPulse.Core.Bases.errors import (
    ClientError,
    ClientConnectionError,
    ClientTimeoutError,
    HTTPStatusError,
    WebSocketDisconnect,
    WebSocketError,
)

# HTTP リクエスト例外の処理
try:
    resp = await client.get("https://api.example.com/data")
    data = await resp.json()
except ClientConnectionError:
    print("サーバーに接続できません")
except ClientTimeoutError:
    print("リクエストがタイムアウトしました")
except ClientError as e:
    print(f"リクエストが失敗しました: {e}")

# WebSocket 例外の処理
try:
    ws = await client.ws_connect("wss://example.com/ws")
    async for text in ws.iter_text():
        await ws.send_text(f"Echo: {text}")
except WebSocketDisconnect as e:
    print(f"接続が切断されました: code={e.code}, reason={e.reason}")
except WebSocketError as e:
    print(f"WebSocket エラー: {e}")

統一的なキャッチ

ClientError を使用して、HTTP/WS クライアントリクエストのすべての例外を一括でキャッチします：

from ErisPulse.Core.Bases.errors import ClientError

try:
    resp = await client.get("https://api.example.com/data")
except ClientError as e:
    print(f"クライアントエラー: {e}")

HTTPStatusError

リクエスト後にステータスコードを確認して例外をスローする必要がある場合、手動で使用できます：

from ErisPulse.Core.Bases.errors import HTTPStatusError

resp = await client.get("https://api.example.com/data")
if resp.status >= 400:
    raise HTTPStatusError(resp.status, await resp.text())

アダプターでの使用

アダプターはグローバルクライアントまたは独自のクライアントインスタンスを使用して、プラットフォームAPIリクエストを送信できます：

from ErisPulse.Core import client
from ErisPulse.Core.Bases import BaseAdapter
from ErisPulse.Core.Bases.errors import ClientError

class MyAdapter(BaseAdapter):
    async def call_api(self, endpoint, **params):
        try:
            resp = await client.post(
                f"https://api.platform.com/{endpoint}",
                json=params,
                headers={"Authorization": f"Bearer {self.token}"},
            )
            return await resp.json()
        except ClientError as e:
            self.logger.error(f"API 調用が失敗しました: {e}")
            raise

from ErisPulse import sdk を使用して sdk.client を使うこともできます。効果は同じです。

最適な実践

1. グローバルクライアントを優先する：from ErisPulse.Core import client を使用してグローバルシングルトンを取得し、フレームワークによる統一的な管理と監視が可能
2. aiohttp の直接インポートを避ける：client を使用して aiohttp.ClientSession を代用し、将来の下層実装の変更に伴うコード変更を必要としない。旧コードは直接 aiohttp を使用しても正常に動作し、両方の方法を共存させることができる
3. ErisPulse 例外体系を使用する：sdk.client からリクエストする際は aiohttp.ClientError ではなく ClientError をキャッチし、特定の HTTP ライブラリに依存しないコードを保証する。直接 aiohttp を使用する旧コードは影響を受けない
4. 適切なタイムアウトを設定する：API の応答速度に応じて適切なタイムアウト時間を設定し、長時間のブロッキングを防ぐ
5. リトライメカニズムを使用する：不安定な API にはリトライを有効にして信頼性を高める
6. リクエスト統計を監視する：sdk.client.stats または client.request ライフサイクルイベントを使用してリクエスト状況を監視する
7. WebSocket では高レベルメソッドを使用する：iter_text / iter_json などの高レベルメソッドを優先し、メッセージの種類を識別する必要がある場合にのみ iter_messages を使用する

関連ドキュメント

• ルーターマネージャー - HTTP/WebSocket サーバーサイドルーティング（クライアントとサーバーの WebSocketConnection は同一基底クラスを共有）
• アダプター開発ガイド - アダプターで HTTP クライアントを使用する
• ライフサイクル管理 - リクエストイベントを監視する