網路客戶端

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.Core import client

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

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

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/(檔名, 檔)/(檔名, 檔, 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,           # 請求總超時 60s
    connect_timeout=5,    # 連接超時 5s
    max_retries=3,        # 失敗自動重試 3 次
    retry_delay=2,        # 重試間隔 2s
)

# 單次請求覆蓋超時
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="Normal closure")

異常體系

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 請求時捕獲 ClientError 而非 aiohttp.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 客戶端
• 生命週期管理 - 監聽請求事件