API 参考手册

本文档详细介绍了 Python SDK 提供的所有 API 接口，包括核心功能、配置管理、日志系统、异常处理等模块的完整 API 参考。

核心特性

完整覆盖：涵盖所有核心 API 和工具函数
详细文档：每个 API 都有完整的参数说明和示例
实用示例：提供真实场景的使用案例
类型安全：完整的类型注解和参数验证
高性能：优化的实现和最佳实践指导

API 分类概览

模块	描述	主要功能
ServerApplication	服务器应用核心	应用初始化、路由管理、中间件
GConfig	配置管理	配置文件加载、环境变量、动态配置
GLogger	日志系统	结构化日志、多输出、性能监控
RPCException	异常处理	自定义异常、错误码、异常链
中间件系统	请求处理	认证、限流、CORS、请求日志
工具函数	辅助功能	数据验证、格式化、加密解密

simplejrpc 模块

ServerApplication 类

ServerApplication 是一个 JSON-RPC 服务器应用程序类，用于创建和管理基于 Unix 域套接字的 RPC 服务。该类提供了配置管理、路由注册、参数验证、中间件支持和日志设置等功能。

核心功能

路由管理：支持装饰器方式注册 RPC 方法
参数验证：集成表单验证机制
中间件支持：可扩展的请求处理管道
配置管理：支持多种配置文件格式
日志集成：内置日志记录功能

快速开始

from simplejrpc import ServerApplication

# 创建服务器实例
app = ServerApplication("/tmp/rpc.sock")

# 注册 RPC 方法
@app.route("hello")
def hello(name: str):
    return f"Hello, {name}!"

# 启动服务器
app.run()

`init(socket_path: str, config: Optional[object] = Settings(), config_path: Optional[str] = None)`

初始化 ServerApplication 实例。

参数:

socket_path: 字符串类型，Unix 域套接字路径
config: 可选参数，配置对象，默认为 Settings() 实例
config_path: 可选参数，字符串类型，配置文件路径

返回值: 无

示例:

# 基本初始化
app = ServerApplication("/tmp/my_rpc.sock")

# 使用配置文件初始化
app = ServerApplication(
    socket_path="/tmp/my_rpc.sock",
    config_path="config.yaml"
)

# 使用自定义配置对象
from simplejrpc import Settings
config = Settings({"debug": True})
app = ServerApplication("/tmp/my_rpc.sock", config=config)

`from_config(config_content: Optional[Dict[str, Any]] = None, config_path: Optional[str] = None)`

从配置内容或配置文件加载配置。

参数:

config_content: 可选参数，字典类型，配置内容
config_path: 可选参数，字符串类型，配置文件路径

返回值:

返回配置对象 (Settings 实例)

`route(name: Optional[str] = None, form: Optional[Form] = BaseForm, fn=None)`

路由装饰器，用于注册 RPC 方法并提供参数验证功能。

参数:

name: 可选参数，字符串类型，RPC 方法名称，默认使用函数名
form: 可选参数，Form 类型，用于参数验证的表单类，默认为 BaseForm
fn: 可选参数，被装饰的函数

返回值:

返回装饰后的函数

示例:

from simplejrpc import ServerApplication, BaseForm
from typing import Optional

app = ServerApplication("/tmp/rpc.sock")

# 基本路由注册
@app.route()
def get_user_info(user_id: int):
    return {"user_id": user_id, "name": "John Doe"}

# 自定义方法名
@app.route("user.create")
def create_user(name: str, email: str):
    return {"id": 123, "name": name, "email": email}

# 使用表单验证
class UserForm(BaseForm):
    name: str
    email: str
    age: Optional[int] = None

@app.route("user.update", form=UserForm)
def update_user(data: UserForm):
    return {"success": True, "data": data.dict()}

`load_config(config_path: str)`

从指定路径加载配置文件。支持 YAML、JSON 和 INI 格式的配置文件。

参数:

config_path: 字符串类型，配置文件路径

返回值:

返回字典类型的配置内容

异常:

如果文件不存在，抛出 FileNotFoundError 异常
如果文件格式不支持，抛出 ValueError 异常

支持的配置文件格式:

YAML (.yml, .yaml)
INI (.ini)
JSON (.json)

示例:

app = ServerApplication("/tmp/rpc.sock")

# 加载 YAML 配置文件
app.load_config("config.yaml")

# 加载 JSON 配置文件
app.load_config("config.json")

# 配置文件示例 (config.yaml)
# debug: true
# log_level: "INFO"
# max_connections: 100
# timeout: 30

`setup_logger(config_content: Dict[str, Any])`

根据配置内容设置日志记录器。

参数:

config_content: 字典类型，配置内容

返回值: 无

注意:

配置中必须包含 "logger" 部分，且 "sink" 参数是必需的

`clear_socket()`

清理 Unix 域套接字。

参数: 无

返回值: 无

`middleware(middleware_instance: RPCMiddleware)`

注册中间件。中间件按添加顺序执行，支持请求预处理和响应后处理。

参数:

middleware_instance: RPCMiddleware 实例，中间件对象

返回值:

返回服务器实例

示例:

from simplejrpc import ServerApplication
import time

app = ServerApplication("/tmp/rpc.sock")

# 简单的日志中间件
def logging_middleware(request, call_next):
    start_time = time.time()
    print(f"请求开始: {request.method}")
    
    response = call_next(request)
    
    duration = time.time() - start_time
    print(f"请求完成: {request.method}, 耗时: {duration:.3f}s")
    return response

# 认证中间件
def auth_middleware(request, call_next):
    # 检查认证信息
    if not request.headers.get("Authorization"):
        raise RPCException("未授权访问", code=-32001)
    return call_next(request)

# 添加中间件
app.middleware(logging_middleware)
app.middleware(auth_middleware)

`run()`

启动 RPC 服务器，并注册退出时清理套接字的回调函数。该方法会阻塞当前线程，开始监听指定的 Unix 域套接字。

参数: 无

返回值: 无

示例:

from simplejrpc import ServerApplication

app = ServerApplication("/tmp/rpc.sock")

@app.route("ping")
def ping():
    return "pong"

# 启动服务器（阻塞运行）
print("启动 RPC 服务器...")
app.run()

注意

run() 方法会阻塞当前线程
确保 socket 路径有适当的权限
服务器启动前会自动清理已存在的 socket 文件

Settings 类

Settings 类是一个配置管理类，用于处理和访问应用程序的配置信息。该类提供了简单的接口来获取配置项和配置部分。

配置特性

