Hagworm

Network Development Suite - 异步网络开发套件

项目理念

Hagworm 是一个面向异步编程的 Python 工具库，旨在提供高效、可靠的网络开发基础设施。项目以 asyncio 为核心，封装了从基础工具函数到分布式组件的全栈开发能力，帮助开发者快速构建高性能的异步应用。

核心理念：

• 异步优先：所有 I/O 操作均基于 asyncio，支持 uvloop 加速
• 开箱即用：提供 FastAPI/gRPC/RabbitMQ/Nacos 等框架的集成方案
• 生产就绪：包含日志、配置、事务、缓存、限流等生产级组件
• 优雅抽象：通过上下文管理器、装饰器等模式简化资源管理

安装

pip install hagworm

要求 Python >= 3.11。

---

工程目录结构

hagworm/
├── hagworm/                          # 核心代码
│   ├── __init__.py                   # 版本信息与品牌标识
│   ├── extend/                       # 基础扩展模块
│   │   ├── asyncio/                  # 异步扩展模块
│   │   │   ├── base.py               # 异步基础工具类
│   │   │   ├── buffer.py             # 队列缓冲与文件缓冲
│   │   │   ├── command.py            # 多进程命令框架
│   │   │   ├── file.py               # 带缓存的文件加载器
│   │   │   ├── future.py             # Future/线程/子进程工具
│   │   │   ├── mail.py               # SMTP 邮件客户端
│   │   │   ├── mongo.py              # MongoDB 连接管理
│   │   │   ├── mysql.py              # MySQL 连接管理
│   │   │   ├── net.py                # HTTP 客户端
│   │   │   ├── pool.py               # 通用对象池
│   │   │   ├── process.py            # 子进程托管
│   │   │   ├── redis.py              # Redis 连接管理
│   │   │   ├── socket.py             # TCP/Unix Socket 服务端与客户端
│   │   │   ├── task.py               # 定时任务/并发任务/限流器
│   │   │   └── transaction.py        # 异步事务
│   │   ├── base.py                   # 同步基础工具类
│   │   ├── cache.py                  # 堆栈缓存
│   │   ├── compile.py                # PYC 编译工具
│   │   ├── config.py                 # 配置管理
│   │   ├── crypto.py                 # RSA 签名/验签
│   │   ├── error.py                  # 异常处理
│   │   ├── event.py                  # 同步事件总线
│   │   ├── igraph.py                 # 图数据结构
│   │   ├── interface.py              # 接口定义
│   │   ├── logging.py                # 日志系统
│   │   ├── metaclass.py              # 单例元类
│   │   ├── ntp.py                    # NTP 时间同步
│   │   ├── process.py                # 多进程管理
│   │   ├── qrcode.py                 # 二维码生成
│   │   ├── struct.py                 # 数据结构
│   │   ├── text.py                   # 文本处理
│   │   ├── trace.py                  # 链路追踪
│   │   ├── transaction.py            # 同步事务
│   │   └── validator.py              # 验证器
│   ├── frame/                        # 应用框架
│   │   ├── fastapi/                  # FastAPI 集成
│   │   │   ├── base.py               # FastAPI 应用创建与中间件
│   │   │   ├── field.py              # 自定义字段类型
│   │   │   ├── model.py              # 数据模型
│   │   │   └── response.py           # 响应封装
│   │   ├── gunicorn.py               # Gunicorn Worker
│   │   └── stress_tests.py           # 压力测试框架
│   ├── static/                       # 静态资源（Swagger UI）
│   └── third/                        # 第三方集成
│       ├── grpc/                     # gRPC 框架
│       │   ├── bridge.py             # gRPC 网桥
│       │   ├── client.py             # gRPC 客户端
│       │   └── server.py             # gRPC 服务端
│       ├── nacos/                    # Nacos 服务发现
│       │   ├── client.py             # Nacos 客户端
│       │   └── config.py             # Nacos 配置
│       └── rabbitmq/                 # RabbitMQ 消息队列
│           ├── consume.py            # 消费者
│           ├── publish.py            # 生产者
│           └── rpc.py                # RPC 实现
├── example/                          # 使用示例
│   ├── fastapi_demo/                 # FastAPI 示例项目（main.py / setting.py / controller / service / model / config / build.sh / main.sh / gunicorn.config.py）
│   ├── grpc_demo/                    # gRPC 示例（server.py / client.py）
│   ├── main_rabbitmq.py              # RabbitMQ 示例
│   ├── main_tcp.py                   # TCP 服务示例
│   └── main_test.py                  # 压力测试示例
├── testing/                          # 单元测试（含 main.py 入口与 setup.cfg pytest 配置）
├── .gitignore                        # Git 忽略规则
├── LICENSE                           # 许可证
├── README.md                         # 项目说明
├── install.sh                        # 安装脚本
├── upload.sh                         # 发布脚本
└── pyproject.toml                    # 项目配置

---

模块详解

---

hagworm.extend.base — 同步基础工具类

Utils

聚合工具类，继承自 UtilsSystem、UtilsCodec、UtilsTime、UtilsConvert、UtilsCrypto、UtilsMisc，可直接通过 Utils.xxx 访问全部方法。此外提供以下常用类属性与静态方法：

名称	类型	说明
log	loguru.Logger	全局日志对象
path	模块	os.path 路径工具
math / random / textwrap / itertools	模块	常用标准库模块引用
func_wraps(func)	静态方法	等价于 functools.wraps
func_partial(func, *args, **kwargs)	静态方法	等价于 functools.partial
randint(a, b)	静态方法	随机整数
randstr(population, k)	静态方法	随机采样（random.sample）
getenv(key) / getpid() / getppid()	静态方法	环境变量 / 进程 ID
date_parse(str)	静态方法	解析日期字符串
urlparse(url) / deepcopy(obj)	静态方法	URL 解析 / 深拷贝
suppress(*exceptions)	静态方法	上下文管理器，抑制指定异常

UtilsSystem

系统与环境相关工具。

方法	入参	返回值	说明
get_class_name(val)	val: Type	str	获取类的全限定名
environment()	无	Dict	获取 Python 和系统环境信息
print_slogan()	无	None	打印 Hagworm 品牌 Logo
deprecation_warning(val)	val: str	None	发出弃用警告
get_node_id()	无	str	获取基于 PID 和 MAC 的节点 ID（MD5）
get_cpu_percent()	无	float	获取 CPU 使用率（1秒缓存）
kill_process(*names)	*names: str	None	按名称杀死进程

UtilsCodec

编码与序列化相关工具。

方法	入参	返回值	说明
utf8(val)	val: AnyStr	bytes	将字符串或 bytes 转为 UTF-8 bytes
basestring(val)	val: AnyStr	str	将 bytes 或 str 转为 str
vague_string(val, vague_str, vague_len, min_vague, max_vague)	val: str 等	str	字符串脱敏，如 "13812345678" → "138******78"
json_encode(val, **kwargs)	val: Any	str	JSON 编码（使用 ujson）
json_decode(val, **kwargs)	val: AnyStr	Any	JSON 解码（ujson 优先，回退 json5）
msgpack_encode(val, **kwargs)	val: Any	bytes	MessagePack 编码
msgpack_decode(val, **kwargs)	val: AnyStr	Any	MessagePack 解码
xml_encode(dict_val, root_tag)	dict_val: Dict 等	xml.dom.minidom.Document	字典转 XML
xml_decode(val)	val: str	Dict	XML 转字典
b32_encode(val, standard)	val: AnyStr, standard: bool=False	str	Base32 编码
b32_decode(val, standard, for_bytes)	val: AnyStr 等	AnyStr	Base32 解码
b64_encode(val, standard, for_bytes)	val: AnyStr 等	AnyStr	Base64 编码（默认 urlsafe）
b64_decode(val, standard, for_bytes)	val: AnyStr 等	AnyStr	Base64 解码（默认 urlsafe）
yaml_encode(data, stream)	data: Any 等	str	YAML 编码
yaml_decode(stream)	stream: Any	Any	YAML 解码
pickle_dumps(val)	val: Any	bytes	Pickle 序列化（带 zlib 压缩）
pickle_loads(val)	val: bytes	Any	Pickle 反序列化（带 zlib 解压）

注：utf8 / basestring 对 None 入参原样返回 None（不会转为 bytes / str）。

UtilsTime

时间与日期相关工具。

方法	入参	返回值	说明
today()	无	datetime	获取今天零点时间
yesterday()	无	datetime	获取昨天零点时间
localnow()	无	datetime	获取当前本地时间（分钟级截断，秒与微秒清零）
localnow_on_hour()	无	datetime	获取当前本地时间（小时级截断）
utcnow()	无	datetime	获取当前 UTC 时间（分钟级截断，秒与微秒清零）
utcnow_on_hour()	无	datetime	获取当前 UTC 时间（小时级截断）
timestamp()	无	int	获取当前秒级时间戳
timestamp_ms()	无	int	获取当前毫秒级时间戳
time2stamp(time_str, format_type, timezone_)	time_str: str, format_type: str='%Y-%m-%d %H:%M:%S', timezone_: Optional[str]=None	int	时间字符串转时间戳
stamp2time(time_int, format_type, timezone_)	time_int: Optional[int]=None, format_type: str='%Y-%m-%d %H:%M:%S', timezone_: Optional[str]=None	str	时间戳转时间字符串
get_today_region(today)	today: datetime=None	Tuple[int, int]	获取当天的时间戳范围
get_month_region(today)	today: datetime=None	Tuple[int, int]	获取当月的时间戳范围
get_week_region(today)	today: datetime=None	Tuple[int, int]	获取本周的时间戳范围

UtilsConvert

类型转换与字符串相关工具。

方法	入参	返回值	说明
convert_bool(val)	val: Any	bool	将值转为布尔值，null/none/nil/false/'0'/''/False/0 为 False
convert_int(val, default)	val: Any, default: int=0	int	将 int/float/str 转为 int
convert_float(val, default)	val: Any, default: float=0	float	将值转为 float
interval_limit(val, min_val, max_val)	val, min_val, max_val	int/float	将值限制在区间内
split_float(val, sep, min_split, max_split)	val: str 等	List[float]	按分隔符拆分字符串为浮点数列表
join_float(val, sep)	val: Iterable, sep: str	str	将浮点数列表合并为字符串
split_int(val, sep, min_split, max_split)	val: str 等	List[int]	按分隔符拆分字符串为整数列表
join_int(val, sep)	val: Iterable, sep: str	str	将整数列表合并为字符串
split_str(val, sep, min_split, max_split)	val: str 等	List[str]	按分隔符拆分字符串
join_str(val, sep)	val, sep	str	将值列表合并为字符串
list_extend(iterable, val)	iterable: List, val: Any	None	智能扩展列表，可迭代则 extend，否则 append
str_len(str_val)	str_val: str	int	计算字符串显示宽度（中文占2）
sub_str(val, length, suffix)	val: str, length: int 等	str	按显示宽度截断字符串
re_match(pattern, value)	pattern: str, value: str	bool	正则匹配
ordered_dict(val)	val: Dict=None	OrderedDict	创建按键排序的有序字典
is_iterable(obj, standard)	obj: Any, standard: bool=False	bool	判断是否可迭代
params_join(params, excludes)	params: Dict, excludes: List	str	参数排序拼接
params_sign(*args, **kwargs)	*args, **kwargs	str	参数签名（MD5）

UtilsCrypto

进制、哈希与加密相关工具。

方法	入参	返回值	说明
radix24(val, align)	val: int, align: int=0	str	10 进制转 24 进制
radix24_to_10(val)	val: str	int	24 进制转 10 进制
radix36(val, align)	val: int, align: int=0	str	10 进制转 36 进制
radix36_to_10(val)	val: str	int	36 进制转 10 进制
radix62(val, align)	val: int, align: int=0	str	10 进制转 62 进制
radix62_to_10(val)	val: str	int	62 进制转 10 进制
radix_n(val, base, radix, align)	val: int 等	str	通用进制转换
radix_n_to_10(val, base, radix)	val: str 等	int	通用进制转 10 进制
jwt_encode(val, key, algorithms)	val: dict, key: str, algorithms: str='HS256'	str	JWT 编码
jwt_decode(val, key, algorithms)	val: str, key: str, algorithms: str='HS256'	Dict	JWT 解码
uuid1(node, clock_seq)	node, clock_seq	str	生成 UUID1（hex 格式）
uuid1_urn(node, clock_seq)	node, clock_seq	str	生成 UUID1（URN 格式）
crc32(val)	val: AnyStr	int	CRC32 校验
md5(val)	val: AnyStr	str	MD5 哈希
md5_u32(val)	val: AnyStr	int	MD5 取前 32 位整数
md5_u64(val)	val: AnyStr	int	MD5 取前 64 位整数
sha1(val)	val: AnyStr	str	SHA1 哈希
sha256(val)	val: AnyStr	str	SHA256 哈希
sha512(val)	val: AnyStr	str	SHA512 哈希
hmac_md5(key, data, b64_std)	key, data: AnyStr 等	str	HMAC-MD5 签名
hmac_sha1(key, data, b64_std)	key, data: AnyStr 等	str	HMAC-SHA1 签名

UtilsMisc

网络、随机、校验与文件相关工具。

方法	入参	返回值	说明
rand_hit(val, prob)	val: List, prob: List/Callable	Any	按概率随机命中
urandom_seed()	无	int	使用 os.urandom 生成随机种子并播种 random
read_urandom(size)	size: int	bytes	使用 os.urandom 读取随机字节
get_host_ip()	无	str	获取本机默认路由的出口 IP（本地接口地址，通常为内网地址）
ip2int(val)	val: str	int	IP 地址转整数
int2ip(val)	val: int	str	整数转 IP 地址
luhn_valid(val)	val: str	bool	Luhn 算法校验（银行卡）
luhn_sign(val)	val: str	str	Luhn 算法签名
identity_card(val)	val: str	bool	中国身份证号校验
zip_file(zip_file, *file_paths)	zip_file: str, *file_paths: str	None	压缩文件
unzip_file(zip_file, file_paths)	zip_file: str, file_paths: str	None	解压文件

