Core Concepts

This guide introduces the core concepts of ErisPulse, helping you understand the framework's design philosophy and basic architecture.

Event-Driven Architecture

ErisPulse adopts an event-driven architecture, where all interactions are passed and processed through events.

Event Flow

User sends message
      │
      ▼
Platform receives
      │
      ▼
Adapter receives native platform event
      │
      ▼
Converts to OneBot12 standard event
      │
      ▼
Submits to event system
      │
      ▼
Distributes to registered handlers
      │
      ▼
Module processes event
      │
      ▼
Sends response via adapter
      │
      ▼
Platform displays to user

OneBot12 Standard

ErisPulse uses OneBot12 as its core event standard. OneBot12 is a generic chatbot application programming interface standard that defines a unified event format.

All adapters convert platform-specific events into OneBot12 format, ensuring code consistency.

Core Components

1. SDK Object

The SDK is the unified entry point for all features, providing access to core components.

from ErisPulse import sdk

# Access core modules
sdk.storage    # Storage system
sdk.config     # Configuration system
sdk.logger     # Logging system
sdk.adapter    # Adapter system
sdk.module     # Module system
sdk.router     # Router system
sdk.client     # HTTP client
sdk.lifecycle  # Lifecycle system

2. Event Object

The Event object encapsulates event data and provides convenient access methods.

@command("info")
async def info_handler(event):
    # Get event information
    event_id = event.get_id()
    user_id = event.get_user_id()
    platform = event.get_platform()
    text = event.get_text()
    
    # Send reply
    await event.reply(f"User: {user_id}, Platform: {platform}")

3. Adapters

Adapters serve as bridges between ErisPulse and external platforms.

Responsibilities:

Receive native platform events
Convert them into OneBot12 standard format
Send standard format events back to the platform

Example Adapters:

Yunhu adapter: Communicates with the Yunhu platform
Telegram adapter: Communicates with the Telegram Bot API
OneBot11 adapter: Communicates with OneBot11-compatible applications
Email adapter: Handles email sending and receiving

4. Modules

Modules are the basic units for functionality extensions, capable of:

Registering event handlers
Implementing business logic
Calling adapters to send messages
Using services provided by core modules

Module Discovery Mechanism

ErisPulse discovers installed modules via Python's importlib.metadata.entry_points. Modules declare entry points in pyproject.toml:

[project.entry-points."erispulse.module"]
MyModule = "my_package:Main"

During SDK initialization, all entry points under the erispulse.module group are scanned, the module class is registered to ModuleManager, and then initialized in topological order based on dependencies.

Minimal Viable Module

from ErisPulse.Core.Bases import BaseModule
from ErisPulse import sdk

class Main(BaseModule):
    def __init__(self):
        self.sdk = sdk
        self.logger = sdk.logger.get_child("MyModule")

    async def on_load(self, event):
        self.logger.info("Module loaded")

    async def on_unload(self, event):
        self.logger.info("Module unloaded")

Module Lifecycle

Registration: SDK discovers the module class and registers it with the manager
Loading: Creates the module instance and calls on_load(event) (event = {"module_name": "MyModule"})
Unloading: Calls on_unload(event) and cleans up resources

Loading Strategy

Declare the module's loading behavior using get_load_strategy():

from ErisPulse.loaders import ModuleLoadStrategy