多源配置：支持字典、文件、环境变量
类型验证：自动类型检查和转换
默认值：提供合理的默认配置
环境变量：支持环境变量覆盖
配置合并：多个配置源智能合并

配置优先级

配置加载的优先级（从高到低）：

环境变量
配置文件
代码中的配置字典
默认值

`init(config_content: Dict[str, Any] = None)`

初始化 Settings 实例。

参数:

config_content: 可选参数，字典类型，配置内容。如果为 None，则使用空字典。

返回值: 无

示例:

from simplejrpc import Settings

# 使用默认配置
settings = Settings()

# 使用自定义配置
config = {
    "debug": True,
    "log_level": "DEBUG",
    "max_connections": 50,
    "timeout": 60
}
settings = Settings(config)

# 访问配置
print(f"调试模式: {settings.debug}")
print(f"日志级别: {settings.log_level}")

`get_section(key: str, default=None) -> Dict[str, Any]`

获取指定键名的配置部分。支持点号分隔的嵌套键名。

参数:

key: 字符串类型，配置部分的键名（支持嵌套，如 "server.database"）
default: 可选参数，如果指定键名不存在时返回的默认值

返回值:

返回字典类型的配置部分，如果键名不存在则返回默认值

示例:

from simplejrpc import Settings

config = {
    "server": {
        "host": "localhost",
        "port": 8080,
        "database": {
            "url": "sqlite:///app.db",
            "pool_size": 10
        }
    },
    "logging": {
        "level": "INFO",
        "format": "%(asctime)s - %(levelname)s - %(message)s"
    }
}
settings = Settings(config)

# 获取服务器配置部分
server_config = settings.get_section("server", {})
print(f"服务器配置: {server_config}")

# 获取数据库配置部分
db_config = settings.get_section("server.database", {})
print(f"数据库配置: {db_config}")

# 获取不存在的配置部分（返回默认值）
cache_config = settings.get_section("cache", {"enabled": False})
print(f"缓存配置: {cache_config}")

`get_config_object()`

获取整个配置对象。

参数: 无

返回值:

返回 DefaultMapping 类型的配置对象

`get_option(key: str) -> DefaultMapping`

获取指定键名的配置选项。

参数:

key: 字符串类型，配置选项的键名

返回值:

返回 DefaultMapping 类型的配置选项

示例:

from simplejrpc import Settings

config = {
    "server": {
        "host": "localhost",
        "port": 8080
    }
}
settings = Settings(config)

# 获取服务器配置选项
server_option = settings.get_option("server")
print(f"主机: {server_option.host}")
print(f"端口: {server_option.port}")

# 通过属性方式访问
print(f"服务器地址: {server_option.host}:{server_option.port}")

注意事项

配置内容会被转换为 DefaultMapping 对象，支持通过属性方式访问
当访问不存在的配置项时，会引发属性错误
可以通过 get_section 方法提供默认值来避免访问不存在的配置部分时出错

GConfig 类

GConfig 是一个全局配置管理类，提供了统一的配置访问接口。该类采用单例模式，确保整个应用程序中配置的一致性。

全局配置特性

单例模式：全局唯一的配置实例
统一接口：提供一致的配置访问方式
自动加载：支持多种配置源自动加载
线程安全：支持多线程环境
动态更新：运行时配置更新支持

使用场景

应用程序全局配置管理
跨模块配置共享
配置的集中式管理
环境相关的配置切换

`init(config_content: Optional[Dict[str, Any]] = None, config_path: Optional[str] = None)`

初始化 GConfig 实例。由于采用单例模式，多次初始化会返回同一个实例。

参数:

config_content: 可选参数，字典类型，配置内容
config_path: 可选参数，字符串类型，配置文件路径

返回值: 无

示例:

from simplejrpc import GConfig

# 方式1：使用配置字典初始化
config_data = {
    "app": {
        "name": "MyApp",
        "version": "1.0.0",
        "debug": True
    },
    "database": {
        "host": "localhost",
        "port": 5432,
        "name": "mydb"
    }
}
config = GConfig(config_content=config_data)

# 方式2：使用配置文件初始化
config = GConfig(config_path="config.yaml")

# 方式3：组合使用（配置文件 + 额外配置）
config = GConfig(
    config_path="config.yaml",
    config_content={"debug": True}
)

# 单例验证
config1 = GConfig()
config2 = GConfig()
print(config1 is config2)  # True

`get_instance()`

获取 GConfig 的单例实例。如果实例不存在，会创建一个新的实例。

参数: 无

返回值:

返回 GConfig 实例

示例:

from simplejrpc import GConfig

# 获取全局配置实例
config = GConfig.get_instance()

# 在不同模块中获取同一个实例
# module_a.py
def get_app_name():
    config = GConfig.get_instance()
    return config.get("app.name", "Unknown")

# module_b.py
def get_db_config():
    config = GConfig.get_instance()
    return config.get_section("database", {})

# 两个模块访问的是同一个配置实例
print(get_app_name())  # MyApp
print(get_db_config())  # {"host": "localhost", "port": 5432, ...}

`get(key: str, default: Any = None)`

获取配置值。支持点号分隔的嵌套键名访问。

参数:

key: 字符串类型，配置键名（支持嵌套，如 "app.database.host"）
default: 任意类型，默认值

返回值:

返回配置值或默认值

示例:

from simplejrpc import GConfig

# 初始化配置
config_data = {
    "app": {
        "name": "MyRPCApp",
        "version": "2.1.0",
        "features": {
            "auth": True,
            "cache": False
        }
    },
    "server": {
        "host": "0.0.0.0",
        "port": 8080,
        "workers": 4
    }
}
config = GConfig(config_content=config_data)

# 获取简单配置
app_name = config.get("app.name", "DefaultApp")
print(f"应用名称: {app_name}")

# 获取嵌套配置
auth_enabled = config.get("app.features.auth", False)
print(f"认证功能: {auth_enabled}")

# 获取服务器配置
server_host = config.get("server.host", "localhost")
server_port = config.get("server.port", 3000)
print(f"服务器: {server_host}:{server_port}")

# 获取不存在的配置（使用默认值）
timeout = config.get("server.timeout", 30)
print(f"超时时间: {timeout}秒")

`get_section(key: str, default: Dict[str, Any] = None)`

获取配置节。返回指定键名对应的整个配置节。

参数:

key: 字符串类型，配置节键名
default: 字典类型，默认值

返回值:

返回配置节字典或默认值

示例:

from simplejrpc import GConfig

config = GConfig.get_instance()

# 获取整个数据库配置节
db_config = config.get_section("database", {})
print(f"数据库配置: {db_config}")