⚠️ 易混淆点：b64_encode/b64_decode 默认使用 urlsafe 模式；编码时会去除 = 填充，解码时会自动补全 = 填充。如需标准 Base64 请设置 standard=True。

from hagworm.extend.base import Utils

# 字符串脱敏
Utils.vague_string("13812345678")  # "138******78"

# 进制转换
Utils.radix62(12345)  # "3d7"
Utils.radix62_to_10("3d7")  # 12345

# 时间范围
start, end = Utils.get_today_region()
start, end = Utils.get_month_region()

# Base64 编码（注意 urlsafe 与 standard 的区别）
Utils.b64_encode("hello")  # urlsafe, 无填充
Utils.b64_encode("hello", standard=True)  # 标准 Base64

# JSON 编解码
Utils.json_encode({"key": "value"})
Utils.json_decode('{"key": "value"}')

# 身份证校验
Utils.identity_card("110101199001011234")  # True/False

# 参数签名
Utils.params_sign("arg1", "arg2", key1="val1", key2="val2")

FuncWrapper

将多个函数包装成一个可调用对象，调用时依次执行所有注册的函数。

方法	入参	返回值	说明
__call__(*args, **kwargs)	任意参数	None	依次调用所有注册函数
is_valid (属性)	无	bool	是否有注册函数
add(func)	func: Callable	bool	添加函数，已存在返回 False
remove(func)	func: Callable	bool	移除函数，不存在返回 False

from hagworm.extend.base import FuncWrapper

wrapper = FuncWrapper()

def handler1(data):
    print(f"handler1: {data}")

def handler2(data):
    print(f"handler2: {data}")

wrapper.add(handler1)
wrapper.add(handler2)

wrapper("test")  # 依次调用 handler1 和 handler2

TimerMS

代码块耗时统计（毫秒）。作为上下文管理器使用，退出 with 时统计耗时；传入限时 limit（毫秒）后，耗时超过限时时通过 Utils.log 以警告级别打印，默认 0 或不超限时不打印。构造时传入标签用于日志区分。同时可当数值使用（读取、比较、格式化）。

成员	入参	返回值	说明
__init__(label, limit)	label: str, limit: float=0	TimerMS	创建计时器，label 用于日志区分；limit 为限时(毫秒)，0 表示不处理
__enter__()	无	TimerMS	进入 with 块并重置计时起点（实例可复用）
__exit__(...)	异常信息	None	退出 with 块统计耗时；超过 limit 时以警告级别打印 [label] cost xxx.xxxms exceed limit Xms
cost (属性)	无	float	只读耗时(毫秒)；块内为 0.0，退出后有效
int(timer) / float(timer)	无	int / float	取耗时（int 为四舍五入）
str(timer) / format(timer, spec)	spec: str	str	按耗时格式化（str 固定三位小数，如 12.345）
比较运算	int/float/TimerMS	bool	如 timer > 200、timer == other_timer

from hagworm.extend.base import TimerMS

with TimerMS("load config") as timer:
    load_config()
# 默认 limit=0：不打印日志

# 传入限时(毫秒)：超过则以警告级别输出
with TimerMS("load config", limit=100) as timer:
    load_config()
# 超时输出: [load config] cost 152.001ms exceed limit 100ms

print(timer.cost)    # 12.345
print(int(timer))    # 12
assert timer > 10

⚠️ 注意：耗时在退出 with 后才确定，块内 cost 恒为 0.0；因此对象不可哈希（不能作为 set/dict 的键）。

---

hagworm.extend.struct — 数据结构

Enum / IntEnum

增强版的枚举类，提供字典转换和键值查询功能。

方法	入参	返回值	说明
items()	无	Iterable	返回 (name, value) 对
keys()	无	Iterable	返回所有 name
values()	无	Iterable	返回所有 value
to_dict()	无	Dict	返回 {name: value} 字典
to_keys_dict()	无	Dict	返回 {name: enum_item} 字典
to_values_dict()	无	Dict	返回 {value: enum_item} 字典
has_key(key)	key	bool	是否包含指定 name
has_value(value)	value	bool	是否包含指定 value

from hagworm.extend.struct import Enum, IntEnum

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

Color.to_dict()       # {"RED": "red", "GREEN": "green", "BLUE": "blue"}
Color.has_key("RED")  # True
Color.has_value("red")  # True

class Status(IntEnum):
    ACTIVE = 1
    INACTIVE = 0

Status.items()  # dict_items([("ACTIVE", 1), ("INACTIVE", 0)])

RoundRobin / RoundRobinSimple

轮询调度器。RoundRobin 为命名轮询（成员以 name 关联，get() 返回 (name, member) 元组）；RoundRobinSimple 为无命名轮询，API 不同。

RoundRobin

方法	入参	返回值	说明
RoundRobin(members)	members: Dict[str, T]=None	RoundRobin	创建命名轮询器
append(name, member)	name: str, member: T	Optional[T]	添加成员，返回被替换的旧成员
clear(names)	names: List[str]=None	Dict	清除成员，返回被清除的成员
reset(members)	members: Dict=None	Dict	重置成员列表，返回被替换的旧成员
get()	无	Tuple[str, T]	获取下一个 (name, member)
__len__()	无	int	成员数量，支持 len(rr)
__contains__(name)	name: str	bool	是否包含指定名称成员，支持 name in rr

RoundRobinSimple

方法	入参	返回值	说明
RoundRobinSimple(members)	members: List[T]=None	RoundRobinSimple	创建简单轮询器
append(member)	member: T	None	添加成员
clear()	无	List	清空成员，返回被清空的成员列表
get()	无	T	获取下一个成员
__len__()	无	int	成员数量，支持 len(rrs)

from hagworm.extend.struct import RoundRobin, RoundRobinSimple

# 命名轮询
rr = RoundRobin({"server1": "10.0.0.1", "server2": "10.0.0.2"})
name, member = rr.get()  # 轮询获取 (name, member)

# 简单轮询
rrs = RoundRobinSimple([1, 2, 3])
val = rrs.get()  # 轮询获取单个成员

ThreadList / ThreadDict

多线程安全的列表和字典，基于 threading.local 实现。

from hagworm.extend.struct import ThreadList, ThreadDict

tls_list = ThreadList()
tls_list.data.append("value")

tls_dict = ThreadDict()
tls_dict.data["key"] = "value"

ByteArrayAbstract / ByteArray

ByteArrayAbstract 是结构化二进制读写的抽象基类，下列 read_* / write_* 方法均定义于其中（SharedByteArray 等也基于它）；ByteArray 继承 BytesIO 与 ByteArrayAbstract，是可实例化的内存缓冲实现，支持结构化二进制数据的读写。

支持的字段读写方法：read_char/write_char、read_signed_char/write_signed_char、read_unsigned_char/write_unsigned_char、read_short/write_short、read_unsigned_short/write_unsigned_short、read_int/write_int、read_unsigned_int/write_unsigned_int、read_long/write_long、read_unsigned_long/write_unsigned_long、read_long_long/write_long_long、read_unsigned_long_long/write_unsigned_long_long、read_float/write_float、read_double/write_double、read_bool/write_bool、read_bytes/write_bytes、read_string/write_string、read_pascal_bytes/write_pascal_bytes、read_pascal_string/write_pascal_string、read_pad_byte/write_pad_byte（补位）。

字节序通过 get_endian() / set_endian(value) 读取与设置，支持字节序：NETWORK(!)、NATIVE(=)、NATIVE_ALIGNMENT(@)、LITTLE_ENDIAN(<)、BIG_ENDIAN(>)。

from hagworm.extend.struct import ByteArray

buf = ByteArray()
buf.set_endian(ByteArray.NETWORK)
buf.write_int(42)
buf.write_string("hello")
buf.write_float(3.14)

# 读取需要重新定位
buf.seek(0)
val_int = buf.read_int()    # 42
val_str = buf.read_string(5)  # "hello"
val_float = buf.read_float()  # 3.14

KeyLowerDict

将字典的 CamelCase 键自动转为 snake_case 小写键。

from hagworm.extend.struct import KeyLowerDict

data = KeyLowerDict({"MySQLHost": "localhost", "RedisPort": 6379})
# {"my_sqlhost": "localhost", "redis_port": 6379}

⚠️ 注意：连续大写字母之间不会插入下划线，例如 MySQLHost 会得到 my_sqlhost（而非 my_sql_host）。

⚠️ 注意：KeyLowerDict 会递归处理嵌套字典，所有层级的键都会被转换。

---

hagworm.extend.cache — 堆栈缓存

StackCache

基于 TTLCache 的内存高速缓存。

方法	入参	返回值	说明
__init__(maxsize, ttl)	maxsize: float=0xff, ttl: float=60	StackCache	初始化缓存
has(key)	key: str	bool	是否存在缓存
get(key, default)	key: str, default: Any=None	Any	获取缓存值
set(key, val)	key: str, val: Any	None	设置缓存
incr(key, val)	key: str, val: float=1	float	自增
decr(key, val)	key: str, val: float=1	float	自减
delete(key)	key: str	None	删除缓存（key 不存在时抛出 KeyError）
size()	无	int	缓存大小
clear()	无	None	清空缓存

from hagworm.extend.cache import StackCache

cache = StackCache(maxsize=100, ttl=60)

cache.set("user:1", {"name": "Alice"})
cache.get("user:1")  # {"name": "Alice"}
cache.has("user:1")  # True
cache.incr("counter")  # 1
cache.incr("counter")  # 2
cache.decr("counter")  # 1

---

hagworm.extend.config — 配置管理

Field

配置字段定义，用于声明配置项所属 section 和默认值。

from hagworm.extend.config import Field, HostType, StrListType

Port: int = Field("Base", default=8080)
Debug: bool = Field("Base", default=False)
Server: HostType = Field("Server")  # 自动解析 "host:port" 为 (host, port)
Tags: StrListType = Field("App")    # 自动按 "|" 分割为字符串列表

内置类型解码器（通过 decode 类方法将配置字符串解析为目标类型）：

类型	说明
HostType	解析 "host:port" 为 (host, port)
JsonType	解析 JSON 字符串
StrListType	按 | 解析为字符串列表
IntListType	按 , 解析为整数列表
FloatListType	按 , 解析为浮点数列表

配置基类 ConfigureBase：Field 字段通过元类 ConfigureMetaclass（内部以 dataclasses.dataclass(init=False) 包装）自动转为 dataclass 字段；KeySection 为描述单个配置键的 NamedTuple（含 key / type / section / default 四个字段）。

Configure

配置管理类，支持从 INI、JSON、YAML、环境变量等多种来源加载配置。

方法	入参	返回值	说明
read(path, encoding)	path: str, encoding: str='utf-8'	None	从 INI 文件读取
read_env()	无	None	从环境变量读取
read_str(val)	val: str	None	从字符串读取
read_dict(val)	val: Dict	None	从字典读取
read_json(path, encoding)	path: str, encoding: str='utf-8'	None	从 JSON 文件读取
read_yaml(path, encoding)	path: str, encoding: str='utf-8'	None	从 YAML 文件读取
get_option(section, option)	section: str, option: str	Any	获取指定配置项
get_options(section)	section: str	Dict	获取指定 section 的全部配置
set_options(section, **options)	section: str, **options: str	None	动态设置配置
to_dict(section)	section=None	Dict	导出为字典
to_yaml(section)	section=None	str	导出为 YAML

⚠️ 注意：环境变量优先级高于配置文件。环境变量格式为 {SECTION}_{KEY} 全大写。

from hagworm.extend.config import Configure, Field, HostType, StrListType

class AppConfig(Configure):
    Port: int = Field("Base", default=8080)
    Debug: bool = Field("Base", default=False)
    Server: HostType = Field("Server")
    Tags: StrListType = Field("App")

config = AppConfig()
config.read_yaml("./config.yaml")
# config.Port, config.Debug, config.Server 等可直接访问

---

hagworm.extend.error — 异常处理

Ignore

可忽略的异常，用于 with 语句块跳出或多层逻辑跳出。

方法	入参	返回值	说明
__init__(*args, layers)	layers: int=1	Ignore	layers 控制跳出几层 with
log()	无	None	打印 warning 日志
throw()	无	bool	返回是否继续抛出

⚠️ 易混淆点：layers=1（默认）表示跳出当前 with 块后不再继续抛出；layers=2 表示跳出两层 with 块。

catch_warning() / catch_error()

上下文管理器，捕获异常并打印日志。

from hagworm.extend.error import Ignore, catch_error, catch_warning

# 基本用法
with catch_error():
    result = 1 / 0  # 异常被捕获并打印 error 日志

# 使用 Ignore 跳出 with 块
with catch_error():
    if should_skip:
        raise Ignore("skip this block")  # 安全跳出，打印 warning

# 多层跳出
with catch_error():
    with catch_error():
        raise Ignore(layers=2)  # 跳出两层 with 块

---

hagworm.extend.event — 同步事件总线

EventDispatcher

同步事件总线，基于发布-订阅模式。

方法	入参	返回值	说明
dispatch(_type, *args, **kwargs)	_type: str 等	None	触发事件
add_listener(_type, _callable)	_type: str, _callable: Callable	bool	添加事件监听器
remove_listener(_type, _callable)	_type: str, _callable: Callable	bool	移除事件监听器

from hagworm.extend.event import EventDispatcher

dispatcher = EventDispatcher()

def on_login(user_id):
    print(f"User {user_id} logged in")

dispatcher.add_listener("login", on_login)
dispatcher.dispatch("login", "user_123")
dispatcher.remove_listener("login", on_login)

---

