Network Client

ErisPulse provides a unified network client that aggregates HTTP requests, WebSocket connections, and connection pool management. Modules and adapters must use this client instead of directly importing third-party libraries such as aiohttp / httpx / requests.

Overview

The main features of the network client are:

• Unified interface: Provides get / post / put / delete / patch / request methods
• WebSocket client: Establishes client WebSocket connections via ws_connect
• Automatic logging: All requests are automatically logged and statistics are recorded
• Lifecycle integration: Each request triggers the client.request lifecycle event; WebSocket connections trigger the client.ws.connect event
• Retry support: Configurable automatic retry count and interval
• Timeout control: Independent connection timeout and request timeout
• Connection pool reuse: Connection pool management based on aiohttp.ClientSession
• Exception system: aiohttp exceptions are automatically converted to ErisPulse exceptions (ClientError hierarchy)

Quick Start

HTTP Requests

from ErisPulse.Core import client

# GET request
resp = await client.get("https://httpbin.org/get")
data = await resp.json()
print(resp.status)  # 200

# POST request
resp = await client.post(
    "https://httpbin.org/post",
    json={"key": "value"},
)
data = await resp.json()

WebSocket Connection

from ErisPulse import client   # Root import from 2.10+ (ErisPulse.Core.client deep path still compatible)

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

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

Outbound Connections Auto-Registered in Pool (2.10+)

Connections established via ws_connect are automatically registered into the framework connection pool (set track=False to disable): They can be viewed via connections.list(owner=...) and messages can be broadcast to them; connections are automatically unregistered upon closure or remote disconnection; All outbound connections under a module or adapter are uniformly closed by the framework when the module is unloaded or the adapter stops, eliminating leaks.

from ErisPulse import client, connections

ws = await client.ws_connect("wss://example.com/ws")
print(ws.id, ws.owner)   # Already registered: connection id and owner

# When owner context is unavailable (e.g., tool thread callback), explicitly specify the owner for automatic cleanup upon unload:
ws2 = await client.ws_connect("wss://example.com/ws2", owner="MyAdapter")

See Connection Pool and Broadcasting for more details.

HttpResponse

All request methods return an HttpResponse object:

from ErisPulse.Core import client

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

resp.status       # int - HTTP status code (e.g., 200, 404)
resp.reason       # str | None - status description (e.g., "OK")
resp.headers      # response headers (case-insensitive)
resp.content_type # str | None - Content-Type
resp.url          # final URL (may change due to redirects)
resp.raw          # raw underlying response object (currently aiohttp.ClientResponse)

# Read response body
body = await resp.read()       # bytes
text = await resp.text()       # str
data = await resp.json()       # parse JSON
text = await resp.text("gbk")  # specify encoding

Request Methods

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 request body
resp = await client.post(
    "https://api.example.com/users",
    json={"name": "Alice", "age": 30},
)

# Form request body
resp = await client.post(
    "https://api.example.com/login",
    data={"username": "admin", "password": "123"},
)

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

# File upload (using files parameter, no need to import aiohttp)
# Format: {field name: file object/bytes/(filename, file)/(filename, file, content_type)}
resp = await client.post(
    "https://api.example.com/upload",
    data={"description": "Avatar"},            # Optional: also carry regular form fields
    files={
        "file": ("photo.png", open("photo.png", "rb"), "image/png"),
    },
)

# Simplified syntax: directly pass file object
resp = await client.post(
    "https://api.example.com/upload",
    files={"file": open("photo.png", "rb")},
)

# Upload in-memory data directly (no need to write to disk)
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})

Generic request

from ErisPulse.Core import client

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

Parameter Descriptions

HTTP Request Parameters

ws_connect Parameters

Timeout and Retry

from ErisPulse.Core import Client

# Create a client with custom timeout settings
client = Client(
    timeout=60,           # Total request timeout 60s
    connect_timeout=5,    # Connection timeout 5s
    max_retries=3,        # Retry failed requests 3 times
    retry_delay=2,        # Retry interval 2s
)

# Override timeout for a single request
resp = await client.get("https://slow-api.example.com/data", timeout=120)

Custom Default Headers

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

Request Statistics

from ErisPulse.Core import client

# View statistics
stats = client.stats
# {"total_requests": 42, "total_errors": 1, "total_bytes_sent": 0, "total_bytes_received": 0}

# Reset statistics
client.reset_stats()

Lifecycle Events

HTTP Request Events

The client.request event is triggered after each request completes, useful for monitoring:

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 Connection Events

The client.ws.connect event is triggered after each WebSocket connection is established:

from ErisPulse.Core import lifecycle

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

Context Management

# As a context manager, automatically closes the session
async with Client(timeout=30) as client:
    resp = await client.get("https://httpbin.org/get")
    data = await resp.json()

WebSocket Client