# 获取应用配置节
app_config = config.get_section("app", {})
print(f"应用配置: {app_config}")

`set(key: str, value: Any)`

设置配置值。支持运行时动态更新配置。

参数:

key: 字符串类型，配置键名
value: 任意类型，配置值

返回值: 无

示例:

from simplejrpc import GConfig

config = GConfig.get_instance()

# 设置简单配置
config.set("app.debug", True)
config.set("server.port", 9000)

# 设置嵌套配置
config.set("app.features.logging", True)

# 验证设置
print(config.get("app.debug"))  # True
print(config.get("server.port"))  # 9000

注意事项

GConfig 采用单例模式，确保全局配置的一致性
支持多种配置源：字典、文件、环境变量
配置键名支持点号分隔的嵌套访问
线程安全，可在多线程环境中使用
支持运行时动态配置更新

JsonRpcServer 类

JsonRpcServer 是一个基于 Unix 域套接字的 JSON-RPC 服务器实现类，用于处理 RPC 请求和响应。该类提供了中间件支持、请求处理、响应序列化和错误处理等功能。

`init(socket_path: str)`

初始化 JsonRpcServer 实例。

参数:

socket_path: 字符串类型，Unix 域套接字路径

返回值: 无

`to_serializable(response: ResponseType)`

将响应对象转换为可序列化的对象。

参数:

response: ResponseType 类型，响应对象

返回值:

返回可序列化的响应对象

`middleware(middleware_instance: RPCMiddleware)`

注册中间件。

参数:

middleware_instance: RPCMiddleware 实例，中间件对象

返回值:

返回中间件实例

`dispatch_to_response(*args, serializer=json.dumps, post_process=None, **kwargs)`

分发请求并生成响应。

参数:

*args: 可变参数，请求数据
serializer: 可选参数，函数类型，序列化函数，默认为 json.dumps
post_process: 可选参数，函数类型，后处理函数
**kwargs: 关键字参数，其他参数

返回值:

返回序列化后的响应字符串

`handle_client(reader, writer)`

处理客户端连接。

参数:

reader: StreamReader 对象，用于读取客户端数据
writer: StreamWriter 对象，用于向客户端写入数据

返回值: 无

`clear_socket()`

清理 Unix 域套接字。

参数: 无

返回值: 无

`start_server(f_socket)`

启动服务器。

参数:

f_socket: 字符串类型，套接字路径

返回值: 无

异常:

如果套接字路径未配置，抛出 RuntimeError 异常

`run()`

运行服务器。

参数: 无

返回值: 无

注意事项

服务器使用 Unix 域套接字进行通信
支持中间件机制，可以处理请求和响应
提供错误处理和日志记录功能
自动清理套接字文件

ClientTransport 类

ClientTransport 是一个抽象基类，用于定义 RPC 客户端传输层的接口。该类定义了客户端与服务器通信所需的基本方法。

`send_message(message: Union[Dict[str, Any], str])`

发送消息到服务器。

参数:

message: 字典或字符串类型，要发送的消息内容
- 当为字典类型时，表示发送 JSON-RPC 请求
- 当为字符串类型时，表示发送原始字符串消息

返回值: 无

`close()`

关闭客户端连接。

参数: 无

返回值: 无

`connect()`

建立与服务器的连接。

参数: 无

返回值: 无

注意事项

这是一个抽象基类，所有方法都是异步的（使用 async 关键字）
继承此类的具体实现必须实现所有抽象方法
所有方法都使用了 @abstractmethod 装饰器，表示这些方法必须在子类中被重写

UnixSocketTransport 类

UnixSocketTransport 是一个基于 Unix 域套接字的传输层实现类，继承自 ClientTransport 抽象基类。该类提供了与 RPC
服务器进行通信的具体实现。

`init(socket_path: str)`

初始化 UnixSocketTransport 实例。

参数:

socket_path: 字符串类型，Unix 域套接字路径

返回值: 无

`connect()`

建立与服务器的异步连接。

参数: 无

返回值: 无

`make_header(message: str)`

构造消息头部。

参数:

message: 字符串类型，要发送的消息内容

返回值: 无

`make_body(body: str)`

构造消息主体。

参数:

body: 字符串类型，消息主体内容

返回值: 无

`send_message(message)`

发送消息并接收响应。

参数:

message: 字典或字符串类型，要发送的消息
- 当为字典类型时，会自动转换为 JSON 字符串

返回值:

返回服务器的响应（JSON 解析后的对象）

`read_from_header()`

读取消息头部。

参数: 无

返回值:

返回字节类型的头部内容，直到遇到 \r\n\r\n

`read_content_length_from_header(header: str)`

从头部解析内容长度。

参数:

header: 字符串类型，消息头部

返回值:

返回整数类型的内容长度

`_read_response()`

读取完整的响应消息。

参数: 无

返回值:

返回 JSON 解析后的响应对象

`close()`

关闭连接。

参数: 无

返回值: 无

注意事项

所有网络操作都是异步的，需要在异步环境中使用
消息格式遵循 Content-Length 头部协议
支持自动的 JSON 序列化和反序列化
使用完毕后需要手动调用 close() 方法关闭连接

Request 类

Request 类是一个 RPC 客户端请求类，用于发送 JSON-RPC 请求并处理响应。该类支持同步和异步操作方式。

`init(socket_path: str, adapter: Optional[ClientTransport] = None)`

初始化 Request 实例。

参数:

socket_path: 字符串类型，Unix 域套接字路径
adapter: 可选参数，ClientTransport 实例，默认为 None，会创建一个新的 UnixSocketTransport 实例

返回值: 无

`make_session()`

创建与服务器的连接会话。

参数: 无

返回值: 无

`send_request(method: str, params: Union[Dict[str, Any], Tuple[Any, ...], None] = None, id: Any = NOID)`

同步发送 RPC 请求（内部使用 asyncio.run 执行异步请求）。

参数:

method: 字符串类型，RPC 方法名
params: 可选参数，字典或元组类型，RPC 方法参数
id: 可选参数，请求 ID，默认为 NOID

返回值:

返回服务器的响应

注意事项

类支持同步和异步两种调用方式
每次请求都会建立新的连接并在请求完成后关闭
使用 async_send_request 需要在异步环境中调用
使用 send_request 可以在同步环境中调用
可以通过自定义 adapter 来实现不同的传输方式

GmRequest 类

GmRequest 是一个继承自 Request 的专用请求类，用于处理特定的 GM（Game Master）请求。该类使用默认的 GM 套接字路径进行初始化。

`init(socket_path=DEFAULT_GA_SOCKET, adapter=None)`