hagworm.extend.interface — 接口定义

RunnableInterface

可运行接口，需实现 run() 方法。

TaskInterface

任务接口，需实现 start()、stop()、is_running() 方法。

ContextManager / AsyncContextManager

上下文资源管理器基类，子类实现 _context_release() 即可使用 with/async with 语句管理资源；进入上下文时会调用可覆写的 _context_initialize()（默认空实现），退出时调用 _context_release()。支持 Ignore 异常的安全跳出。

from hagworm.extend.interface import ContextManager

class MyResource(ContextManager):
    def _context_release(self):
        print("资源释放")

with MyResource() as res:
    print("使用资源")
# 自动调用 _context_release

---

hagworm.extend.metaclass — 单例元类

Singleton / SafeSingleton

单例基类。Singleton 非线程安全，SafeSingleton 线程安全。

⚠️ 注意：在多进程 fork 场景下，单例可能会因内存复制导致资源冲突，会发出警告。

Singleton / SafeSingleton 分别由元类 SingletonMetaclass / SafeSingletonMetaclass 实现；SafeSingletonMetaclass 通过加锁保证线程安全，并在跨进程序列化时通过 check_process_id 发出告警。

from hagworm.extend.metaclass import Singleton, SafeSingleton

class Database(Singleton):
    def __init__(self):
        self.connection = "connected"

db1 = Database()
db2 = Database()
assert db1 is db2  # True，同一个实例

---

hagworm.extend.text — 文本处理

StrUtils

字符串工具类。

方法	入参	返回值	说明
to_half_width(value)	value: str	str	全角转半角
to_full_width(value)	value: str	str	半角转全角
to_camel(val)	val: str	str	snake_case → camelCase
to_snake(val)	val: str	str	CamelCase → snake_case
to_upper_camel(val)	val: str	str	snake_case → UpperCamelCase
to_camel_dict(val)	val: Any	Any	递归转换字典键为 camelCase
to_snake_dict(val)	val: Any	Any	递归转换字典键为 snake_case
to_upper_camel_dict(val)	val: Any	Any	递归转换字典键为 UpperCamelCase

from hagworm.extend.text import StrUtils

StrUtils.to_camel("hello_world")       # "helloWorld"
StrUtils.to_snake("helloWorld")        # "hello_world"
StrUtils.to_upper_camel("hello_world") # "HelloWorld"
StrUtils.to_half_width("Ｈｅｌｌｏ")    # "Hello"

StrUtils.to_snake_dict({"firstName": "Alice", "lastName": "Bob"})
# {"first_name": "Alice", "last_name": "Bob"}

TextFinder

基于 Aho-Corasick 算法的关键词查找与替换。

方法	入参	返回值	说明
__init__(words)	words: List[str]	TextFinder	初始化关键词列表
find_all(content)	content: str	List[str]	查找所有匹配关键词
replace_all(content, _chars)	content: str, _chars: str='*'	str	替换所有匹配关键词

from hagworm.extend.text import TextFinder

finder = TextFinder(["敏感词1", "敏感词2", "测试"])
finder.find_all("这是一个测试文本")       # ["测试"]
finder.replace_all("这是一个测试文本")     # "这是一个**文本"

---

hagworm.extend.validator — 验证器

提供常用格式校验函数（IPv4/IPv6/CIDR 基于标准库 ipaddress，UUID 基于标准库 uuid，其余为正则实现）。

函数	入参	返回值	说明
asc_visible(val)	val: str	bool	ASC 可见字符验证
uuid(val)	val: str	bool	UUID 格式验证
email(val)	val: str	bool	邮箱格式验证
domain(val)	val: str	bool	域名格式验证
url(val)	val: str	bool	URL 格式验证
mac_addr(val)	val: str	bool	MAC 地址验证
ipv4(val)	val: str	bool	IPv4 地址验证
ipv4_cidr(val)	val: str	bool	IPv4 CIDR 验证
ipv6(val)	val: str	bool	IPv6 地址验证
ipv6_cidr(val)	val: str	bool	IPv6 CIDR 验证

from hagworm.extend.validator import email, url, ipv4

email("test@example.com")  # True
url("https://example.com")  # True
ipv4("192.168.1.1")        # True
ipv4_cidr("192.168.1.0/24")  # True

---

hagworm.extend.trace — 链路追踪

函数/装饰器	入参	返回值	说明
get_trace_id()	无	Optional[str]	获取当前 trace_id（未初始化时为 None）
refresh_trace_id(trace_id)	trace_id: str=None	str	刷新 trace_id
@trace_wrapper	函数	函数	为函数自动生成 trace_id
@tracing	函数	函数	同步函数耗时追踪日志
@async_tracing	函数	函数	异步函数耗时追踪日志

TRACE_ID_CONTEXT 为保存当前 trace_id 的 ContextVar（默认 None）。

from hagworm.extend.trace import trace_wrapper, async_tracing, refresh_trace_id

@trace_wrapper
async def handle_request():
    # 自动生成新的 trace_id
    pass

@async_tracing
async def slow_operation():
    # 自动记录执行耗时
    pass

refresh_trace_id("custom-trace-id-123")

---

hagworm.extend.logging — 日志系统

init_logger

初始化日志系统。

init_logger(level, *, handler=None, file_path=None, file_name=DEFAULT_LOG_FILE_NAME, file_rotation=DEFAULT_LOG_FILE_ROTATOR, file_retention=0xff, extra=None, enqueue=False, debug=False)，除 level 外全部为关键字参数。

入参	类型	默认值	说明
level	str	必填	日志级别
handler	logging.Handler	None	自定义日志处理器
file_path	str	None	日志文件路径
file_name	str	runtime_{time}.log	日志文件名
file_rotation	Callable	默认轮转器	文件轮转策略
file_retention	int	0xff	保留文件数
extra	Dict	None	额外上下文
enqueue	bool	False	是否队列化
debug	bool	False	调试模式

行为说明：

• 写入文件时，文件名会被改写为 {name}.pid-{pid}{ext}（如 runtime_2026-01-01_00-00-00.pid-1234.log），便于多进程区分。
• 未传 handler 且未传 file_path 时，日志回退输出到 sys.stdout。

模块级常量：

名称	默认值	说明
DEFAULT_LOG_FILE_NAME	runtime_{time}.log	默认日志文件名
DEFAULT_LOG_FILE_ROTATOR	LogFileRotator.make()	默认文件轮转策略
DEFAULT_LOG_MAXSIZE	0xaff	单条日志最大长度（超出截断）
DEFAULT_BUFFER_MAXSIZE	0xfff	缓冲上限
DEFAULT_BULK_MAXSIZE	0x800	批量写入条数上限
DEFAULT_WRITE_MAX_DELAY	5	批量写入最大延迟（秒）

LogFileRotator

日志文件轮转策略，支持按大小和时间轮转。

from hagworm.extend.logging import init_logger, LogFileRotator

init_logger(
    "INFO",
    file_path="/var/log/app",
    file_rotation=LogFileRotator.make(500, "00:00"),  # 500MB 或每天0点轮转
    file_retention=30,
)

KafkaSink / ElasticsearchSink

日志投递到 Kafka 或 Elasticsearch。

• KafkaSink(servers, topic, *, log_maxsize=DEFAULT_LOG_MAXSIZE, buffer_maxsize=DEFAULT_BUFFER_MAXSIZE, bulk_maxsize=DEFAULT_BULK_MAXSIZE, write_max_delay=DEFAULT_WRITE_MAX_DELAY, **kwargs)：servers 为 Kafka 地址，topic 为投递主题。
• ElasticsearchSink(hosts, stream_name, *, log_maxsize=..., buffer_maxsize=..., bulk_maxsize=..., write_max_delay=..., rollover_max_age='7d', rollover_max_primary_shard_size='50gb', delete_min_age='30d', refresh_interval='5s', number_of_replicas=0, **kwargs)：hosts 为 ES 地址，stream_name 为 Data Stream 名称，其余为 ILM/索引模板参数。

from hagworm.extend.logging import KafkaSink, ElasticsearchSink

# Kafka 日志投递
kafka_sink = KafkaSink("localhost:9092", "app-logs")

# Elasticsearch 日志投递
es_sink = ElasticsearchSink("http://localhost:9200", "app-logs")

InterceptHandler

将标准库 logging 的日志转发到 loguru。

QueuedFileSink

队列化的文件 sink，由后台线程按 bulk_maxsize / write_max_delay 批量写入日志文件（与 asyncio 无关）。

QueuedFileSink(path, *, log_maxsize=DEFAULT_LOG_MAXSIZE, buffer_maxsize=DEFAULT_BUFFER_MAXSIZE, bulk_maxsize=DEFAULT_BULK_MAXSIZE, write_max_delay=DEFAULT_WRITE_MAX_DELAY, **kwargs)

ElasticsearchDataStreamUtil / AsyncElasticsearchDataStreamUtil

Elasticsearch Data Stream 索引模板与 ILM 策略初始化工具。

---

hagworm.extend.crypto — RSA 签名/验签

RsaUtil

方法	入参	返回值	说明
gen_rsa_key(rsa_key, private)	rsa_key: str, private: bool=False	str	生成 PEM 格式密钥
rsa_sign(rsa_key, sign_data)	rsa_key: str, sign_data: str	str	RSA 签名
rsa_verify(public_key, verify_data, verify_sign)	三个 str	bool	RSA 验签

说明：rsa_sign / rsa_verify 还支持可选参数 hash_alg（默认 SHA256），用于指定签名哈希算法。

from hagworm.extend.crypto import RsaUtil

private_key = "MIIEvQ..."  # 裸密钥字符串
public_key = "MIIBIj..."

pem_private = RsaUtil.gen_rsa_key(private_key, private=True)
pem_public = RsaUtil.gen_rsa_key(public_key)

signature = RsaUtil.rsa_sign(private_key, "data to sign")
is_valid = RsaUtil.rsa_verify(public_key, "data to sign", signature)

---

hagworm.extend.transaction — 同步事务

TransactionAbstract

事务抽象基类，提供 add_commit_callback / add_rollback_callback / commit / rollback / bind 方法。

Transaction

基于上下文管理器的事务对象，未显式 commit 则自动 rollback。

方法	入参	返回值	说明
add_commit_callback(_callable, *args, **kwargs)	_callable: Callable 等	None	添加提交回调
add_rollback_callback(_callable, *args, **kwargs)	_callable: Callable 等	None	添加回滚回调
commit()	无	None	提交事务，执行所有提交回调
rollback()	无	None	回滚事务，执行所有回滚回调
bind(trx)	trx: TransactionAbstract	None	绑定另一个事务（级联提交/回滚）

TransactionAbstract(*, commit_callback: Optional[Callable] = None, rollback_callback: Optional[Callable] = None) 可在构造时注入初始提交/回滚回调。

from hagworm.extend.transaction import Transaction

with Transaction() as trx:
    trx.add_commit_callback(lambda: print("committed"))
    trx.add_rollback_callback(lambda: print("rolled back"))
    # 如果不调用 trx.commit()，退出 with 时自动 rollback
    trx.commit()

---

hagworm.extend.compile — PYC 编译

compile_pyc

将 Python 源码编译为 PYC 字节码。

入参	类型	说明
file_path	str	源码目录
pyc_file_path	str	输出目录（默认同源码目录）
exclude	str	排除目录（用 | 分隔）

python -m hagworm.extend.compile -i ./src -o ./dist -e "test|tmp"

模块同时暴露 main() CLI 入口，可通过 python -m hagworm.extend.compile ... 直接调用。

---

hagworm.extend.qrcode — 二维码

QRCode

方法	入参	返回值	说明
make(data, **kwargs)	data: Any	bytes	生成 PNG 二维码
make_svg(data, **kwargs)	data: Any	str	生成 SVG 二维码

from hagworm.extend.qrcode import QRCode

png_data = QRCode.make("https://example.com")
svg_str = QRCode.make_svg("https://example.com")

---

hagworm.extend.igraph — 图数据结构

Graph / DiGraph

基于 igraph（包名 igraph）的图数据结构封装。Graph 构造为 Graph(directed, name=None)；DiGraph(name=None) 为有向图快捷类（内部 directed=True）。顶点标识类型别名为 VERTEX_PARAMETER_TYPE = Union[Vertex, str, int]；len(graph) 返回已登记顶点数，repr(graph) 形如 <Graph|DiGraph: name, vertices: n, edges: m>。

方法	入参	返回值	说明
Graph(directed, name)	directed: bool, name: str	Graph	创建图
add_vertex(name, **kwargs)	name: str=None	Vertex	添加顶点
find_vertex(name)	name: str	Vertex	查找顶点
find_vertices(names)	names: List[str]	List[Vertex]	批量查找顶点
add_edge(source, target)	顶点标识	None	添加边
init_edges(edges)	edges: List[Tuple]	None	批量初始化边
get_edge(source, target)	顶点标识	Edge	获取边
del_edge(source, target)	顶点标识	None	删除边
del_vertex_edge(name, mode)	name: str, mode: str='all'	None	删除顶点的关联边
clear()	无	None	清空图
deep_search(name, mode)	name: str, mode: str='all'	List[str]	深度搜索
vertices (属性)	无	Dict[str, Vertex]	顶点字典
name (属性)	无	str	图名称
data (属性)	无	igraph.Graph	底层 igraph 对象
vertex_seq (属性)	无	VertexSeq	顶点序列
edge_seq (属性)	无	EdgeSeq	边序列

DiGraph 额外方法：

方法	入参	返回值	说明
find_in(name)	name: str	List[Vertex]	查找入边顶点
find_out(name)	name: str	List[Vertex]	查找出边顶点
find_tree(name, depth, mode)	name: str, depth: int=0xff, mode: str='out'	Dict	查找依赖树

from hagworm.extend.igraph import DiGraph

