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 by preference, rather than importing third-party libraries such as aiohttp, httpx, or requests.

Overview

The main features of the network client:

• Unified Interface: Provides get / post / put / delete / patch / request methods
• WebSocket Client: Establishes a client WebSocket connection via ws_connect
• Automatic Logging: All requests are automatically logged and statistics are recorded
• Lifecycle Integration: Each request triggers the client.request lifecycle event, and WS 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 system)

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.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

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          # underlying native 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: include 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 data directly from memory (no disk storage required)
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})

General request

from ErisPulse.Core import client

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

Parameters

HTTP Request Parameters

ws_connect Parameters

Timeouts and Retries

from ErisPulse.Core import Client

# Create a client with custom timeouts
client = Client(
    timeout=60,           # Total request timeout of 60s
    connect_timeout=5,    # Connection timeout of 5s
    max_retries=3,        # Automatic retry 3 times on failure
    retry_delay=2,        # Retry interval of 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 and can be used 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

Establish a WebSocket client connection using client.ws_connect(), which returns a ClientWebSocket object. The client and server WebSocket share the same base class WebSocketConnectionBase, and their send/receive/iter interfaces are completely consistent.

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

# Receive single message
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, allowing differentiation 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")

# Receive single raw message
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 disconnected: {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 System

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

Backward Compatibility: Old modules/adapters that directly use aiohttp.ClientSession are completely unaffected. Exception conversion only takes effect when requests are initiated through sdk.client. Code that directly uses aiohttp continues to catch native exceptions such as aiohttp.ClientError. 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               # Base class for WebSocket exceptions
    └── WebSocketDisconnect      # WebSocket connection disconnected (applicable to both 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 disconnected: 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 uniformly:

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 use it manually:

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 sdk.client via from ErisPulse import sdk, which has the same effect.

Best Practices

1. Prefer using the global client: Use from ErisPulse.Core import client to obtain the global singleton, which facilitates unified management and monitoring by the framework.
2. Avoid directly importing aiohttp: Use client instead of aiohttp.ClientSession, so that future changes to the underlying implementation do not require code modifications. Code that directly uses aiohttp will continue to work normally, and both approaches can coexist.
3. Use ErisPulse's exception system: When making requests via sdk.client, catch ClientError instead of aiohttp.ClientError to ensure your code does not depend on a specific HTTP library. Code that directly uses aiohttp remains unaffected.
4. Set timeouts appropriately: Set reasonable timeout values based on the API response speed to avoid prolonged blocking.
5. Use retry mechanisms: Enable retries for unstable APIs to improve reliability.
6. Monitor request statistics: Monitor request situations through sdk.client.stats or lifecycle events of client.request.
7. Use advanced methods for WebSocket: Prefer advanced methods such as iter_text / iter_json, and use iter_messages only when distinguishing message types is necessary.

Related Documentation

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