初始化 GmRequest 实例。

参数:

socket_path: 可选参数，字符串类型，Unix 域套接字路径，默认使用 DEFAULT_GA_SOCKET
adapter: 可选参数，ClientTransport 实例，默认为 None

返回值: 无

注意事项

该类继承了 Request 类的所有功能
使用默认的 GM 套接字路径（DEFAULT_GA_SOCKET）
可以通过自定义 adapter 来实现不同的传输方式
主要用于处理 GM 相关的 RPC 请求

RPCException 类

RPCException 是一个基础的 RPC 异常类，继承自 Python 内置的 Exception 类。该类用于处理 RPC 调用过程中的各种异常情况，并提供统一的错误响应格式。

__init__(message=http.HTTPStatus.BAD_REQUEST.description, code=http.HTTPStatus.BAD_REQUEST.value, data: Optional[str] = None, *args, **kwargs)

初始化 RPCException 实例。

参数:

message: 字符串类型，错误消息，默认为 HTTP 400 Bad Request 的描述
code: 整数类型，错误代码，默认为 HTTP 400 Bad Request 的状态码
data: 可选参数，字符串类型，额外的错误数据
*args: 可变位置参数，传递给父类 Exception
**kwargs: 可变关键字参数，额外的参数

返回值: 无

`str()`

将异常转换为字符串表示。

参数: 无

返回值:

返回 JSON 格式的字符串，包含以下字段：
- code: 错误代码
- data: 错误数据（如果有）
- msg: 错误消息

注意事项

该类作为基础异常类，其他特定类型的 RPC 异常应该继承自此类
错误响应会被自动序列化为 JSON 格式
默认使用 HTTP 400 Bad Request 的状态码和描述
支持自定义错误数据的传递

RPC 异常类文档

以下是 RPC 相关的异常类，所有异常类都继承自基础的 RPCException 类。

UnauthorizedError

未授权异常，用于处理认证和授权失败的情况。

继承: RPCException

ValidationError

验证异常，用于处理参数验证失败的情况。

继承: RPCException

FileNotFoundError

文件未找到异常，用于处理文件操作相关的错误。

继承: RPCException

ValueError

值错误异常，用于处理无效的参数值。

继承: RPCException

RuntimeError

运行时错误异常，用于处理程序运行时的一般性错误。

继承: RPCException

AttributeError

属性错误异常，用于处理访问不存在的属性时的错误。

继承: RPCException

注意事项

所有异常类都继承自 RPCException，因此都具有相同的基础功能
每个异常类都可以接受自定义的错误消息、错误代码和额外数据
异常信息会被自动序列化为 JSON 格式
默认使用 HTTP 400 Bad Request 的状态码和描述
这些异常类提供了细粒度的错误处理能力，可以根据具体的错误情况选择合适的异常类

Language 类

Language 是一个字符串枚举类，用于定义支持的语言代码。该类继承自 str 和 enum.Enum，提供了标准化的语言代码表示。

枚举值

EN: 英语 ("en")
JA: 日语 ("ja")
RU: 俄语 ("ru")
ZH_CN: 简体中文 ("zh-CN")
ZH_TW: 繁体中文 ("zh-TW")

`values()`

获取所有支持的语言代码值。

参数: 无

返回值:

返回列表类型，包含所有语言代码的字符串值

注意事项

语言代码遵循标准的语言标识格式
支持五种主要语言：英语、日语、俄语、简体中文和繁体中文
可以通过枚举成员直接访问语言代码
通过 values() 方法可以获取所有支持的语言代码列表

GI18n 类

GI18n 是一个国际化（i18n）管理类，提供了便捷的多语言翻译和资源加载功能。该类封装了 T 类的功能，提供了更简洁的接口。

`init(path: str, lang: str, adapter: TranslateBase = None)`

初始化 GI18n 实例。

参数:

path: 字符串类型，国际化配置文件位置
lang: 字符串类型，国际化语言配置
adapter: 可选参数，TranslateBase 类型，自定义的国际化翻译器

返回值: 无

`translate(key) -> str`

获取指定键的翻译。

参数:

key: 翻译键名

返回值:

返回字符串类型的翻译内容

`get_lang() -> str`

获取当前语言设置。

参数: 无

返回值:

返回字符串类型的当前语言代码

`set_path(path: str)`

设置国际化配置文件路径。

参数:

path: 字符串类型，配置文件路径

返回值: 无

`set_lang(lang: str)`

设置当前语言。

参数:

lang: 字符串类型，语言代码

返回值: 无

`translate_ctx(key, *value) -> str`

获取带有上下文参数的翻译。

参数:

key: 翻译键名
*value: 可变参数，用于格式化翻译内容

返回值:

返回字符串类型的格式化翻译内容

`translate_load(file: Union[str, Path], ignore=False) -> str`

从文件加载翻译。

参数:

file: 字符串或 Path 类型，翻译文件路径
ignore: 布尔类型，是否忽略文件不存在的错误，默认为 False

返回值:

返回字符串类型的翻译内容

`translate_load_partial(file: Path, ignore=False) -> str`

从文件加载部分翻译。

参数:

file: Path 类型，翻译文件路径
ignore: 布尔类型，是否忽略文件不存在的错误，默认为 False

返回值:

返回字符串类型的翻译内容

`translate_app_info(file_root: Path, filename: str) -> str`

加载应用信息的翻译。

参数:

file_root: Path 类型，翻译文件根目录
filename: 字符串类型，翻译文件名

返回值:

返回字符串类型的翻译内容

注意事项

该类是对 T 类的封装，提供了更简洁的接口
支持自定义翻译适配器
支持从文件加载翻译内容
提供了带格式化参数的翻译功能
可以动态切换语言和配置文件路径

T 类

T 是一个国际化（i18n）核心类，提供了文本翻译的基础功能，支持多语言源和缓存机制。

类属性

I18n_ROOT: Path 类型，国际化文件根目录
LANG: Language 类型，当前语言设置，默认为英语（EN）
Translate: DefaultTranslate 类型，翻译器实现类

翻译相关方法

`translate(key: str)`

获取指定键的翻译。

参数:

key: 字符串类型，翻译键名

返回值:

返回翻译后的文本

`translate_ctx(key, *value)`

获取带格式化参数的翻译。

参数:

key: 翻译键名
*value: 可变参数，用于格式化翻译文本

返回值:

返回格式化后的翻译文本

部分翻译方法

`translate_partial(key: str)`

创建一个部分应用的翻译函数。

参数:

key: 字符串类型，翻译键名

返回值:

返回一个偏函数，用于后续翻译