graph = DiGraph("service_deps")
graph.add_vertex("service_a")
graph.add_vertex("service_b")
graph.add_vertex("service_c")
graph.add_edge("service_a", "service_b")
graph.add_edge("service_b", "service_c")

graph.find_out("service_a")  # [service_b]
graph.deep_search("service_a", mode="out")  # ["service_b", "service_c"]

---

hagworm.extend.ntp — NTP 时间同步

NTPClient

NTP 客户端（基于后台线程调度器，非 asyncio 异步），定期同步时间偏移。

NTPClient(host, *, version=4, port='ntp', timeout=5, interval=3600, sampling=5)，host 之后均为关键字参数。

入参	类型	默认值	说明
host	str	ntp.aliyun.com	NTP 服务器
version	int	4	NTP 协议版本
port	str	ntp	NTP 端口
timeout	int	5	请求超时（秒）
interval	int	3600	同步间隔（秒）
sampling	int	5	每次采样次数（取中位数作为偏移）

方法	入参	返回值	说明
start()	无	None	启动调度器
stop()	无	None	停止调度器

属性	类型	说明
offset	float	时间偏移量（多次采样的中位数，未成功同步时为 0）
timestamp	float	校准后的秒级时间戳
timestamp_ms	int	校准后的毫秒级时间戳

from hagworm.extend.ntp import NTPClient

ntp = NTPClient("ntp.aliyun.com", interval=1800, sampling=5)
ntp.start()

calibrated_time = ntp.timestamp
calibrated_time_ms = ntp.timestamp_ms

ntp.stop()

---

hagworm.extend.process — 多进程管理

⚠️ 注意：导入 hagworm.extend.process 会执行 multiprocessing.set_start_method('spawn', force=True)，强制改变整个进程的多进程启动方式。此外 fork_processes() 依赖 os.fork，仅 POSIX 可用。

fork_processes()

封装 os.fork()，在子进程中重新播种随机数。返回值与 os.fork() 一致：父进程返回子进程 pid，子进程返回 0。

Daemon

守护进程管理器，管理多个子进程。

Daemon(target: Callable, sub_process_num: int, *, cpu_affinity: bool = False, join_timeout: int = 10, keep_alive: bool = False, **kwargs)

入参	类型	默认值	说明
target	Callable	必填	子进程入口函数，实际以 target(idx, **kwargs) 调用，idx 为槽位序号（0 ~ sub_process_num-1）
sub_process_num	int	必填	子进程数量
cpu_affinity	bool	False	是否绑定 CPU 亲和性（按槽位绑定到同序号核心，数量超过核心数时构造即报错）
join_timeout	int	10	等待子进程退出的超时（秒）
keep_alive	bool	False	是否在子进程退出后自动补齐
**kwargs	Any	无	仅原样透传给 target，不参与主进程逻辑

cpu_affinity / join_timeout / keep_alive 均为关键字参数（sub_process_num 之后的位置参数非法）。

方法	入参	返回值	说明
run()	无	None	启动子进程；keep_alive=True 时子进程退出后自动补齐并持续运行，否则等所有子进程退出后返回

SharedByteArray

基于共享内存的字节数组，支持跨进程读写。

SharedByteArray(name=None, create=False, size=0)，实现 ByteArrayAbstract 的 read(size) -> memoryview 与 write(buffer)，其余 read_* / write_* 方法均继承自该基类。

提供 release() 方法：关闭共享内存句柄，创建模式下同时 unlink。

⚠️ 注意：创建方（create=True）退出 with 块时会关闭并 unlink 共享内存段，因此读取方必须在创建方进程存活期间、于另一个进程中并发打开同一 name。

from hagworm.extend.process import SharedByteArray

# 创建模式（进程 A）：必须在 with 块内完成写入并保持进程存活，
# 读取方（进程 B）需在此期间并发打开同名共享内存
with SharedByteArray(name="shared_buf", create=True, size=1024) as buf:
    buf.write_int(42)
    # ... 保持进程运行并协调进程 B 读取 ...
    # 退出 with 后会 unlink，进程 B 将无法再打开

# 读取模式（进程 B，需在进程 A 仍处于上述 with 块期间并发执行）
with SharedByteArray(name="shared_buf", create=False) as buf:
    val = buf.read_int()  # 42

---

hagworm.extend.asyncio.base — 异步基础工具类

Utils（异步版）

继承同步 Utils，增加异步工具方法。

方法	入参	返回值	说明
sleep(delay, result=None)	delay: float 等	None	异步等待（等价于 asyncio.sleep）
is_coroutine_function(func)	func: Callable	bool	判断是否协程函数（支持 partial）
awaitable_wrapper(obj)	obj: Any	Any	自适应 awaitable 对象
wait_frame(count)	count: int=10	Generator	暂停指定帧数
get_event_loop()	无	AbstractEventLoop	获取事件循环
get_current_loop()	无	AbstractEventLoop	获取当前任务的事件循环
loop_time()	无	float	获取事件循环时间
wait_any_completed(*args)	协程/任务	None	等待任一完成，取消其余
call_soon(callback, *args, loop, context)	callback: Callable 等	Handle	尽快调度（隔离上下文）
call_later(delay, callback, *args, loop, context)	delay: float 等	Handle	延时指定秒数调用
call_at(when, callback, *args, loop, context)	when: float 等	Handle	指定时间调用
create_task(coro)	coro: Coroutine	Task	创建任务
async_timeout(timeout)	timeout: float	异步上下文管理器	异步超时等待

install_uvloop()

尝试安装 uvloop 加速事件循环。

async_adapter(func)

异步函数适配装饰器，使异步函数可以在同步代码中非阻塞调用。

WeakContextVar

弱引用版的 ContextVar，避免上下文变量阻止对象回收。

WeakContextVar(name: str)；get() -> Any（引用已被回收时返回 None）；set(value) -> Token（不可弱引用的对象退化为强引用存储）。

robust_iterator

异步循环器，用于需要重复执行的场景。

入参	类型	默认值	说明
timeout	int/float	0	超时时间（0=不限）
interval	int/float	1	间隔秒数
max_times	int/None	None	最大执行次数（None=不限）

产出 index 从 1 开始计数；timeout 为整段循环的总时限（> 0 才生效）；每轮 yield 之后会 await Utils.sleep(interval)。

from hagworm.extend.asyncio.base import Utils, robust_iterator, ShareFuture, FuncCache

# 异步循环
async for index in robust_iterator(timeout=10, interval=1):
    # 每秒执行一次，最多10秒
    await check_status()

# 共享 Future（并发调用共享计算结果）
@ShareFuture()
async def fetch_data(key):
    # 同一时刻相同参数的调用共享结果
    return await expensive_operation(key)

# 函数缓存
@FuncCache(maxsize=100, ttl=10)
async def get_config(key):
    return await load_config(key)

ShareFuture

共享 Future 装饰器：同一时刻以相同函数签名并发调用时，只执行一次并共享结果（内部用 Utils.params_sign 生成签名，复用 asyncio.Task，并以 asyncio.shield 避免调用方取消影响其他等待者）。

ShareFuture() 无构造参数，用法为 @ShareFuture()。

FuncCache

继承 ShareFuture 的函数缓存，基于 StackCache，在有效期内函数签名一致即命中缓存。

FuncCache(maxsize: int = 0xffff, ttl: float = 10)

注意：函数返回 None 时不会写入缓存，因此每次都需重新计算。

NonBlockingFuncWrapper

非阻塞异步函数包装器，继承自 base.FuncWrapper，将多个同步或异步函数包装成一个可调用对象。调用时通过 call_soon 非阻塞式执行所有注册的函数。

方法	入参	返回值	说明
__call__(*args, **kwargs)	任意参数	None	非阻塞调用所有注册函数
is_valid (属性)	无	bool	是否有注册函数
add(func)	func: Callable	bool	添加函数，已存在返回 False
remove(func)	func: Callable	bool	移除函数，不存在返回 False

from hagworm.extend.asyncio.base import NonBlockingFuncWrapper

wrapper = NonBlockingFuncWrapper()
wrapper.add(sync_handler)
wrapper.add(async_handler)

wrapper("event_data")  # 非阻塞，通过 call_soon 调度

AsyncConstructor

支持异步构造函数的基类。

from hagworm.extend.asyncio.base import AsyncConstructor

class AsyncClient(AsyncConstructor):
    async def __async_init__(self):
        self.connection = await create_connection()

client = await AsyncClient()  # 异步构造

AsyncFuncWrapper

阻塞式异步函数包装器，将多个同步/异步函数包装为一个可 await 调用对象。

from hagworm.extend.asyncio.base import AsyncFuncWrapper

wrapper = AsyncFuncWrapper()
wrapper.add(async_handler1)
wrapper.add(sync_handler2)
await wrapper("event_data")  # 依次 await 所有处理器

---

hagworm.extend.asyncio.future — Future 与线程工具

SignalManager

信号管理器（静态类）。asyncio 的 loop.add_signal_handler 对同一信号是覆盖式注册，多个组件各自注册会互相覆盖（只有最后注册者生效）。SignalManager 在进程内只安装一个信号处理器，所有回调集中登记，信号到达时广播给全部回调，因此同进程内的多个组件（WaitForever、MainProcess/SubProcess、AsyncTcpServer、Worker 等）可以各自注册而不互相覆盖。

处理器直接挂在系统层（signal.signal，要求主线程、且需在运行中的事件循环内登记），再通过 loop.call_soon_threadsafe 唤醒事件循环广播。这样即使事件循环被同步代码阻塞（长计算、阻塞 IO、time.sleep），信号依然会被处理——这是不用 loop.add_signal_handler 的原因（它会用 noop 处理器接管信号，只在循环迭代时才响应）。

重复信号语义：同一信号第二次到达说明上一次广播后进程仍未退出（回调可能只关闭了资源、没有结束事件循环），此时强制退出：事件循环仍在运行时，SIGINT 抛 KeyboardInterrupt，其余信号抛 SystemExit(128 + 信号编号)（如 SIGTERM 退出码 143），两者都会执行 finally 与 atexit，因此子进程兜底清理不会丢；若此时事件循环已停止或关闭（进程已进入收尾阶段，可能正在执行 GC/弱引用回调），则改用 os._exit(128 + 信号编号) 直接强制退出，避免抛异常被解释器当作 Exception ignored 吞掉并打印堆栈。由于处理器在系统层，事件循环被阻塞时重复信号同样生效。

退出事件语义：首次登记时 SignalManager 会自行 atexit.register 一个钩子；解释器退出时把退出事件走同一份登记集合广播一次，让仍登记着的组件有机会收尾，随后 set_wakeup_fd(-1) 并恢复 SIGINT/SIGTERM 的默认处理、清空登记。注意退出阶段没有运行中的事件循环，回调需自行保证在无事件循环环境下安全（框架内回调已处理：Worker 会跳过、MainProcess/SubProcess 仅在循环运行时才 stop）。

信号广播与退出广播都对每个回调做异常隔离（except Exception + warning），单个回调失败不会阻断其余回调与进程退出。

方法	入参	返回值	说明
register(func, *, loop=None)	func: Callable	None	登记 SIGINT/SIGTERM 回调；loop 缺省取当前事件循环（信号广播所在循环，切换时重置重复信号计数）；安装系统层处理器；同一回调只登记一次；首次登记时挂上退出钩子
unregister(func)	func: Callable	None	注销回调，按等值比较（兼容绑定方法）
dispatch()	无	None	在事件循环内广播给所有已登记回调，快照遍历（允许回调中注销自己）

注册与注销示例（框架内各组件即按此方式使用）：

from hagworm.extend.asyncio.future import SignalManager

SignalManager.register(handler, loop=loop)   # 登记回调，首次调用时安装分发器
SignalManager.unregister(handler)            # 组件结束时注销，避免残留回调

WaitForever

永远等待的 Future，通过 SIGINT/SIGTERM 信号退出。

方法	入参	返回值	说明
__init__(loop)	loop: AbstractEventLoop=None	WaitForever	初始化并通过 SignalManager 注册信号处理
__await__()	无	无	await WaitForever() 阻塞等待直到 exit() 被触发
done()	无	bool	是否已退出
exit()	无	None	触发退出并注销信号处理

DelayTask

延迟启动的协程任务，在调用 start() 之前不会执行，完成后可通过 result() 和 exception() 获取结果。

方法	入参	返回值	说明
__init__(coroutine, **kwargs)	coroutine: Coroutine	DelayTask	初始化，传入协程对象
start()	无	None	启动任务执行
cancel()	无	bool	取消任务
running()	无	bool	是否尚未启动
done()	无	bool	任务是否已完成
result()	无	Any	获取任务结果（完成前调用会异常）
exception()	无	Optional[BaseException]	获取任务异常
__await__()	无	Any	await task 等待任务结果

from hagworm.extend.asyncio.future import DelayTask

task = DelayTask(some_coroutine())
# ... 暂未执行
task.start()         # 启动
result = await task  # 等待结果

Thread

继承 threading.Thread 的增强版线程（固定 daemon=True，不可关闭），支持优雅退出。构造签名为 Thread(group=None, target=None, name=None, args=(), kwargs=None)。

方法	入参	返回值	说明
stop(timeout)	timeout: Optional[int]=None	None	设置退出标记；timeout 非 None 时额外 join(timeout) 等待线程退出
is_stopped()	无	bool	是否已标记退出

ThreadPool

线程池，桥接线程与协程。

方法	入参	返回值	说明
__init__(max_workers)	max_workers: int=None	ThreadPool	初始化
run(_callable, *args, **kwargs)	_callable: Callable	Any	在线程中执行函数并 await 结果（不支持协程函数）

ThreadWorker

装饰器，将同步函数转为非阻塞异步函数。

from hagworm.extend.asyncio.future import ThreadWorker, ThreadPool

@ThreadWorker(max_workers=4)
def blocking_io(path):
    with open(path) as f:
        return f.read()

