我想知道是否有一种标准化的方法或最佳实践来扫描/自动发现装饰器,就像它在这里完成的那样,而且在 Django、Flask 等其他几个库中也是如此。通常,装饰器会在调用内部函数时提供额外/包装的功能。
在下面显示的示例中以及在 Flask/Django(路由装饰器)中,装饰器用于添加总体功能,例如最初在装饰器逻辑中生成 tcp 客户端,然后在收到要处理的消息时调用内部函数它。
Flask/ Django 注册一个 url 路由,其中内部函数仅在稍后请求 url 时调用。所有示例都需要对装饰器逻辑进行初始注册(扫描/发现),以便初始启动总体功能。对我来说,这似乎是装饰器的另一种用途,如果有的话,我想了解最佳实践方法。
请参见下面的浮士德示例,其中装饰器 app.agent() 在异步事件循环中自动触发侦听(kafka 流)客户端,然后传入消息由内部函数 hello()稍后处理,仅当收到消息时,需要初始在脚本开始时首先检查/扫描/发现相关的装饰器逻辑。
import faust
class Greeting(faust.Record):
from_name: str
to_name: str
app = faust.App('hello-app', broker='kafka://localhost')
topic = app.topic('hello-topic', value_type=Greeting)
@app.agent(topic)
async def hello(greetings):
async for greeting in greetings:
print(f'Hello from {greeting.from_name} to {greeting.to_name}')
@app.timer(interval=1.0)
async def example_sender(app):
await hello.send(
value=Greeting(from_name='Faust', to_name='you'),
)
if __name__ == '__main__':
app.main()