`translate_ctx_partial(key: str, value: List[Any])`

创建一个带格式化参数的部分应用翻译函数。

参数:

key: 字符串类型，翻译键名
value: 列表类型，格式化参数列表

返回值:

返回一个偏函数，用于后续带参数翻译

文件加载相关方法

`translate_load(file: Path, ignore=False)`

从文件加载翻译。

参数:

file: Path 类型，翻译文件路径
ignore: 布尔类型，是否忽略文件不存在错误，默认为 False

返回值:

返回加载的翻译内容

`translate_app_info(file_root: Path, filename: str)`

加载应用信息的翻译。

参数:

file_root: Path 类型，国际化目标目录
filename: 字符串类型，需要国际化的文件名

返回值:

返回加载的应用信息翻译

配置相关方法

`set_lang(lang: Language = Language.ZH_CN)`

设置默认语言。

参数:

lang: Language 类型，语言设置，默认为简体中文

`get_lang()`

获取当前语言设置。

返回值:

返回当前语言设置

`set_path(path: str)`

设置国际化文件根目录。

参数:

注意事项

该类是国际化功能的核心实现
支持多种翻译方式：直接翻译、带参数翻译、部分应用翻译
提供文件级别的翻译加载功能
支持动态切换语言和配置路径
使用了类方法（@classmethod）实现，方便静态调用
提供了缓存机制以提高性能

ConfigParser 类

ConfigParser 是一个抽象基类，用于定义配置文件解析器的基本接口。该类提供了读取和写入配置文件的标准方法定义。

`init(config_path: str)`

初始化配置解析器实例。

参数:

config_path: 字符串类型，配置文件的路径

返回值: 无

`read()`

读取配置文件的抽象方法。

参数: 无

返回值:

具体实现类决定返回值类型

注意:

这是一个抽象方法，必须由子类实现

`write()`

写入配置文件的抽象方法。

参数: 无

返回值:

具体实现类决定返回值类型

注意:

这是一个抽象方法，必须由子类实现

注意事项

这是一个抽象基类，不能直接实例化
所有子类必须实现 read() 和 write() 方法
该类为不同类型的配置文件解析器提供了统一的接口
子类可以根据具体的配置文件格式（如 JSON、YAML、INI 等）实现相应的解析逻辑

JsonConfigParser 类

JsonConfigParser 是一个继承自 ConfigParser 的 JSON 配置文件解析器类，用于读取和写入 JSON 格式的配置文件。

`read() -> Any`

读取 JSON 配置文件的内容。

参数: 无

返回值:

返回解析后的 JSON 内容（任意类型）

实现细节:

使用 UTF-8 编码打开文件
使用 json.load() 解析 JSON 内容
自动处理文件的打开和关闭

`write(content: Dict[str, Any]) -> NoReturn`

将内容写入 JSON 配置文件。

参数:

content: 字典类型，要写入的配置内容

返回值:

无返回值（NoReturn）

实现细节:

使用默认编码打开文件进行写入
使用 json.dump() 将内容序列化为 JSON
自动处理文件的打开和关闭

注意事项

读取时使用 UTF-8 编码，确保正确处理中文等特殊字符
写入的内容必须是可 JSON 序列化的字典类型
文件操作使用 with 语句，确保资源正确释放
继承自 ConfigParser 基类，实现了标准的配置文件接口

YamlConfigParser 类

YamlConfigParser 是一个继承自 ConfigParser 的 YAML 配置文件解析器类，用于读取和写入 YAML 格式的配置文件。

`read() -> Any`

读取 YAML 配置文件的内容。

参数: 无

返回值:

返回解析后的 YAML 内容（任意类型）

实现细节:

使用 UTF-8 编码打开文件
使用 yaml.safe_load() 安全地解析 YAML 内容
自动处理文件的打开和关闭

`write(content: Dict[str, Any]) -> NoReturn`

将内容写入 YAML 配置文件。

参数:

content: 字典类型，要写入的配置内容

返回值:

无返回值（NoReturn）

实现细节:

使用默认编码打开文件进行写入
使用 yaml.dump() 将内容序列化为 YAML
自动处理文件的打开和关闭

注意事项

读取时使用 UTF-8 编码，确保正确处理中文等特殊字符
使用 safe_load() 而不是 load()，提供更安全的 YAML 解析
写入的内容必须是可 YAML 序列化的字典类型
文件操作使用 with 语句，确保资源正确释放
继承自 ConfigParser 基类，实现了标准的配置文件接口

IniConfigParser 类

IniConfigParser 是一个继承自 ConfigParser 的 INI 配置文件解析器类，用于读取和写入 INI 格式的配置文件。

`read()`

读取 INI 配置文件的内容。

参数: 无

返回值:

返回配置文件的所有节（sections）内容

实现细节:

使用 Python 标准库的 configparser.ConfigParser() 创建解析器
读取指定路径的 INI 文件
返回 _sections 属性，包含所有配置节的内容

`write(content: str)`

将内容写入 INI 配置文件。

参数:

content: 字符串类型，要写入的配置内容

返回值: 无

实现细节:

以写入模式打开文件
直接写入字符串内容
自动处理文件的打开和关闭

注意事项

读取方法返回的是配置文件中所有节的字典表示
写入方法接受原始字符串内容，需要确保内容符合 INI 格式
文件操作使用 with 语句，确保资源正确释放
继承自 ConfigParser 基类，实现了标准的配置文件接口

StringLangValidator

StringLangValidator 是一个语言验证器类，继承自 BaseValidator，用于验证和处理表单中的语言字段。

`init(lang="en")`

初始化语言验证器实例。

参数:

lang: 字符串类型，默认语言代码，默认值为 "en"（英语）

返回值: 无

`validator(form, field)`

验证表单字段中的语言值。

参数:

form: Form 对象，WTForms 表单实例
field: Field 对象，表单字段实例

返回值:

返回验证后的语言代码

实现细节:

获取字段值，如果为空则使用默认语言
使用获取的语言值初始化 GI18n 实例
返回最终的语言代码

注意事项

该验证器主要用于处理国际化相关的表单字段
使用 setting.PROJECT_I18n_PATH 作为国际化配置文件路径
如果字段值为空，会使用构造函数中设置的默认语言
验证过程中会创建 GI18n 实例，确保语言设置生效

StrRangeValidator

StrRangeValidator 是一个字符串范围验证器类，继承自 BaseValidator，用于验证字符串字段的值是否在允许的范围内。

类属性

allows: 列表类型，存储允许的值列表，默认为空列表
err_message: 字符串类型，存储错误消息模板，默认为空字符串