# 在异步代码中调用
result = await blocking_io("/path/to/file")

# 也可以直接使用 ThreadPool
pool = ThreadPool(max_workers=8)
result = await pool.run(os.path.exists, "/path/to/file")

SubProcess

异步子进程管理，通过命令方式启动子进程。

SubProcess(program: str, *args, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, **kwargs)

方法	入参	返回值	说明
create(program, *args, stdin=None, stdout=None, stderr=None, **kwargs)	program: str 等	SubProcess	创建并启动子进程（三个流参数默认值为 None，与构造函数不同）
start()	无	bool	启动子进程（已运行返回 False）
stop()	无	None	结束子进程
kill()	无	None	强制结束子进程
wait(timeout)	timeout: float=None	None	等待子进程结束
is_running()	无	bool	是否运行中
create_stdout_task()	无	Task	创建 stdout 日志任务
create_stderr_task()	无	Task	创建 stderr 日志任务

属性	类型	说明
pid	int	进程 ID
process	asyncio.subprocess.Process	底层进程对象
stdin	StreamWriter	标准输入流
stdout	StreamReader	标准输出流
stderr	StreamReader	标准错误流

生命周期：start() 会注册 atexit 兜底钩子，stop() / kill() 自动注销；解释器退出时若外部命令仍存活则强制 kill，避免残留孤儿进程。

from hagworm.extend.asyncio.future import SubProcess

proc = await SubProcess.create("python", "script.py", stdout=asyncio.subprocess.PIPE)
await proc.wait(timeout=30)

---

hagworm.extend.asyncio.task — 定时任务与并发控制

TaskAbstract

任务基类，实现 TaskInterface，IntervalTask / CronTask 均由它派生（DCSCronTask 直接实现 TaskInterface，与本类无继承关系）。

TaskAbstract(scheduler: Optional[AsyncIOScheduler] = None)：不传 scheduler 时按 DEFAULT_SCHEDULER_CONFIG 创建 AsyncIOScheduler（时区 DEFAULT_SCHEDULER_TIMEZONE，即 Asia/Shanghai，job_defaults = {coalesce: False, max_instances: 1, misfire_grace_time: 10}）。

成员	说明
scheduler	属性，底层 AsyncIOScheduler
start() / stop()	启动 / 关闭调度器（重复调用为空操作）
is_running()	调度器是否运行中
add_job(...)	抽象方法，由子类实现
remove_job(job_id)	移除指定任务
remove_all_jobs()	移除全部任务

IntervalTask

间隔任务，按固定秒数重复执行。

IntervalTask.add_job(interval, func, *args, **kwargs)；由于内部设置 next_run_time=now，启动后会立即执行一次，之后每隔 interval 秒重复。

from hagworm.extend.asyncio.task import IntervalTask

task = IntervalTask.create(60, my_func, arg1, arg2)
task.start()

CronTask

Cron 定时任务。

CronTask.add_job(crontab, func, *args, **kwargs)，crontab 为标准 5 段表达式，时区 Asia/Shanghai。

from hagworm.extend.asyncio.task import CronTask

task = CronTask.create("0 8 * * *", morning_job)  # 每天8点
task.start()

DCSCronTask

分布式 Cron 任务，基于 Redis 实现分布式锁，确保多实例环境下任务只执行一次。锁通过 SET key name NX EX 3600 获取，任务结束后 DELETE；未获得锁时打 debug 日志 dcs cron task idle。

DCSCronTask(redis_client, name, crontab, func, *args, **kwargs)：直接实现 TaskInterface，仅提供 start() / stop() / is_running()，不具备 TaskAbstract 的 scheduler / add_job / remove_job 等成员。

from hagworm.extend.asyncio.task import DCSCronTask
from hagworm.extend.asyncio.redis import RedisPool

dcs_task = DCSCronTask(redis_pool, "daily_report", "0 9 * * *", generate_report)
dcs_task.start()

MultiTasks

多任务并发管理器，通过信号量控制并发数。

提供 append(coroutine) 与 extend(coroutines) 方法，分别用于追加单个协程与批量追加协程。

from hagworm.extend.asyncio.task import MultiTasks

tasks = MultiTasks(tasks_num=10)
tasks.append(fetch_url("http://example.com/1"))
tasks.append(fetch_url("http://example.com/2"))
tasks.append(fetch_url("http://example.com/3"))
results = await tasks  # 最多 10 个并发

RateLimiter

流量控制器，保护计算资源。

进入队列的任务，排队或执行超时会通过 Future 引发 TimeoutError；仅当任务被显式 cancel 时才会引发 CancelledError。

RateLimiter(task_limit: int, wait_limit: int = 256, timeout: int = 60)

入参	类型	默认值	说明
task_limit	int	必填	最大并发任务数
wait_limit	int	256	等待队列长度
timeout	int	60	等待超时（秒）

方法	入参	返回值	说明
append(func, *args, **kwargs)	func: Callable	bool	提交任务（不等待结果）
call(func, *args, **kwargs)	func: Callable	Any	提交任务并等待结果
start()	无	None	启动消费协程
size()	无	int	当前队列长度
join()	无	None	等待所有任务完成
close()	无	None	关闭限流器

from hagworm.extend.asyncio.task import RateLimiter

limiter = RateLimiter(task_limit=5, wait_limit=10, timeout=60)

# 不等待结果
await limiter.append(process_data, data1)
await limiter.append(process_data, data2)

# 等待结果
result = await limiter.call(process_data, data3)

await limiter.join()
limiter.close()

---

hagworm.extend.asyncio.buffer — 缓冲区

QueueBuffer

异步队列缓冲区，批量消费数据。

提供 append(data, timeout=None)（await 入队，达到 slice_size 时唤醒消费协程；timeout 为入队等待超时）、safe_append(data)（线程安全的入队）、size()（当前队列长度）与 close()（异步，排空队列后关闭消费协程与内部 RateLimiter）方法。

入参	类型	默认值	说明
handler	Callable	必填	批量处理函数
slice_size	int	必填	每批大小
slice_time	float	必填	最大等待时间
task_limit	int	必填	并发处理数
wait_limit	int	0	等待队列长度（0=不限）
timeout	int	60	超时

from hagworm.extend.asyncio.buffer import QueueBuffer

async def batch_insert(records):
    await db.insert_many(records)

buffer = QueueBuffer(batch_insert, slice_size=100, slice_time=5, task_limit=4)
await buffer.append(record1)
await buffer.append(record2)
# 满100条或5秒后自动批量处理
await buffer.close()

FileBuffer

文件缓冲区，支持大文件分片读写。

提供 close() 方法：关闭并释放所有分片临时文件。

from hagworm.extend.asyncio.buffer import FileBuffer

with FileBuffer(slice_size=0x1000000) as fb:
    fb.write(b"large data...")
    data = fb.read(1024)

---

hagworm.extend.asyncio.net — HTTP 客户端

HTTPClientPool

基于 httpx 的异步 HTTP 客户端，内置重试机制。

HTTPClientPool(retries: int = 5, verify: ssl.SSLContext | str | bool = False, limits: httpx.Limits = DEFAULT_LIMITS, timeout: TimeoutTypes = DEFAULT_TIMEOUT, **kwargs)

方法	入参	返回值	说明
get(url, **kwargs)	url: str 等	bytes	GET 请求
post(url, **kwargs)	url: str 等	bytes	POST 请求
put(url, **kwargs)	url: str 等	bytes	PUT 请求
patch(url, **kwargs)	url: str 等	bytes	PATCH 请求
delete(url, **kwargs)	url: str 等	bytes	DELETE 请求
options(url, **kwargs)	url: str 等	Dict	OPTIONS 请求，返回响应头
head(url, **kwargs)	url: str 等	Dict	HEAD 请求，返回响应头
send_request(method, url, **kwargs)	method: str 等	Result	通用请求（content/data/files/json/params/headers/cookies 均为关键字参数）
close()	无	None	异步关闭底层 httpx 客户端

send_request 返回 Result 命名元组，字段为 status / headers / cookies / body。

模块级常量：

名称	说明
DEFAULT_RETRIES	默认最大尝试次数 5（首次 + 最多 4 次重试）
DEFAULT_TIMEOUT	默认请求超时 60s
DEFAULT_LIMITS	默认连接池上限（max_connections=200 / max_keepalive_connections=50）
DOWNLOAD_TIMEOUT	Downloader 的默认超时 600s

HTTPTextClientPool

返回文本响应的 HTTP 客户端。

HTTPJsonClientPool

返回 JSON 响应的 HTTP 客户端。

Downloader

流式文件下载器，继承 _ClientBase（超时默认 DOWNLOAD_TIMEOUT，即 600s）。

fetch(file: BinaryIO, url: URL | str, *, chunk_size: int = 65536, params=None, headers=None, cookies=None) -> None，以流式方式将响应体按 chunk_size 写入 file（文件句柄由调用方自行打开与关闭）。

from hagworm.extend.asyncio.net import HTTPClientPool, HTTPJsonClientPool, Downloader

# 基本用法
client = HTTPClientPool(retries=3)
body = await client.get("https://api.example.com/data")

# JSON 客户端
json_client = HTTPJsonClientPool()
data = await json_client.get("https://api.example.com/json")  # 返回 dict/list

# 文件下载
downloader = Downloader()
with open("output.zip", "wb") as f:
    await downloader.fetch(f, "https://example.com/file.zip")

await client.close()

---

hagworm.extend.asyncio.redis — Redis 连接管理

RedisPool / RedisClusterPool

Redis 单机/集群连接池管理。

入参	类型	默认值	说明
host	str	必填	主机
port	int	必填	端口
db	int	0	数据库序号（仅 RedisPool）
startup_nodes	Iterable[Node]	None	集群起始节点（仅 RedisClusterPool）
username	str	None	用户名
password	str	None	密码
max_connections	int	32	最大连接数
max_idle_time	float	43200	最大空闲时间（秒）
idle_check_interval	float	1	空闲检查间隔（秒）
protocol_version	Literal[2, 3]	2	RESP 协议版本

方法	入参	返回值	说明
open()	无	self	初始化连接池
close()	无	None	关闭连接池
set_key_prefix(value)	value: str	None	设置键前缀
get_safe_key(key, *args, **kwargs)	key: str 等	str	生成安全键名
allocate_lock(name, **kwargs)	name: str 等	LuaLock	分配分布式锁
get_obj(name)	name: str	Any	获取 Pickle 对象
getset_obj(name, value)	name: str, value: Any	Any	设置并返回旧的 Pickle 对象
set_obj(name, value, seconds)	name: str, value: Any, seconds: Optional[float]=None	Any	存储 Pickle 对象
decode(value)	value: bytes / tuple / list / dict	Any	静态方法，对顶层及一层容器元素做 utf-8 解码（非递归）

属性 max_connections 为连接池上限。

LuaLock

Redis 分布式锁（coredis 的 LuaLock 封装），实现 AsyncContextManager，可 async with 使用。

LuaLock(client, name, timeout: Optional[float] = None, sleep: float = 0.1, blocking: bool = True, blocking_timeout: Optional[float] = None)

• acquire()：本任务已持锁时改为 extend(timeout) 续期，避免重复加锁。
• release()：仅在本任务持锁时释放。
• async with 退出时自动 release()。

RedisDelegate

Redis 功能组件，简化初始化。

方法	入参	返回值	说明
init_redis_single(...)	同 RedisPool 构造参数	RedisPool	初始化单机连接池
init_redis_cluster(...)	同 RedisClusterPool 构造参数	RedisClusterPool	初始化集群连接池
set_redis_key_prefix(value)	value: str	None	设置键前缀
close_redis_pool()	无	None	关闭连接池
get_redis_client()	无	RedisPool / RedisClusterPool	获取连接池

属性 redis_pool 为当前连接池。注意 init_redis_single / init_redis_cluster 在 username, password 之后的关键字参数（max_connections / max_idle_time / idle_check_interval / protocol_version）均为关键字限定，与 RedisPool 构造函数（无 *）并不完全一致。

ShareCache

基于分布式锁的共享缓存，用于避免并发重复计算。注意：互斥并非强制，仅当调用方严格按 get → 计算 → set → release 流程使用时，才能保证同一时刻业务逻辑只执行一次。

ShareCache(redis_client, share_key, lock_timeout=60, lock_blocking_timeout=60)，实现 AsyncContextManager；属性 result 保存最近一次读到的值，提供 get()、set(value, expire=None) 与异步 release()。

PeriodCounter

周期计数器，基于 Redis 的 incrby/decrby。

PeriodCounter(redis_client, key_prefix, time_slice)，time_slice 为周期秒数（必填）。

方法	入参	返回值	说明
get()	无	Any	获取当前周期计数值
incr(val)	val: int=1	int	增加计数
decr(val)	val: int=1	int	减少计数

from hagworm.extend.asyncio.redis import RedisPool, RedisDelegate, ShareCache

# 直接使用
redis = await RedisPool("localhost", 6379, max_connections=32).open()
await redis.set("key", "value")
val = await redis.get("key")

# 使用 Delegate
class MyService(RedisDelegate):
    async def init(self):
        await self.init_redis_single("localhost", 6379)

    async def get_data(self, key):
        return await self.redis_pool.get(key)

# 共享缓存
async with ShareCache(redis, "expensive_result", lock_timeout=60) as cache:
    result = await cache.get()
    if result is None:
        result = await expensive_computation()
        await cache.set(result, expire=300)

---

hagworm.extend.asyncio.mysql — MySQL 连接管理

MySQLPool

MySQL 连接池管理。