class Main(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(
            lazy_load=True,   # Whether to lazy load (default: True)
            priority=0        # Loading priority; higher values are initialized first
        )

lazy_load=True (default): The module is initialized only when first accessed via sdk.MyModule, reducing startup time
lazy_load=False: The module is initialized immediately during SDK startup, suitable for modules that need to listen to lifecycle events or execute scheduled tasks
priority: Modules with the same priority are loaded in registration order; higher values are initialized first

For detailed information about the lazy loading mechanism, refer to Lazy Loading System.

Event Types

ErisPulse supports five types of events:

Event Type	Decorator	Description
Message Event	`@message.on_message()`	Any message sent by the user (private chat, group chat)
Command Event	`@command("name")`	Messages starting with a command prefix (e.g., `/hello`)
Notice Event	`@notice.on_friend_add()` etc.	System notifications (friend added, group member changes, etc.)
Request Event	`@request.on_friend_request()` etc.	User requests (friend requests, group invitations)
Meta Event	`@meta.on_connect()` etc.	System-level events (connection, disconnection, heartbeat)

For detailed usage and code examples of each event type, refer to Event Handling Introduction.

Core Module Descriptions

Storage (Storage)

A key-value storage system based on SQLite, used for persistent data storage.

# Set value
sdk.storage.set("key", "value")

# Get value
value = sdk.storage.get("key", "default_value")

# Batch operations
sdk.storage.set_multi({
    "key1": "value1",
    "key2": "value2"
})

# Transactions
with sdk.storage.transaction():
    sdk.storage.set("key1", "value1")
    sdk.storage.set("key2", "value2")

Config (Configuration)

TOML-based configuration file management.

# Get configuration
config = sdk.config.getConfig("MyModule", {})

# Set configuration
sdk.config.setConfig("MyModule", {"key": "value"})

# Read nested configuration
value = sdk.config.getConfig("MyModule.subkey", "default")

Logger (Logging)

A modular logging system.

# Log messages
sdk.logger.info("This is an info message")
sdk.logger.warning("This is a warning message")
sdk.logger.error("This is an error message")

# Get child logger
child_logger = sdk.logger.get_child("submodule")
child_logger.info("Submodule log message")

Attribute Access Syntactic Sugar

In addition to using the get_child() method, you can also create a child logger using attribute access, which is a more concise syntactic sugar:

# Create child logger using attribute access
sdk.logger.mymodule.info("Module message")

# Supports nested access
sdk.logger.mymodule.database.info("Database message")

Router (Routing)

HTTP and WebSocket routing management, based on FastAPI + Uvicorn. Supports decorator-based routing, middleware, grouping, rate limiting, and CORS.

from ErisPulse.Core import HttpRequest

@sdk.router.get("MyModule", "/api")
async def handler(request: HttpRequest):
    data = await request.json()
    return {"status": "ok"}

For the complete routing API (WebSocket, middleware, rate limiting, CORS, etc.), refer to Router Manager.

Client (Network Client)

A unified network client that aggregates HTTP requests, WebSocket connections, connection pool management, automatic retries, timeout control, request statistics, and lifecycle event integration.

from ErisPulse.Core import client

# HTTP request
resp = await client.get("https://api.example.com/users")
data = await resp.json()

# With retry and timeout
resp = await client.get(url, timeout=30, max_retries=3)

# WebSocket connection
ws = await client.ws_connect("wss://example.com/ws")
async for text in ws.iter_text():
    await ws.send_text(f"Echo: {text}")

For the complete network client API, refer to Network Client.

SendDSL Message Sending

Adapters provide a chain-call interface for message sending.

Basic Sending

# Get adapter instance
yunhu = sdk.adapter.get("yunhu")

# Send message
await yunhu.Send.To("user", "U1001").Text("Hello")

# Specify sending account
await yunhu.Send.Using("bot1").To("group", "G1001").Text("Group message")

Chain Modifiers

# @ user
await yunhu.Send.To("group", "G1001").At("U2001").Text("@ message")

# Reply to message
await yunhu.Send.To("group", "G1001").Reply("msg123").Text("Reply")

# @ all
await yunhu.Send.To("group", "G1001").AtAll().Text("Announcement")

Event Reply Methods

The Event object provides convenient reply methods:

@command("test")
async def test_handler(event):
    # Simple text reply
    await event.reply("Reply content")
    
    # Send image
    await event.reply("http://example.com/image.jpg", method="Image")
    
    # Send voice
    await event.reply("http://example.com/voice.mp3", method="Voice")

Lazy Loading System

ErisPulse enables module lazy loading by default. Modules are only initialized when first accessed (e.g., sdk.MyModule), significantly improving startup speed.

from ErisPulse.loaders import ModuleLoadStrategy

class Main(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(
            lazy_load=True,   # Enable lazy loading (default)
            priority=0        # Loading priority; higher values are initialized first
        )

Scenarios requiring disabling lazy loading (lazy_load=False):

Modules that listen to lifecycle events (e.g., core.init.complete)
Modules that start scheduled tasks or background services
Modules that need to complete initialization before other modules load

For detailed information about the lazy loading mechanism and considerations, refer to Lazy Loading System.

Next Steps

Event Handling Introduction - Learn how to handle various types of events
Common Task Examples - Master the implementation of common features