`init(allows, message=None)`

初始化验证器实例。

参数:

allows: 列表类型，允许的值列表
message: 可选参数，字符串类型，自定义错误消息，默认为 None

返回值: 无

`validator(form, field)`

验证表单字段值是否在允许的范围内。

参数:

form: Form 对象，WTForms 表单实例
field: Field 对象，表单字段实例

返回值: 无

异常:

如果字段值不在允许的范围内，抛出 validators.ValidationError 异常
- 如果设置了 err_message，使用国际化翻译后的消息
- 否则使用默认错误消息格式 expected value {allows}

注意事项

验证器支持国际化错误消息
错误消息可以通过构造函数自定义
验证失败时会抛出 ValidationError 异常
适用于需要限制字段值在特定范围内的场景

IntLimitValidator

IntLimitValidator 是一个整数范围验证器类，用于验证表单中整数字段的值是否在指定的范围内。

类属性

min: 整数类型，最小允许值
max: 整数类型，最大允许值
err_message: 字符串类型，错误消息模板，默认为空字符串

`init(min=1, max=1000)`

初始化验证器实例。

参数:

min: 整数类型，最小允许值，默认为 1
max: 整数类型，最大允许值，默认为 1000

返回值: 无

`validator(form, field)`

验证表单字段值是否在允许的范围内。

参数:

form: Form 对象，WTForms 表单实例
field: Field 对象，表单字段实例

返回值: 无

异常:

如果字段值小于最小值或大于最大值，抛出 validators.ValidationError 异常
- 如果设置了 err_message，使用国际化翻译后的消息，并传入最小值和最大值作为参数
- 否则使用默认错误消息格式 expected value [min,max]

注意事项

验证器支持国际化错误消息
错误消息可以包含最小值和最大值的占位符
验证失败时会抛出 ValidationError 异常
适用于需要限制整数字段值在特定范围内的场景
默认范围为 1-1000，可以通过构造函数自定义

BaseForm 类

BaseForm 是一个基础表单类，继承自 WTForms 的 Form 类，提供了基本的语言设置功能。

类属性

lang: StringField 类型，语言字段
- 使用 StringLangValidator 验证器
- 用于设置和验证表单的语言选项

特点

所有继承自 BaseForm 的表单类都会自动包含语言字段
语言字段使用 StringLangValidator 进行验证，确保语言设置的有效性
默认支持国际化功能

注意事项

该类作为基础表单类，应该被其他具体的表单类继承
语言验证器默认使用英语（"en"）作为默认语言
通过继承此类可以确保所有表单都具有一致的语言处理机制

以下是你提供的类和方法的文档总结：

类：`nan`

用于表示“非数字”（NaN）值的占位符类。

方法：
- __str__(self)：返回类的名称作为字符串。
- __repr__(self)：返回类名称的表示形式。

BaseField 类

表单字段的基类，提供字段验证的基本功能。

构造函数：
- __init__(self, validators=None, err_msg="", default=None, valid="", label="")：初始化字段，支持传入可选的验证器、错误信息、默认值和验证选项。
方法：
- validator(self, value)：抽象方法，子类需要实现，用于对字段值进行验证。
- get_validators(self)：返回与字段关联的验证器列表。
- required(self)：检查字段是否有 required 验证器。
- raise_except(self, err_msg, except_type=None)：抛出指定类型和消息的异常。

Field 类

一个描述符类，用于验证和清理字段的值。继承自 BaseField

构造函数：
- __init__(self, validators=None, err_msg="", default=None, valid="", label="")：初始化字段，支持传入验证器、错误信息、默认值和验证选项。
方法：
- _get_value(self, value)：返回字段值，若值为 None 则返回默认值。
- __set__(self, instance, value)：设置字段的值，经过验证和清理后赋值。
- validator(self, value)：验证字段值（在基类中是空操作，子类可实现具体验证）。
- clean_data(self, instance, value)：通过应用验证器来清理字段值。
- __delete__(self, instance)：从实例中删除字段值。

StringField 类

用于处理字符串类型值的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 str。

StringIdentifierField 类

处理特定于标识符的字符串字段，包含长度限制和正则表达式验证。继承自 Field

构造函数：
- __init__(self, validators=None, err_msg="", max_len=25, min_len=1, valid="", label="")：初始化字段，支持验证器、错误信息和长度限制。
方法：
- validator(self, value)：使用正则表达式和长度检查来验证值。

StringRegexField 类

处理字符串类型值并使用正则表达式进行验证。继承自 StringField

构造函数：
- __init__(self, validators=None, err_msg="", regex=None, default=None, valid="", label="")：初始化字段，支持正则表达式和验证选项。
方法：
- validator(self, value)：使用提供的正则表达式来验证字段值。

BooleanField 类

处理布尔类型值的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 bool。

IntegerField 类

处理整数类型值的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 int。

ListField 类

处理列表类型值的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 list。

DictField 类

处理字典类型值的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 dict。

NoValueField 类

处理 "NoValue"（用于占位或表示值的缺失）的字段。继承自 Field

构造函数：
- 继承自 Field，并将 expected_type 设置为 NoValue。

RangeField 类

一个字段类，限定值必须在一个允许的范围内。继承自 Field

构造函数：
- __init__(self, validators=None, err_msg="", allow=None, default=None, valid="", label="")：初始化字段，支持设置允许的值范围。
方法：
- validator(self, value)：验证值是否在允许的范围内。

StrRangeField 类

一个字符串类型字段，值必须在指定的范围内。继承自 RangeField 和 StringField

IntRangeField 类

一个整数类型字段，值必须在指定的范围内。继承自 RangeField 和 IntegerField

PortField 类

一个处理端口值的字段，值必须在 1 到 65535 的范围内。继承自 Field

构造函数：
- __init__(self, validators=None, min_value=1, max_value=65535, err_msg="", default=None, valid="", label="")
  ：初始化字段，支持端口范围的设置。
方法：
- validator(self, value)：验证端口值是否合法。

LengthLimitField 类

处理字符串类型值，并限制其长度在指定范围内。继承自 StringField

构造函数：
- __init__(self, validators=None, min_length=0, max_length=32, err_msg="", default=None, valid="", label="")
  ：初始化字段，支持设置字符串的最小和最大长度。
方法：
- validator(self, value)：验证值的长度是否在指定范围内。

StringRegexLimitField 类

结合正则表达式和长度限制的字段，处理字符串值。继承自 StringField

构造函数：
__init__(self, validators=None, min_length=0, max_length=32, err_msg="", default=None, regex=None, valid="", label="")
：初始化字段，支持正则表达式、最小和最大长度的设置。
方法：
- validator(self, value)：验证值的长度和正则表达式。