入参	类型	说明
host	str	主机
port	int	端口
db	str	数据库名
user	str	用户名
password	str	密码
name	str	连接池名称（默认自动生成；name 之后参数均为关键字限定）
minsize	int	最小连接数（默认 8）
maxsize	int	最大连接数（默认 32）
echo	bool	是否回显 SQL（默认 False）
pool_recycle	int	连接回收时间（秒，默认 21600）
charset	str	字符集（默认 utf8）
autocommit	bool	是否自动提交（默认 True）
cursor_class	cursors.Cursor	游标类型（默认 DictCursor）
readonly	bool	是否只读（默认 False）

方法	入参	返回值	说明
open()	无	None	初始化连接池
close()	无	None	关闭连接池
acquire()	无	Connection	获取连接
release(connection)	connection: Connection	None	归还连接
get_client()	无	MySQLClient	获取客户端
get_transaction()	无	MySQLTransaction	获取事务对象

MySQLClient

MySQL 客户端，支持 SQLAlchemy 查询构建器。

方法	入参	返回值	说明
execute(query, args)	query: str, args=None	Cursor	执行 SQL
select(query)	query: Select	List	查询列表
find(query)	query: Select	Any	查询单条
count(column, where_clause)	column, where_clause=None	int	计数
insert(query)	query: Insert	int	插入，返回 insert_id
update(query)	query: Update	int	更新，返回影响行数
delete(query)	query: Delete	int	删除，返回影响行数
release()	无	None	释放连接
readonly (属性)	无	bool	是否只读
insert_id (属性)	无	int	最近一次插入的自增 ID

MySQLTransaction

MySQL 事务对象，未显式 commit 则自动 rollback。

方法	入参	返回值	说明
commit()	无	None	提交事务
rollback()	无	None	回滚事务

MySQLDelegate

MySQL 功能组件。

方法	入参	返回值	说明
init_mysql_rw(...)	同 MySQLPool 构造参数	None	初始化读写连接池
init_mysql_ro(...)	同 MySQLPool 构造参数	None	初始化只读连接池
close_mysql()	无	None	关闭连接池
get_mysql_client(readonly)	readonly: bool=False	MySQLClient	获取客户端
get_mysql_transaction()	无	MySQLTransaction	获取事务对象

属性 mysql_rw_pool / mysql_ro_pool 分别为读写与只读连接池。

from hagworm.extend.asyncio.mysql import MySQLPool, MySQLClient, MySQLDelegate
import sqlalchemy as sa

# 直接使用
pool = MySQLPool("localhost", 3306, "mydb", "root", "password", minsize=8, maxsize=32)
await pool.open()

async with pool.get_client() as client:
    users = await client.select(sa.select(sa.table("users")))
    user = await client.find(sa.select(sa.table("users")).where(sa.column("id") == 1))

# 事务
async with pool.get_transaction() as trx:
    await trx.insert(sa.insert(sa.table("users")).values(name="Alice"))
    await trx.insert(sa.insert(sa.table("logs")).values(action="create"))
    await trx.commit()  # 不调用则自动 rollback

# 使用 Delegate
class MyService(MySQLDelegate):
    async def init(self):
        await self.init_mysql_rw("localhost", 3306, "mydb", "root", "password")
        await self.init_mysql_ro("localhost", 3307, "mydb", "reader", "password")

⚠️ 易混淆点：readonly=True 的连接池调用 insert/update/delete 或 get_transaction() 都会抛出 MySQLReadOnlyError。

属性：name（连接池名称）、readonly、cursor_class。

模块级常量与异常：

名称	说明
MYSQL_POOL_WATER_LEVEL_WARNING_LINE	连接池水位告警线（0x10）
MYSQL_EXECUTE_ARGUMENTS_TYPE	SQL 执行参数类型别名 Union[None, Tuple, List, Dict]
MySQLReadOnlyError	只读连接池执行写操作时抛出，继承 errors.MySQLError

---

hagworm.extend.asyncio.mongo — MongoDB 连接管理

MongoPool

MongoDB 连接池管理。

MongoPool(host: List[str], username: str, password: str, *, name=None, auth_source='admin', min_pool_size=8, max_pool_size=32, max_idle_time=3600, wait_queue_timeout=10, compressors='zlib', zlib_compression_level=6, **settings)

下列方法均为同步方法（无需 await）：

方法	入参	返回值	说明
close()（同步）	无	None	关闭连接池
get_database(db_name)（同步）	db_name: str	AsyncIOMotorDatabase	获取数据库

from hagworm.extend.asyncio.mongo import MongoPool, MongoDelegate

pool = MongoPool(
    host=["mongo1:27017", "mongo2:27017"],
    username="admin", password="secret",
    min_pool_size=8, max_pool_size=32
)

db = pool.get_database("mydb")
collection = db["users"]
await collection.insert_one({"name": "Alice"})

MongoDelegate

Mongo 功能组件。

方法	入参	返回值	说明
init_mongo(...)（同步）	同 MongoPool 构造参数	None	初始化连接池
close_mongo_pool()（同步）	无	None	关闭连接池
get_mongo_database(db_name)（同步）	db_name: str	AsyncIOMotorDatabase	获取数据库
get_mongo_collection(db_name, collection)（同步）	db_name: str, collection: str	AsyncIOMotorCollection	获取集合

属性 mongo_pool 为当前连接池。模块级常量 MONGO_POOL_WATER_LEVEL_WARNING_LINE（0x10）当前仅定义、未被使用。

---

hagworm.extend.asyncio.mail — SMTP 邮件

SMTPClient

异步 SMTP 邮件客户端。

SMTPClient(username: str, password: str, hostname: str, port: int, retry_count: int = 5, **kwargs)

方法	入参	返回值	说明
send_messages(email_body_list)	email_body_list: List[EmailBody]	Optional[Dict]	批量发送邮件
send_message(sender, recipients, message)	sender: str, recipients: Sequence[str], message: EmailMessage	Optional[Dict]	发送单封邮件
send(sender, recipients, subject, content)	均为 str 等	Optional[Dict]	简单发送
format_address(nickname, mailbox)	nickname: str, mailbox: str	str	静态方法，格式化为 nickname<mailbox>
format_addresses(addresses)	addresses: List[Tuple[str, str]]	str	格式化多个地址

EmailBody

EmailBody(sender: str, recipients: Sequence[str], message: EmailMessage)，提供 sender / recipients / message 属性，静态方法 create_message(sender, recipients, subject, content, content_type='text/html') 用于构建消息。

from hagworm.extend.asyncio.mail import SMTPClient, EmailBody

client = SMTPClient("user@gmail.com", "password", "smtp.gmail.com", 587)

# 简单发送
await client.send(
    sender="sender@example.com",
    recipients=["receiver@example.com"],
    subject="Hello",
    content="<h1>Hi there!</h1>"
)

# 带附件或自定义消息
message = EmailBody.create_message(
    sender="sender@example.com",
    recipients=["receiver@example.com"],
    subject="Report",
    content="<p>Please find the report attached.</p>"
)
await client.send_message("sender@example.com", ["receiver@example.com"], message)

---

hagworm.extend.asyncio.socket — TCP/Unix Socket

AsyncTcpServer

异步 TCP 服务器，实现 RunnableInterface 与 async with 协议。关闭兜底统一走 SignalManager（服务运行期间登记 _exit，收到信号或解释器退出时都会关闭全部 server），不再单独注册 atexit 钩子。

AsyncTcpServer(client_connected_cb: Callable, address: Any, *, family: int = socket.AF_INET, backlog: Optional[int] = None, reuse_port: bool = True, buffer_limit: int = BUFFER_LIMIT, on_startup: Callable = None, on_shutdown: Callable = None)

from hagworm.extend.asyncio.socket import AsyncTcpServer

async def handle_connection(reader, writer):
    while True:
        data = await reader.readline()
        if not data:
            break
        writer.write(data)
        await writer.drain()
    writer.close()
    await writer.wait_closed()

server = AsyncTcpServer(handle_connection, ("0.0.0.0", 8080))
server.run()

AsyncTcpServer.add_listener(client_connected_cb, address, *, family, backlog, reuse_port, buffer_limit) 可追加监听地址（支持 socket.AF_UNIX）。

on_startup / on_shutdown 为无参协程，分别在建立监听之前、退出时执行；run(*, debug=None) 会通过 SignalManager 监听 SIGINT/SIGTERM，收到信号后关闭全部 server。内部 _run() 用 try/finally 收尾，因此正常关闭、信号中断、任务取消等任何路径下都会注销该回调并执行 on_shutdown。

RobustConnection

自动重连的 TCP 客户端连接。

RobustConnection(host: Optional[str] = None, port: Optional[str] = None, *, limit: int = BUFFER_LIMIT, recycle: int = 86400, **kwargs)，属性 reader / writer 为底层 StreamReader / StreamWriter。

方法	入参	返回值	说明
open()	无	None	建立连接
close()	无	None	关闭连接
reset()	无	None	关闭后重连
readline()	无	bytes	读取一行
readuntil(separator)	separator: bytes=b'\n'	bytes	读取到分隔符
read(n)	n: int=-1	bytes	读取 n 字节
write(data)	data: bytes	None	写入数据
writelines(data)	data: Iterable[bytes]	None	写入多行数据

说明：open() 失败只记日志、不抛异常，需以 writer is None 判断是否连接成功；每次读写前会 _check_recycle()，未连接或超过 recycle 周期（默认 86400 秒）先 reset() 重连，重连后仍无 writer 则抛 ConnectionError；读写失败时会先 reset() 再重新抛出原异常。

UnixSocketServer / UnixSocketClient

Unix Domain Socket 服务端与客户端。默认使用 msgpack 编解码（可在 StreamFramer 实例化时注入自定义编/解码器），线协议为 MAGIC(8) + length(4) + body。

类	构造签名	说明
UnixSocketServer	(client_connected_cb: Callable, path: str, limit: int = BUFFER_LIMIT)	open() 会先删除已存在的 path，再 start_unix_server 并 start_serving()
UnixSocketClient	(path: str, limit: int = BUFFER_LIMIT)	属性 reader / writer；内部固定使用默认 StreamFramer()，如需自定义编解码器请直接使用 StreamFramer

limit 是 asyncio stream 缓冲区上限，并非消息大小上限。

UnixSocketClient 方法：

方法	入参	返回值	说明
open()	无	None	建立连接
close()	无	None	关闭连接
recv_msg(timeout)	timeout: float=None	Any	接收一条消息
send_msg(data)	data: Any	None	发送一条消息

StreamFramer

实例类，封装 recv_msg / send_msg，线协议为 MAGIC(8) + length(4, 大端) + body。

方法	入参	返回值	说明
StreamFramer(encoder, decoder)	encoder: Callable=msgpack.dumps, decoder: Callable=msgpack.loads	StreamFramer	编/解码器均为单参数可调用对象，可注入自定义实现
recv_msg(reader, timeout)	reader: StreamReader, timeout: float=None	Any	接收一条消息
send_msg(writer, data)	writer: StreamWriter, data: Any	None	发送一条消息

模块级常量与类型：

名称	说明
MESSAGE_LIMIT	单条消息 body 最大字节数（1 MiB）
BUFFER_LIMIT	asyncio stream 缓冲区上限（2 * MESSAGE_LIMIT）
MAGIC	帧头魔数（8 字节）
DEFAULT_UNIX_SOCKET_ENDPOINT	默认 Unix Domain Socket 路径
SocketConfig	NamedTuple，描述单个监听配置（client_connected_cb/address/family/backlog/reuse_port/buffer_limit）

---

hagworm.extend.asyncio.command — 多进程命令框架

MainProcess / SubProcess

异步多进程框架，主进程管理子进程的生命周期。

MainProcess(target: Callable, sub_process_num: int, *, cpu_affinity: bool = False, join_timeout: int = 10, keep_alive: bool = False, **kwargs)，继承 hagworm.extend.process.Daemon，沿用同一套 target(idx, **kwargs) 调用约定与 keep_alive 语义。run() 内部安装 uvloop、通过 SignalManager 注册信号，并依次执行 initialize() → _execute() → release()，结束时注销该信号回调。

SubProcess(process_num: int, *args, **kwargs) 实现 RunnableInterface，process_num 即主进程分配的槽位序号（Daemon 以 args=(idx,) 传入）：

方法	入参	返回值	说明
create(*args, **kwargs)	同构造函数	None	类方法，等价于 cls(*args, **kwargs).run()，常直接作为主进程的 target
close(*_)	无	None	唤醒阻塞协程并关闭事件循环
initialize()	无	None	子进程启动钩子（可覆盖）
release()	无	None	子进程退出钩子（可覆盖）
_execute()	无	None	子进程主体逻辑，默认 await WaitForever() 常驻（可覆盖）
run()	无	None	安装 uvloop、通过 SignalManager 注册信号并运行上述流程，结束时注销

from hagworm.extend.asyncio.command import MainProcess, SubProcess

class MyWorker(SubProcess):
    async def initialize(self):
        print(f"Worker {self._process_id} initializing...")

    async def _execute(self):
        # 业务逻辑
        pass

    async def release(self):
        print(f"Worker {self._process_id} releasing...")

class MyMain(MainProcess):
    async def initialize(self):
        print("Main process initializing...")

    async def release(self):
        print("Main process releasing...")

if __name__ == "__main__":
    MyMain(MyWorker.create, sub_process_num=4).run()

MainProcessWithIPC / SubProcessWithIPC

基于 Unix Domain Socket 的进程间通信版主/子进程，主进程通过 _client_connected_cb 处理子进程连接。

类	构造签名	说明
MainProcessWithIPC	(target, sub_process_num, *, cpu_affinity=False, join_timeout=10, keep_alive=False, unix_socket_path=DEFAULT_UNIX_SOCKET_ENDPOINT, unix_socket_limit=BUFFER_LIMIT, **kwargs)	额外维护 UnixSocketServer；子类需实现抽象协程 _client_connected_cb(reader, writer)
SubProcessWithIPC	(process_num, unix_socket_path, unix_socket_limit=BUFFER_LIMIT)	unix_socket_path 为必填位置参数（process_num 由主进程以 args=(idx,) 传入）；缓冲区上限须与主进程侧一致，否则读包会死锁