WebSocket client connections are established via client.ws_connect(), returning a ClientWebSocket object. The client and server WebSocket share the same WebSocketConnectionBase base class, with identical send/receive/iter interfaces.

Basic Usage

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"})

Receiving Messages

High-Level Methods (Recommended)

Automatically filter message types, and raise WebSocketDisconnect on disconnection:

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

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

# Single message receive
text = await ws.receive_text()    # str
data = await ws.receive_bytes()   # bytes
obj = await ws.receive_json()     # dict / list

# Iterate messages (automatically stops on disconnection)
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)

Low-Level Methods

Use receive() and iter_messages() to handle raw message types, distinguishing between 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")

# Single raw message receive
msg = await ws.receive()
# msg.type  -> WSMessage.TEXT / WSMessage.BINARY / WSMessage.CLOSE / WSMessage.ERROR
# msg.data  -> str | bytes | None

# Iterate raw messages (automatically stops on CLOSE/ERROR)
async for msg in ws.iter_messages():
    if msg.type == WSMessage.TEXT:
        print(f"Text: {msg.data}")
    elif msg.type == WSMessage.BINARY:
        print(f"Binary: {len(msg.data)} bytes")

WSMessage

WSMessage is a unified WebSocket message type, independent of the underlying library:

ClientWebSocket Properties

Lifecycle Hooks

Consistent with server-side WebSocketConnection, supports on_disconnect and on_error callbacks:

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"Connection closed: {reason}")

@ws.on_error
async def handle_error(ws, error=""):
    print(f"Connection error: {error}")

Closing the Connection

await ws.close(code=1000, reason="Normal closure")

Exception Hierarchy

ErisPulse defines a unified exception hierarchy. Requests initiated via sdk.client automatically convert underlying aiohttp exceptions into ErisPulse exceptions.

Backward Compatibility: Old modules/adapters that directly use aiohttp.ClientSession are unaffected. Exception conversion only occurs when requests are made via sdk.client. Code using aiohttp directly still catches aiohttp.ClientError and other native exceptions. Both approaches can coexist.

Exception Hierarchy

ErisPulseError
├── ClientError                  # Base class for all HTTP/WS client request exceptions
│   ├── ClientConnectionError    # Connection failed (DNS resolution failed, connection refused, network unreachable)
│   ├── ClientTimeoutError       # Connection timeout or request timeout
│   └── HTTPStatusError          # HTTP 4xx/5xx status code errors
└── WebSocketError               # WebSocket exception base class
    └── WebSocketDisconnect      # WebSocket connection closed (common for client and server)

Exception Handling

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

# HTTP request exception handling
try:
    resp = await client.get("https://api.example.com/data")
    data = await resp.json()
except ClientConnectionError:
    print("Unable to connect to the server")
except ClientTimeoutError:
    print("Request timed out")
except ClientError as e:
    print(f"Request failed: {e}")

# WebSocket exception handling
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"Connection closed: code={e.code}, reason={e.reason}")
except WebSocketError as e:
    print(f"WebSocket error: {e}")

Unified Exception Handling

Use ClientError to catch all HTTP/WS client request exceptions:

from ErisPulse.Core.Bases.errors import ClientError

try:
    resp = await client.get("https://api.example.com/data")
except ClientError as e:
    print(f"Client error: {e}")

HTTPStatusError

When you need to check the status code after a request and raise an exception, you can manually use:

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())

Using in Adapters

Adapters can use the global client or create their own client instance to send platform API requests:

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 call failed: {e}")
            raise

You can also use from ErisPulse import sdk to access sdk.client, with the same effect.

Best Practices

1. Prefer the global client: Use from ErisPulse.Core import client to obtain the global singleton, facilitating unified management and monitoring by the framework.
2. Avoid directly importing aiohttp: Use client instead of aiohttp.ClientSession, so future changes to the underlying implementation require no code changes. Old code using aiohttp directly still works normally, and both approaches can coexist.
3. Use the ErisPulse exception hierarchy: When using sdk.client for requests, catch ClientError instead of aiohttp.ClientError to ensure code independence from specific HTTP libraries. Old code using aiohttp directly remains unaffected.
4. Set timeouts appropriately: Set reasonable timeout values based on API response speed to avoid long blocking.
5. Use retry mechanisms: Enable retries for unstable APIs to improve reliability.
6. Monitor request statistics: Use sdk.client.stats or the client.request lifecycle event to monitor request status.
7. Use high-level WebSocket methods: Prefer high-level methods such as iter_text / iter_json, and only use iter_messages when distinguishing message types is necessary.

Related Documentation

• Router Manager - HTTP/WebSocket server-side routing (server-side WebSocketConnection shares the same base class with the client)
• Adapter Development Guide - Using HTTP client in adapters
• Lifecycle Management - Listening to request events