IntegerLimitField 类

处理整数值，并限制其在指定的最小和最大范围内。继承自 IntegerField

构造函数：
- __init__(self, validators=None, min_value=0, max_value=32, err_msg="", default=None, valid="", label="")
  ：初始化字段，支持设置最小和最大值。
方法：
- validator(self, value)：验证整数值是否在指定范围内。

StringValueLimitField 类

处理字符串值，并限制其在指定范围内。继承自 StringField

构造函数：
- __init__(self, validators=None, min_value=nan, max_value=nan, err_msg="", default=None, valid="", label="")
  ：初始化字段，支持设置最小和最大值。
方法：
- validator(self, value)：验证字符串值是否在指定范围内。

StringLengthLimitField 类

结合字符串长度限制的字段。继承自 LengthLimitField

StringRangField 类

结合字符串范围限制的字段。继承自 StrRangeField

IntegerRangeField 类

结合整数范围限制的字段。继承自 IntRangeField

GLogger 类

GLogger 是一个全局日志管理类，基于 loguru 库实现，提供了统一的日志记录接口。该类采用单例模式，确保整个应用程序中日志配置的一致性。

日志特性

单例模式：全局唯一的日志实例
多级别日志：支持 DEBUG、INFO、WARNING、ERROR、CRITICAL
多输出目标：控制台、文件、远程等
格式化输出：支持自定义日志格式
异步写入：高性能异步日志写入
自动轮转：支持按时间或大小轮转日志文件

日志级别说明

级别	数值	用途
DEBUG	10	调试信息，详细的程序执行信息
INFO	20	一般信息，程序正常运行的信息
WARNING	30	警告信息，可能出现问题但不影响运行
ERROR	40	错误信息，程序出现错误但可以继续运行
CRITICAL	50	严重错误，程序可能无法继续运行

`init(config: Optional[Dict[str, Any]] = None)`

初始化 GLogger 实例。由于采用单例模式，多次初始化会返回同一个实例。

参数:

config: 可选参数，字典类型，日志配置信息

返回值: 无

示例:

from simplejrpc import GLogger

# 使用默认配置
logger = GLogger()

# 使用自定义配置
log_config = {
    "level": "INFO",
    "format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{function}:{line} - {message}",
    "handlers": [
        {
            "sink": "logs/app.log",
            "rotation": "1 day",
            "retention": "7 days",
            "compression": "zip"
        },
        {
            "sink": "sys.stdout",
            "colorize": True
        }
    ]
}
logger = GLogger(config=log_config)

# 单例验证
logger1 = GLogger()
logger2 = GLogger()
print(logger1 is logger2)  # True

`get_instance()`

获取 GLogger 的单例实例。如果实例不存在，会创建一个新的实例。

参数: 无

返回值:

返回 GLogger 实例

示例:

from simplejrpc import GLogger

# 在任何模块中获取日志实例
logger = GLogger.get_instance()

# 在不同模块中使用同一个日志实例
# module_a.py
def process_data():
    logger = GLogger.get_instance()
    logger.info("开始处理数据")
    try:
        # 处理逻辑
        logger.debug("数据处理中...")
        result = "处理完成"
        logger.info(f"数据处理成功: {result}")
        return result
    except Exception as e:
        logger.error(f"数据处理失败: {e}")
        raise

# module_b.py
def save_result(data):
    logger = GLogger.get_instance()
    logger.info(f"保存结果: {data}")
    # 保存逻辑
    logger.info("结果保存成功")

`debug(message: str)`

记录调试级别的日志。用于详细的程序执行信息，通常在开发和调试阶段使用。

参数:

message: 字符串类型，日志消息

返回值: 无

示例:

from simplejrpc import GLogger

logger = GLogger.get_instance()

# 记录调试信息
logger.debug("进入函数 calculate_total")
logger.debug(f"参数值: price={100}, quantity={5}")

def calculate_total(price, quantity):
    logger.debug(f"开始计算总价: price={price}, quantity={quantity}")
    total = price * quantity
    logger.debug(f"计算结果: {total}")
    return total

result = calculate_total(100, 5)
logger.debug(f"函数返回值: {result}")

`info(message: str)`

记录信息级别的日志。用于记录程序正常运行的重要信息。

参数:

message: 字符串类型，日志消息

返回值: 无

示例:

from simplejrpc import GLogger

logger = GLogger.get_instance()

# 记录应用启动信息
logger.info("应用程序启动")
logger.info("加载配置文件: config.yaml")
logger.info("连接数据库成功")

# 记录业务操作
def create_user(username, email):
    logger.info(f"创建新用户: {username}")
    # 创建用户逻辑
    user_id = 12345
    logger.info(f"用户创建成功: ID={user_id}, 用户名={username}")
    return user_id

# 记录系统状态
logger.info("系统运行正常")
logger.info("当前活跃连接数: 25")

`warning(message: str)`

记录警告级别的日志。用于记录可能出现问题但不影响程序继续运行的情况。

参数:

message: 字符串类型，日志消息

返回值: 无

示例:

from simplejrpc import GLogger

logger = GLogger.get_instance()

# 记录性能警告
logger.warning("数据库查询耗时过长: 5.2秒")
logger.warning("内存使用率达到 85%")

# 记录配置警告
def load_config(config_path):
    if not os.path.exists(config_path):
        logger.warning(f"配置文件不存在，使用默认配置: {config_path}")
        return default_config
    return load_from_file(config_path)

# 记录业务警告
def process_payment(amount):
    if amount > 10000:
        logger.warning(f"大额支付警告: {amount} 元")
    # 处理支付逻辑

`error(message: str)`

记录错误级别的日志。用于记录程序出现错误但可以继续运行的情况。

参数:

message: 字符串类型，日志消息

返回值: 无

示例:

from simplejrpc import GLogger

logger = GLogger.get_instance()

# 记录异常错误
def divide_numbers(a, b):
    try:
        result = a / b
        return result
    except ZeroDivisionError as e:
        logger.error(f"除零错误: {a} / {b} - {str(e)}")
        return None
    except Exception as e:
        logger.error(f"计算错误: {str(e)}")
        return None

# 记录网络错误
def fetch_data(url):
    try:
        response = requests.get(url, timeout=5)
        return response.json()
    except requests.RequestException as e:
        logger.error(f"网络请求失败: {url} - {str(e)}")
        return None

# 记录数据库错误
def save_user(user_data):
    try:
        # 保存用户数据
        db.save(user_data)
        logger.info("用户数据保存成功")
    except DatabaseError as e:
        logger.error(f"数据库保存失败: {str(e)}")
        raise