unix_socket_path / unix_socket_limit 会经 **kwargs 自动透传给子进程入口，因此子类一般无需再手工传递。

---

hagworm.extend.asyncio.process — 子进程托管

Worker

基于 multiprocessing 的异步子进程托管器，实现 TaskInterface。固定使用 multiprocessing.get_context('spawn') 启动（PROCESS_CONTEXT），不依赖 hagworm.extend.process 的 set_start_method 副作用，也避免 fork 复制父进程的事件循环与连接状态；因此 target 及其关键字参数必须可 pickle（模块级函数、类方法等），lambda 或局部函数会启动失败，入口脚本需保留 if __name__ == "__main__" 保护。

Worker 与子进程一一绑定：一个实例只启动一个子进程，启动后不可复用，子进程退出即终止。

多个实例各自注册信号回调，框架内部由 SignalManager 统一多播，因此主进程收到信号时多类子进程都会被强杀。

入参	类型	默认值	说明
name	str	必填	子进程名称，用于日志区分不同类型的子进程
target	Callable	必填	子进程入口函数，需可 pickle
**kwargs	Any	无	透传给 target 的关键字参数（以 Process(target=target, kwargs=kwargs) 启动，不含位置参数）

方法	入参	返回值	说明
is_running()	无	bool	子进程是否存活（同步方法）
start()	无	bool	启动子进程；每个实例只能启动一次，已启动过返回 False（协程）
stop()	无	None	向子进程发送 SIGINT 并按 CHECK_INTERVAL 轮询等待其退出，超过 STOP_TIMEOUT 未退出则 kill()，最后 join()（协程）

模块级常量：

名称	说明
CHECK_INTERVAL	stop() 等待子进程退出的轮询间隔（秒），默认 0.05
STOP_TIMEOUT	stop() 等待子进程优雅退出的超时时间（秒），默认 10，超时强杀
PROCESS_CONTEXT	子进程启动上下文，固定为 multiprocessing.get_context('spawn')

行为说明：

• name 会随启动、停止、信号日志一并打印，用于区分不同类型的子进程。
• start() 在子进程启动成功后通过 SignalManager.register 注册 SIGINT/SIGTERM 回调，收到信号后直接强杀子进程；框架内由 SignalManager 多播，多实例不会互相覆盖。
• stop() 先 os.kill(pid, signal.SIGINT) 通知子进程退出（在子进程中表现为 KeyboardInterrupt），再轮询至退出；若超过 STOP_TIMEOUT 仍未退出则 process.kill() 强杀，避免子进程忽略信号时无限等待；未启动或已退出时直接返回。
• 兜底与释放：构造时 atexit.register(self._kill)，解释器退出时若子进程仍存活则强制 kill() 并 join()；stop() 完成后会通过 SignalManager.unregister 注销信号回调并 atexit.unregister(self._kill)，避免长期持有实例引用。
• 启动失败（如 target 不可 pickle）时不会记录进程对象，start() 可再次调用重试。

import asyncio

from hagworm.extend.asyncio.process import Worker

# 子进程入口必须是模块级函数（可 pickle）
def serve(port: int):
    # 子进程内的业务逻辑，通常阻塞运行
    ...

async def main():
    worker = Worker("http-server", serve, port=8080)

    if await worker.start():
        print("worker started")

    while worker.is_running():
        await asyncio.sleep(1)

    await worker.stop()

if __name__ == "__main__":
    asyncio.run(main())

---

hagworm.extend.asyncio.transaction — 异步事务

Transaction（异步版）

异步事务对象，支持同步和异步回调。

构造签名 Transaction(*, commit_callback=None, rollback_callback=None)；提供 add_commit_callback(_callable, *args, **kwargs)、add_rollback_callback(_callable, *args, **kwargs)、commit()、rollback() 与 bind(trx) 方法，语义与同步事务一致（异步回调会被 await）。

from hagworm.extend.asyncio.transaction import Transaction

async with Transaction() as trx:
    trx.add_commit_callback(lambda: print("committed"))
    trx.add_commit_callback(async_cleanup)
    await trx.commit()

---

hagworm.extend.asyncio.file — 文件加载器

FileLoader

带缓存的网络/本地文件加载器。

FileLoader(maxsize=0xff, ttl=3600, thread=32)；read(file) 读取本地文件，fetch(url, params=None, *, cookies=None, headers=None) 拉取网络文件。

from hagworm.extend.asyncio.file import FileLoader

loader = FileLoader(maxsize=100, ttl=3600, thread=32)
content = await loader.read("/path/to/file")
html = await loader.fetch("https://example.com/page.html")

---

hagworm.extend.asyncio.pool — 通用对象池

ObjectPool

通用异步对象池，需子类实现 _create() 方法；ObjectInterface 为其管理的对象基类，需实现抽象方法 open(*args, **kwargs) 与 close(*args, **kwargs)。

提供 size 属性（池容量）及 open() / close() / get() 方法（均可透传 *args / **kwargs）。

from hagworm.extend.asyncio.pool import ObjectPool, ObjectInterface

class MyConnection(ObjectInterface):
    async def open(self):
        self.conn = await create_connection()

    async def close(self):
        await self.conn.close()

class MyConnectionPool(ObjectPool):
    def _create(self):
        return MyConnection()

pool = MyConnectionPool(maxsize=10)
await pool.open()

async with pool.get() as conn:
    # 使用连接
    pass

await pool.close()

---

hagworm.frame.fastapi — FastAPI 集成

create_fastapi

创建 FastAPI 应用，自动配置日志、uvloop、中间件和异常处理。

create_fastapi(log_level='info', log_handler=None, log_file_path=None, log_file_name=DEFAULT_LOG_FILE_NAME, log_file_rotation=DEFAULT_LOG_FILE_ROTATOR, log_file_retention=0xff, log_extra=None, log_enqueue=False, debug=False, routes=None, **setting) -> fastapi.FastAPI

内部依次执行 init_logger → Utils.print_slogan() → install_uvloop，然后构造 fastapi.FastAPI 并注册 http 中间件 http_default_middleware，再注册 BusinessException / HTTPException / RequestValidationError 三个异常处理器（business_exception_handler / http_exception_handler / request_validation_exception_handler），最后挂载 /static 静态目录；**setting 原样透传给 fastapi.FastAPI。

from hagworm.frame.fastapi.base import create_fastapi, uvicorn_run

app = create_fastapi(
    log_level="INFO",
    log_file_path="/var/log/app",
    debug=False,
)

if __name__ == "__main__":
    uvicorn_run(lambda: app, port=8080)

uvicorn_run

uvicorn_run(app, host='0.0.0.0', port=8080)：开发环境启动入口（生产环境建议使用 gunicorn）。

BusinessException

BusinessException(code, content=None)：业务异常，会被异常处理器转为 ErrorResponse（code 会透传为响应中的 code）。

http_default_middleware

默认 HTTP 中间件，注入 trace_id 并写入 server / x-trace-id / x-request-ttl-ms / x-request-payload 响应头。

APIRouter

支持末尾斜杠兼容的路由器。

Request

增强的请求对象，提供 client_ip、referer 等便捷属性。

Response / ErrorResponse

统一响应格式：Response 为 {"code": 0, "data": ..., "trace_id": ...}；ErrorResponse 为 {"code": <构造传入的 code>, "error": ..., "trace_id": ...}（内置 HTTP/校验异常处理器传 -1，业务异常处理器透传 BusinessException.code）。

BaseModel / Depends

增强的 Pydantic 模型（元类 ModelMetaclass），支持 params 方法生成 FastAPI 依赖签名。

Depends(dependency=None, *, use_cache=True)：dependency 为 BaseModel 子类时会被转换为该模型的 params 依赖。

field — 自定义字段类型

提供 IDCardType、BankCardType、UUIDType、DateType、JsonType、Base64Type、StrListType、IntListType、FloatListType、ASCVisibleType、EmailType、DomainType、URLType、MacAddrType、IPvAnyType、IPv4Type、IPv4CidrType、IPv6Type、IPv6CidrType 等验证字段。

from hagworm.frame.fastapi.base import APIRouter, Request, BusinessException
from hagworm.frame.fastapi.field import EmailType, IntListType
from hagworm.frame.fastapi.model import BaseModel, Depends

router = APIRouter()

class QueryModel(BaseModel):
    email: EmailType
    ids: IntListType

@router.get("/users")
async def get_users(request: Request):
    return request.client_ip

@router.get("/error")
async def error(request: Request):
    raise BusinessException(-1, "something went wrong")  # 由异常处理器转为 ErrorResponse

@router.get("/search")
async def search(params: QueryModel = Depends(QueryModel)):
    # params.email 已验证为合法邮箱
    # params.ids 已解析为整数列表
    return {"email": params.email, "ids": params.ids}

---

hagworm.frame.gunicorn — Gunicorn Worker

Worker

基于 uvicorn-worker 的 Gunicorn Worker 类（继承 uvicorn_worker.UvicornWorker，构造时强制 factory=True、server_header=False）。

因 factory=True，gunicorn 的 app 目标必须是工厂函数，例如：

gunicorn main:create_app -c gunicorn.config.py

# gunicorn.conf.py
from hagworm.frame.gunicorn import DEFAULT_WORKER_STR, SIMPLE_LOG_CONFIG

worker_class = DEFAULT_WORKER_STR  # 等价于 "hagworm.frame.gunicorn.Worker"
logconfig_dict = SIMPLE_LOG_CONFIG
workers = 4
bind = "0.0.0.0:8080"

模块级常量：DEFAULT_WORKER_STR（'hagworm.frame.gunicorn.Worker'）、SIMPLE_LOG_CONFIG（仅含 gunicorn.error）与 DEFAULT_LOG_CONFIG（含 gunicorn.error / gunicorn.access），均将标准库日志经 InterceptHandler 转发到 loguru。

---

hagworm.frame.stress_tests — 压力测试框架

Launcher / RunnerAbstract

多进程压力测试框架，自动收集样本并生成 Markdown 报表（含统计表格与 Mermaid 图表）。

报表包含：概览（同时存在成功与失败样本时才含请求结果占比饼图）、SLA 达标判定、吞吐/延迟统计表、时间趋势（总吞吐与成功吞吐 goodput、失败率、P50/P95/P99 耗时）、成功样本延迟分布、失败原因分布（全局与按事件拆分，含失败原因占比饼图）、失败样本延迟分布（show_failure_dist=True 时输出）。

样本在收集时即做流式聚合（定长对数直方图 + 计数器），不保留原始样本，内存占用与样本量无关；分位为 0.1% 相对精度估算，均值/标准差采用在线 Welford 算法。

统计口径：Qps 按各事件自身的「首末样本时间差」计算（ALL 行为全局窗口）；时间轴为主进程收包时刻（monotonic），反映样本到达节奏；耗时分位/分布基于 0.1% 相对精度直方图估算，用户自定义分桶为精确计数。

模块同时暴露 LogHistogram（定长对数直方图，epsilon=HISTOGRAM_EPSILON、base=HISTOGRAM_BASE，含 count/minimum/maximum/mean/std 属性与 record/merge/quantile/bucket_counts 方法）、EventAggregator(keep_failure_latency=False)、TrendSeries(max_buckets=TREND_MAX_BUCKETS, histogram_epsilon=TREND_HISTOGRAM_EPSILON) 与 TrendSeriesResult。

• Launcher(target, sub_process_num, *, ..., **kwargs)：target 为子进程执行体（通常传 RunnerAbstract.create）、sub_process_num 为子进程数，两者为必填位置参数；**kwargs 透传给 MainProcessWithIPC（如 cpu_affinity / join_timeout / keep_alive）。
• RunnerAbstract.success(name, resp_time) / RunnerAbstract.failure(name, resp_time, error=None)：在子进程内上报一次成功/失败请求。resp_time 为请求耗时（单位见 time_unit），error 为失败原因（可选，空值归入 unknown）。
• Launcher 构造参数：
  • report_dir：报表输出目录，默认 ./，报表文件名为 stress_report_年月日_时分秒.md。
  • time_unit：延迟指标单位（仅用于展示，默认 ms）。
  • shard_table_bins：自定义延迟分桶边界（须严格递增），不传则按样本分位数自适应生成。
  • percentiles：统计分位数，默认 (0.5, 0.9, 0.95, 0.99, 0.999)。
  • show_std：是否展示标准差列（默认关闭，长尾延迟下分位数更直观）。
  • show_failure_dist：是否额外输出失败样本延迟分布表（默认关闭）。
  • sla：达标判定配置，如 {'p95': 200, 'p99': 500, 'failure': 0.01}，pxx 阈值单位与 time_unit 一致，failure 为失败率阈值(0~1)。
  • sla_scope：SLA 判定范围，'global'（仅全局）/ 'event'（仅按事件）/ 'both'（默认，全局表 + 按事件矩阵，按未达标数降序并附柱状图）。
  • unstable_threshold：每秒吞吐波动系数(标准差/均值)超过该值判定为不稳定（默认 0.2）。
  • unix_socket_path：主/子进程 IPC 的 Unix Domain Socket 路径，默认 /tmp/unix_socket_endpoint，须与子进程侧一致。
  • unix_socket_limit：IPC 流缓冲区上限（字节），默认 BUFFER_LIMIT（2 MiB），须与子进程侧一致。

import time

from hagworm.extend.asyncio.base import Utils
from hagworm.frame.stress_tests import Launcher, RunnerAbstract

class MyRunner(RunnerAbstract):
    async def _execute(self):
        for _ in range(100):
            _begin = time.monotonic()
            await Utils.sleep(Utils.randint(10, 99) / 1000)
            _cost = (time.monotonic() - _begin) * 1000

            if Utils.rand_hit([True, False], [80, 20]):
                await self.success("API", _cost)
            else:
                await self.failure("API", _cost, error="timeout")

