Middleware
onestep 1.x 使用事件钩子替代传统的中间件模式，提供更清晰的生命周期控制。
事件钩子
使用 onestep run 启动时，CLI 默认注册 StructuredEventLogger，应用无需重复配置。直接调用 app.run() 或 app.serve() 的嵌入式应用仍由宿主进程管理日志和事件处理器。
使用 @app.on_event 注册自定义事件处理器：
python
from onestep import InMemoryMetrics, OneStepApp
app = OneStepApp("demo")
metrics = InMemoryMetrics()
# 注册需要由应用持有的指标处理器
app.on_event(metrics)
@app.task(source=...)
async def my_task(ctx, item):
return item执行事件
任务执行过程中会发出以下事件：
| 事件 | 触发时机 |
|---|---|
fetched | 从 Source 获取消息 |
started | 开始执行任务 |
succeeded | 任务执行成功 |
retried | 任务重试 |
failed | 任务最终失败 |
dead_lettered | 消息进入死信队列 |
cancelled | 任务被取消 |
自定义事件处理器
实现自定义事件处理器：
python
from onestep import TaskEvent, TaskEventKind
@app.on_event
def log_event(event: TaskEvent):
if event.kind is TaskEventKind.SUCCEEDED:
print(f"任务成功: {event.task}, 耗时: {event.duration_s:.2f}s")
elif event.kind is TaskEventKind.FAILED and event.failure is not None:
print(f"任务失败: {event.task}, 原因: {event.failure.message}")内置事件处理器
InMemoryMetrics
内存指标收集器：
python
from onestep import InMemoryMetrics
metrics = InMemoryMetrics()
app.on_event(metrics)
# 获取指标快照
snapshot = metrics.snapshot()
print(snapshot["kinds"]) # 各类事件计数StructuredEventLogger
onestep run 默认启用结构化任务事件。嵌入式运行或需要自定义事件 logger 时，可以显式启用：
python
import logging
from onestep import StructuredEventLogger
app.on_event(
StructuredEventLogger(logger=logging.getLogger("billing.task_events"))
)输出包含字段：event_kind, app_name, task_name, source_name, attempts, duration_s, failure_kind
也可以使用 app.enable_structured_event_logging() 幂等地启用默认处理器；如果已经存在 StructuredEventLogger，该方法会复用它。CLI 参数和日志级别规则见 日志与任务事件。
生命周期钩子
@app.on_startup
应用启动时执行：
python
@app.on_startup
async def bootstrap(app):
print("应用启动")
# 初始化资源、预发布消息等@app.on_shutdown
应用关闭时执行：
python
@app.on_shutdown
async def cleanup(app):
print("应用关闭")
# 清理资源任务状态
每个任务可以维护独立的状态：
python
from onestep import InMemoryStateStore, OneStepApp
app = OneStepApp("state-demo", state=InMemoryStateStore())
@app.task(source=...)
async def track_runs(ctx, item):
# 获取状态
runs = await ctx.state.get("runs", 0)
# 更新状态
await ctx.state.set("runs", runs + 1)
print(f"已处理 {runs + 1} 条消息")配置访问
通过 ctx.config 访问应用配置：
python
app = OneStepApp("demo", config={"region": "cn", "debug": True})
@app.task(source=...)
async def my_task(ctx, item):
region = ctx.config["region"]
debug = ctx.config.get("debug", False)
...从 0.5.x 迁移
旧版中间件：
python
# 0.5.x
class MyMiddleware(BaseMiddleware):
def before_consume(self, step, message, *args, **kwargs):
...
@step(from_broker=..., middlewares=[MyMiddleware()])
def task(message):
...新版事件钩子：
python
# 1.x
@app.on_event
def log_event(event):
if event.kind.value == "started":
# 相当于 before_consume
...
elif event.kind.value == "succeeded":
# 相当于 after_consume
...
@app.task(source=...)
async def task(ctx, item):
...消息去重逻辑现在应该在任务处理器中实现，或使用数据库的唯一约束。