【发布时间】:2025-12-03 01:10:01
【问题描述】:
我想知道是否有一种标准化的方法或最佳实践来扫描/自动发现装饰器,就像它在 here 以及在 Django、Flask 等其他几个库中所做的那样。通常,装饰器在调用内部函数时立即提供额外/包装的功能。
在下面显示的示例中,以及在 Flask/Django(路由装饰器)中,装饰器用于添加总体功能,例如最初在装饰器逻辑中生成 tcp 客户端,然后在收到消息来处理它时调用内部 func。
Flask/ Django 注册一个 url 路由,其中内部函数仅在稍后请求 url 时调用。所有示例都需要对装饰器逻辑进行初始注册(扫描/发现),以便初始启动总体功能。对我来说,这似乎是装饰器的另一种用法,如果有的话,我想了解最佳实践方法。
参见下面的Faust 示例,其中装饰器 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()
【问题讨论】:
-
欢迎任何更新此问题的提示,因为我假设我只是在这里搜索错误的关键字..
-
装饰器知识我不是太热,但我不明白你所说的“发现”是什么意思。
-
app.main() 正在收集/发现/搜索所有装饰器以提供附加功能。 Flask 正在做一些非常相似的事情,请参见此处的第一个代码示例:flask.pocoo.org
-
是的!浮士德使用“venusian”来搜索装饰器。这实际上会递归地导入包目录中的所有模块。我不是这种方法的忠实拥护者,但它很方便。需要注意的是,它可以轻松导入不应导入的内容,例如测试模块或不打算导入的脚本。 Django 使用不同的方法,您明确列出 INSTALLED_APPS 并在这些已安装的应用程序中搜索“admin.py”模块。
-
@asksol 你是最棒的,谢谢!