if __name__ == "__main__":
    Launcher(MyRunner.create, sub_process_num=4).run()

---

hagworm.third.grpc — gRPC 框架

GRPCServer

grpc.aio.server 的服务端封装。属性 server；方法 bind_router(router)、start(address, *, family=socket.AF_INET, server_credentials=None, enable_ping=False)（支持 AF_INET / AF_INET6 / AF_UNIX，默认开启 SO_REUSEPORT）、stop(grace=None)、wait(timeout=None)。

默认 msgpack 序列化由 Router 的 request_deserializer=msgpack.loads / response_serializer=msgpack.dumps（以及 GRPCClient 的对称默认值）提供；网桥转发时为原始字节透传。

Router

gRPC 路由器，支持装饰器注册方法：unary_unary / unary_stream / stream_unary / stream_stream，另提供 register / include / get_generic_handlers。

StreamHandler

stream_stream 场景的服务端处理基类：StreamHandler(request, context, *, metadata_model=None)，子类实现 on_message(data) / on_connect() / on_close(context)；提供 join(*, timeout=0)、read()、write(message)、cancel()、close()，以及方法 metadata() / metadata_model() 与属性 peer。

模块级常量：UNARY_UNARY_MODE / UNARY_STREAM_MODE / STREAM_UNARY_MODE / STREAM_STREAM_MODE 与 GRPC_MODE_TYPE；RouteFunction 为描述单个路由的 NamedTuple。

GRPCClient

gRPC 客户端，支持 RoundRobin 负载均衡。

StreamClient / RobustStreamClient

基于 stream_stream 的客户端封装；RobustStreamClient 额外支持消息回调（on_connect / on_message）。

GRPCEntry

worker 侧入口，将 router 注册到 bridge。

GRPCBridge

gRPC 网桥，实现请求转发（转发时 request_serializer / response_deserializer 均为 None，即原始字节透传）。

配套：copy_metadata(remote_context) 复制对端调用元数据并补充 remote-peer；BridgeUnaryStream / BridgeStreamUnary / BridgeStreamStream 为三种流转发器，均提供 join()（等待流转结束）。

GRPCMainProcess / GRPCWorker

gRPC 多进程框架。

from hagworm.third.grpc.server import Router, StreamHandler
from hagworm.third.grpc.bridge import GRPCMainProcess, GRPCWorker
from hagworm.third.grpc.client import GRPCClient

# 服务端
router = Router("demo")

@router.unary_unary
async def echo(request, context):
    return {"echo": request}


class EchoHandler(StreamHandler):

    async def on_message(self, data):
        await self.write(data)

    async def on_connect(self):
        pass

    def on_close(self, context):
        pass


@router.stream_stream
async def stream_echo(request, context):
    await EchoHandler(request, context).join()

if __name__ == "__main__":
    GRPCMainProcess(("0.0.0.0", 8080), [router], GRPCWorker, 2).run()

# 客户端
client = GRPCClient()
await client.open(["127.0.0.1:8080"])
result = await client.unary_unary("/demo/echo", {"msg": "hello"})

---

hagworm.third.rabbitmq — RabbitMQ 消息队列

RabbitMQConsumer / RabbitMQConsumerForExchange

RabbitMQ 消费者。RabbitMQConsumer(connection, queue_name)；RabbitMQConsumerForExchange(connection, queue_name, exchange_name)。

方法	说明
open(*, consume_func=None, consume_no_ack=False, channel_qos_config=None, queue_config=None[, routing_key=None])	打开消费者（ForExchange 版额外支持 routing_key）
block_pull(consume_func, consume_no_ack=False, wait_time=1)	阻塞式拉取消费
close(with_connection=False)	关闭消费者（with_connection=True 时同时关闭连接）

RabbitMQProducer / RabbitMQProducerForExchange

RabbitMQ 生产者。RabbitMQProducer(connection)；RabbitMQProducerForExchange(connection, exchange_name)。

方法	说明
open(*, channel_number=None, publisher_confirms=True, on_return_raises=False)	打开生产者
open(*, exchange_type=aio_pika.ExchangeType.FANOUT, exchange_config=None, channel_number=None, publisher_confirms=True, on_return_raises=False)	打开交换机生产者（ForExchange 版）
publish(message, routing_key='', **kwargs)	发布消息
close(with_connection=False)	关闭生产者

RabbitMQProducerPool / RabbitMQProducerForExchangePool

RabbitMQ 生产者连接池。RabbitMQProducerPool(pool_size, connection)；RabbitMQProducerForExchangePool(pool_size, connection, exchange_name)；提供 open() / publish(message, routing_key='', **kwargs) / close()。

RpcServer / RpcClient

基于 RabbitMQ 的 RPC 实现，支持 Unary 和 Stream 两种模式。

RpcServer(pool_size, connection, exchange_name, *, server_name=None, message_ttl=60, max_priority=10)：open() / close() 管理生命周期（内置 ping），register(name, func) 注册被调用函数。

RpcClient(pool_size, connection, exchange_name, message_ttl=60)：open() / close()；ping(priority=0, server_name=None)；call(func_name, *, func_args=None, func_kwargs=None, priority=0, server_name=None, request_timeout=60)；call_stream(..., request_timeout=3600)。

配套：RpcServerError（服务端业务异常）、FunctionType（Unary / Stream，标识函数模式）、get_routing_key(server_name)（生成路由键），模块常量 STREAM_EOF / ERROR_HEAD（流结束与错误帧标记）。

create_connection

create_connection(url, timeout=None, **kwargs) -> aio_pika.RobustConnection：创建 RobustConnection 并内部 asyncio.create_task(connection.connect(timeout)) 自动发起连接；如需等待连接完成，仍可再 await connection.connect()。

from hagworm.third.rabbitmq import create_connection
from hagworm.third.rabbitmq.publish import RabbitMQProducerForExchange
from hagworm.third.rabbitmq.consume import RabbitMQConsumerForExchange
import aio_pika

# 创建连接
connection = create_connection("amqp://guest:guest@localhost/")  # 内部已自动发起连接

# 生产者
producer = RabbitMQProducerForExchange(connection, "my_exchange")
await producer.open(exchange_type=aio_pika.ExchangeType.TOPIC)
await producer.publish(b"hello", "routing_key")

# 消费者
async def handle_message(message: aio_pika.IncomingMessage):
    print(message.body)
    await message.ack()

consumer = RabbitMQConsumerForExchange(connection, "my_queue", "my_exchange")
await consumer.open(consume_func=handle_message, routing_key="routing_key")

---

hagworm.third.nacos — Nacos 服务发现

NacosConfig

Nacos 配置中心客户端，支持长轮询监听配置变更。

NacosConfig(listener, servers, data_id, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', request_timeout=..., pulling_timeout=...)

方法	说明
start() / stop() / is_running()	启动 / 停止 / 运行状态
get_config()	获取当前配置（(content, content_hash)）

配置变更回调在构造时以 listener 传入（获取到配置后回调），不提供 add_listener / clear_listeners 方法。

NacosInstanceRegister

Nacos 服务实例注册，自动发送心跳。

NacosInstanceRegister(servers, service_name, service_ip, service_port, heartbeat_interval=10, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', cluster=None, weight=None, request_timeout=...)

NacosInstanceQuery

Nacos 服务实例查询，监听实例变更。

NacosInstanceQuery(servers, service_name, listener_interval=10, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', cluster=None, request_timeout=...)

方法	说明
start() / stop() / is_running()	启动 / 停止 / 运行状态
get_host()	按 weight 加权随机返回单个实例 {'ip', 'port', 'weight', 'metadata'}
get_hosts()	返回全部实例列表
add_listener(callback) / clear_listeners()	追加 / 清空实例变更回调

Configure（Nacos 版）

从 Nacos 配置中心加载配置：Configure(servers, data_id, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', data_format='yaml')，提供 open() / close()。

data_format 支持 text / json / yaml（默认 yaml）；当 servers 为空且 data_id 为本地文件路径时，直接从该文件加载。

from hagworm.third.nacos.client import NacosConfig, NacosInstanceRegister, NacosInstanceQuery
from hagworm.third.nacos.config import Configure

# 配置中心
def on_config_change(content):
    print(f"Config updated: {content}")

nacos_config = NacosConfig(
    on_config_change, "localhost:8848", "my-app.yaml",
    group="DEFAULT_GROUP", namespace="dev"
)
nacos_config.start()

# 服务注册
register = NacosInstanceRegister(
    "localhost:8848", "my-service", "10.0.0.1", 8080,
    namespace="dev"
)
register.start()

# 服务发现
query = NacosInstanceQuery("localhost:8848", "my-service", namespace="dev")
instance = query.get_host()  # 按 weight 加权随机返回 {'ip','port','weight','metadata'}
hosts = query.get_hosts()    # 全部实例

---

完整示例

FastAPI 项目示例

参见 example/fastapi_demo/ 目录，包含完整的 FastAPI 项目结构：

• main.py — 应用入口
• setting.py — 配置定义
• controller/home.py — 路由控制器
• gunicorn.config.py — gunicorn 部署配置
• main.sh — 启动脚本（gunicorn main:create_app -c gunicorn.config.py）
• service/init.py — 数据源与业务基类
• model/init.py — 数据模型
• build.sh — PYC 编译打包脚本
• config/config.yaml — 配置文件

gRPC 示例

参见 example/grpc_demo/ 目录：

• server.py — gRPC 服务端
• client.py — gRPC 客户端

TCP 服务示例

参见 example/main_tcp.py：

import asyncio

from hagworm.extend.asyncio.base import Utils
from hagworm.extend.asyncio.socket import AsyncTcpServer

async def connection_handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:

    client_info = writer.transport.get_extra_info(r'socket').getpeername()
    Utils.log.info(f'create connection: {client_info}')

    try:
        while True:
            request = await reader.readline()
            if not request:
                break
            writer.write(request)
            await writer.drain()
    except Exception as err:
        Utils.log.error(err)
    finally:
        if not writer.is_closing():
            writer.close()
            await writer.wait_closed()

    Utils.log.info(f'lost connection: {client_info}')

def main() -> None:
    server = AsyncTcpServer(connection_handle, (r'0.0.0.0', 8080))
    server.run()

if __name__ == r'__main__':
    main()

RabbitMQ 示例

参见 example/main_rabbitmq.py：

import asyncio

import aio_pika

from hagworm.extend.asyncio.base import Utils, install_uvloop
from hagworm.third.rabbitmq import create_connection
from hagworm.third.rabbitmq.publish import RabbitMQProducerForExchange
from hagworm.third.rabbitmq.consume import RabbitMQConsumerForExchange


async def main() -> None:

    connection = create_connection(r'')
    await connection.connect()  # create_connection 内部已自动发起连接，此处可再显式等待

    await asyncio.gather(publish_run(connection), consume_run_0(connection), consume_run_1(connection))


async def publish_run(connection: aio_pika.RobustConnection) -> None:

    producer = RabbitMQProducerForExchange(connection, r'test1983')
    await producer.open(exchange_type=aio_pika.ExchangeType.TOPIC)

    await asyncio.sleep(5)

    for _idx in range(100):
        await producer.publish('content_{}'.format(_idx).encode(), 'test_{}'.format(_idx % 3))

    await asyncio.sleep(5)

    await connection.close()


async def consume_run_0(connection: aio_pika.RobustConnection) -> None:

    async def _consume_func(message: aio_pika.abc.AbstractIncomingMessage) -> None:
        print(r'test_0', message.body)
        await message.ack()

    consumer = RabbitMQConsumerForExchange(connection, r'test0', exchange_name='test1983')
    await consumer.open(routing_key=r'test_0')
    await consumer.block_pull(_consume_func)


async def consume_run_1(connection: aio_pika.RobustConnection) -> None:

    async def _consume_func(message: aio_pika.abc.AbstractIncomingMessage) -> None:
        print(r'test_1', message.body)
        await message.ack()

    consumer = RabbitMQConsumerForExchange(connection, r'test1', exchange_name='test1983')
    await consumer.open(routing_key=[r'test_1', r'test_2'])
    await consumer.block_pull(_consume_func)


if __name__ == r'__main__':

    Utils.print_slogan()

    install_uvloop()

    asyncio.run(main())

压力测试示例

参见 example/main_test.py：

from hagworm.extend.base import TimerMS
from hagworm.extend.asyncio.base import Utils
from hagworm.frame.stress_tests import Launcher, RunnerAbstract

class Runner(RunnerAbstract):
    async def _execute(self) -> None:
        for index in range(5):

            for _ in range(5):

                with TimerMS(f"Test{index}") as timer:
                    await Utils.sleep(Utils.randint(10, 99) / 1000)

                if Utils.rand_hit([True, False], [50, 50]):
                    await self.success(f"Test{index}", timer.cost)
                else:
                    await self.failure(f"Test{index}", timer.cost)

if __name__ == "__main__":
    Launcher(Runner.create, 5).run()

---

单元测试

运行测试：

cd testing
pytest

testing/setup.cfg 提供 asyncio_mode=auto 的 pytest 配置（异步用例依赖该配置）；也可直接运行 testing/main.py 启动 pytest.main()。

测试覆盖模块：

• test_extend_base.py — 基础工具、异常处理、上下文管理器、FuncWrapper
• test_extend_cache.py — StackCache 缓存
• test_extend_struct.py — KeyLowerDict 数据结构
• test_extend_asyncio_base.py — 异步工具、ShareFuture、FuncCache、robust_iterator、事务
• test_extend_asyncio_future.py — ThreadWorker 线程转协程
• test_extend_asyncio_task.py — MultiTasks 并发、RateLimiter 限流
• test_extend_asyncio_net.py — HTTP 客户端

---

许可证

Apache-2.0