python-library-callback 0.2.0

No project description provided

callback

特性

这个库做三件事，记清楚就够用了：

1. 你写一个 Callback 子类，类里的类型注解 = 这次调用要带哪些字段（像一张固定格式的表）。

2. 处理函数按 前（before）、中间、后（after） 三层登记；触发按 先整层前、再整层中间、再整层后 的顺序跑；同一层里多个函数仍然是一起跑、等这一层全部结束再进下一层。

3. 触发结束后的返回值就是这一次的那条实例——不是副本；外面继续用这个对象读字段，就是处理函数改完之后的样子。

推荐写法（少写 register / trigger）

元类把「对子类的类调用」收成两种常用形态，日常可以只写子类名：

写法	等价于	含义
@OrderPaid（装饰在函数上）	OrderPaid.register(函数)	向中间层登记
OrderPaid(order_id="a")	OrderPaid.trigger(order_id="a")	同步跑完三层管线，阻塞至全部回调结束，返回同一条已可能被回调改过的实例

仍可直接调用 register / register_before / register_after / trigger，与上表语义一致。

注意：「单参数、且为可调用对象、且无关键字参数」的类调用一律按装饰器登记处理，不会当数据去触发。若某次载荷里第一个字段就是要传的可调用对象，请改用关键字参数触发，例如 MyCb(fn=my_callable)，避免与 @MyCb 形态撞车。

子类定义「一次调用」的数据

触发时你传构造参数，库按子类字段校验、填默认值。多出来的参数名（类里没声明的）不能传进来。

哪些名字算数据字段

只有「普通类型注解、名字不以 _ 开头」的才算在这一次调用的数据里。

写法	算不算这次调用的字段
count: int	算
_tmp: int	不算（给「只想给类型检查看、不想当数据」用的）
foo: ClassVar[...]	不算（写在类上、不是每次调用一条记录上的字段）

字段类型可以是任意对象

字段不限于数字、字符串：任何 Python 对象都能放进这条实例里（比如一个已经建好的服务对象）。处理函数和调用方拿到的是同一个对象引用。

登记挂在子类私有 _layers 上

没有模块级全局注册表。每个 Callback 子类 在类定义结束时会挂上一份 _layers，类型是 CallbackLayers（不对外以公开属性名暴露，仅供库内部与扩展实现使用）。对外前、后两层请用类方法装饰器 @子类.before、@子类.after（其语义与 register_before / register_after 相同）；中间仍用 @子类 或 register。基类 Callback 不挂容器，只供继承。

CallbackLayers 内部对三层有 before / middle / after 的 LayerTier 引用；每一层都可以 LayerTier.register(函数) 或把 LayerTier 当装饰器用。子类之间各自一份 _layers，不会与父类共用。

处理函数三层：和装饰器 / register* 的对应关系

阶段	推荐装饰器	类方法
前	@类名.before	类名.register_before(fn)
中间	@类名（或 middle 的等价登记）	类名.register(fn)
后	@类名.after	类名.register_after(fn)

中间这一层和以前「直接 @类名」是同一回事。函数照样可以带这次的数据参数，也可以不带。

@OrderPaid.before
def prepare(cb: OrderPaid) -> None:
    ...

@OrderPaid
def bump_total(cb: OrderPaid) -> None:
    cb.total += 1

@OrderPaid.after
def notify(cb: OrderPaid) -> None:
    ...

# 与 OrderPaid.trigger(...) 相同：跑完三层后得到这一条实例
paid = OrderPaid(order_id="x", total=1)

同一个函数在同一层里注册多次，也只保留一份

同一个函数对象在同一层里你登记两遍，内部还是一条，这一层触发时只跑一次。两个不同的函数就各算各的。

（同一函数可以分别出现在不同层，那是三层各一条，各跑一次。）

触发（trigger 与 子类(...)）

• 同步入口：SomeCallback.trigger(...) 与 SomeCallback(...) 等价；在当前线程没有正在运行的事件循环时，内部用 asyncio.run 跑完整条异步管线并阻塞到结束。
• 已在事件循环里（例如在 async def 里直接调用）时，不能在本线程上阻塞式 trigger / 子类(...)；请从普通同步上下文触发，或放到另起线程里再调。
• 登记：同一子类上可以同时登记 def 与 async def。
• 层与层之间：严格按 前 → 中间 → 后；上一层的函数全部跑完，才进下一层。
• 同一层内：多个处理函数一起收尾（协程在本层并发；普通函数在线程里执行，避免卡住事件循环），整层结束后再进下一层。
• 返回值 = 这一次新建的那条子类实例（子类(...) 与 trigger 返回同一条引用语义）。
• 一层里一个函数都没挂，那一层相当于跳过；三层都空也会先建好实例再返回。

get_all、register* 与清空登记

• get_all：列出 Callback 的直接子类（儿子辈）。孙子类、更下面的子类不会出现在这个列表里。
• register / register_before / register_after 与 before / after 装饰器：同一套去重规则；装饰器在登记后仍返回原函数。
• 单测或进程里想一次清空所有子类已挂的函数：调用 Callback.clear_layer_registries()（会递归扫子类，对每个已在 __dict__ 里创建过的 _layers 调用 clear()）。