`critical(message: str)`

记录严重错误级别的日志。用于记录可能导致程序无法继续运行的严重错误。

参数:

message: 字符串类型，日志消息

返回值: 无

示例:

from simplejrpc import GLogger
import sys

logger = GLogger.get_instance()

# 记录系统级严重错误
def initialize_database():
    try:
        # 初始化数据库连接
        db.connect()
    except Exception as e:
        logger.critical(f"数据库初始化失败，程序无法启动: {str(e)}")
        sys.exit(1)

# 记录资源耗尽
def check_system_resources():
    memory_usage = get_memory_usage()
    if memory_usage > 95:
        logger.critical(f"内存使用率过高: {memory_usage}%，系统可能崩溃")
        
    disk_usage = get_disk_usage()
    if disk_usage > 98:
        logger.critical(f"磁盘空间不足: {disk_usage}%，无法继续写入")

# 记录安全问题
def validate_security():
    if detect_security_breach():
        logger.critical("检测到安全漏洞，立即停止服务")
        shutdown_service()

`configure(config: Dict[str, Any])`

配置日志系统。允许运行时动态更新日志配置。

参数:

config: 字典类型，日志配置信息

返回值: 无

配置选项:

level: 日志级别（DEBUG、INFO、WARNING、ERROR、CRITICAL）
format: 日志格式字符串
handlers: 输出处理器列表
rotation: 日志轮转设置
retention: 日志保留设置
compression: 压缩设置

示例:

from simplejrpc import GLogger

logger = GLogger.get_instance()

# 基本配置
basic_config = {
    "level": "INFO",
    "format": "{time} | {level} | {message}"
}
logger.configure(basic_config)

# 高级配置
advanced_config = {
    "level": "DEBUG",
    "format": "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}",
    "handlers": [
        {
            "sink": "logs/app_{time:YYYY-MM-DD}.log",
            "rotation": "00:00",  # 每天午夜轮转
            "retention": "30 days",  # 保留30天
            "compression": "zip",  # 压缩旧日志
            "level": "INFO"
        },
        {
            "sink": "logs/error.log",
            "level": "ERROR",
            "rotation": "10 MB",  # 按大小轮转
            "retention": "1 week"
        },
        {
            "sink": "sys.stdout",
            "colorize": True,
            "level": "DEBUG"
        }
    ]
}
logger.configure(advanced_config)

# 生产环境配置
production_config = {
    "level": "WARNING",
    "format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | {extra[request_id]} | {message}",
    "handlers": [
        {
            "sink": "/var/log/myapp/app.log",
            "rotation": "1 day",
            "retention": "90 days",
            "compression": "gz"
        },
        {
            "sink": "syslog",  # 系统日志
            "level": "ERROR"
        }
    ]
}
logger.configure(production_config)

注意事项

单例模式：GLogger 采用单例模式，确保全局日志配置的一致性
线程安全：基于 loguru 库，天然支持多线程环境
性能优化：支持异步日志写入，不会阻塞主线程
自动轮转：支持按时间、大小等条件自动轮转日志文件
格式化：支持丰富的日志格式化选项，包括时间、级别、位置等
过滤器：可以通过配置实现日志过滤和分级输出
压缩存储：支持自动压缩历史日志文件，节省存储空间

最佳实践

合理使用日志级别：
- DEBUG：开发调试时使用
- INFO：记录重要的业务流程
- WARNING：记录潜在问题
- ERROR：记录错误但程序可继续
- CRITICAL：记录严重错误
结构化日志：使用一致的日志格式，便于后续分析
避免敏感信息：不要在日志中记录密码、密钥等敏感信息
适当的日志轮转：根据应用规模设置合适的轮转策略
监控日志文件：定期检查日志文件大小和磁盘使用情况

开发最佳实践

错误处理

from simplejrpc import ServerApplication, RPCException, GLogger

app = ServerApplication("/tmp/rpc.sock")
logger = GLogger.get_instance()

@app.route("safe_divide")
def safe_divide(a: float, b: float):
    try:
        if b == 0:
            raise RPCException("除数不能为零", code=-32001)
        result = a / b
        logger.info(f"计算成功: {a} / {b} = {result}")
        return result
    except Exception as e:
        logger.error(f"计算错误: {str(e)}")
        raise RPCException(f"计算失败: {str(e)}", code=-32002)

配置管理

from simplejrpc import ServerApplication, GConfig

# 统一配置管理
config = GConfig(config_path="config.yaml")
app = ServerApplication(
    socket_path=config.get("server.socket_path", "/tmp/rpc.sock"),
    config=config
)

@app.route("get_config")
def get_config(key: str):
    return config.get(key, "配置不存在")

中间件使用

from simplejrpc import ServerApplication
import time

app = ServerApplication("/tmp/rpc.sock")

# 性能监控中间件
def performance_middleware(request, call_next):
    start_time = time.time()
    response = call_next(request)
    duration = time.time() - start_time
    
    if duration > 1.0:  # 超过1秒的请求记录警告
        logger.warning(f"慢请求: {request.method} 耗时 {duration:.3f}s")
    
    return response

app.add_middleware(performance_middleware)

表单验证

from simplejrpc import ServerApplication, BaseForm
from simplejrpc.fields import StringField, IntegerField, BooleanField

class UserCreateForm(BaseForm):
    username = StringField(min_length=3, max_length=20)
    age = IntegerField(min_value=1, max_value=120)
    is_active = BooleanField(default=True)

@app.route("create_user", form=UserCreateForm)
def create_user(data: UserCreateForm):
    # 数据已经通过验证
    return {
        "id": 123,
        "username": data.username,
        "age": data.age,
        "is_active": data.is_active
    }

总结

simplejrpc 模块提供了一套完整的 JSON-RPC 服务器开发解决方案，主要特性包括：

核心功能

ServerApplication：RPC 服务器核心类，支持路由注册、中间件、配置管理
配置系统：Settings、GConfig 提供灵活的配置管理
日志系统：GLogger 基于 loguru 的强大日志功能
表单验证：丰富的字段类型和验证器
异常处理：完善的 RPC 异常体系
国际化：T、GI18n 提供多语言支持
传输层：Unix 域套接字高性能通信

优势特点

易于使用：简洁的 API 设计，快速上手
高性能：基于 Unix 域套接字，低延迟通信
可扩展：中间件系统支持功能扩展
类型安全：完整的类型注解支持
生产就绪：完善的日志、配置、异常处理
国际化：内置多语言支持