msaSignal - Starlette/FastAPI 的信号/事件。
Starlette/FastAPI 的信号/事件。在不阻塞创建信号的函数的情况下运行后台任务。msaSDK.signals 尝试继续作为后台任务运行程序。

文档：msaSignal 文档 ( https://msaSignal.u2d.ai/ )

特征

• 中间件：用于信号或任务。
• 装饰器：用于信号注册
• Helpers：对于 Handler，启动 Signal 或 Task

主要依赖

• 小星星 0.20.x

用法 - MSASignalMiddleware

！！！note 每个函数只有一个信号，必须以请求对象为 arg

添加中间件

from msaSignal import MSASignalMiddleware, signal
from fastapi import FastAPI

app = FastAPI()
app.add_midleware(MSASignalMiddleware, handler=signal)

# OR enable MSASetting for this feature (settings.signal_middleware)

添加处理程序

指定触发信号的工作方式。

from msaSignal import signal
import asyncio

@signal.register
async def handler(**kwargs):
    await asyncio.sleep(5)
    print(kwargs)
    print('Works!')

功能中的火警信号

！！！note 使用后台任务只允许一次信号调用。

from msaSignal import initiate_signal
@app.get("/")
async def endpoint(request: Request):
    await initiate_signal(request, 'handler',some_data="test value")
    return {"status":"Success"}

用法 - MSATaskMiddleware

任意数量的任务，不需要请求对象。

添加中间件

from msaSignal import MSATaskMiddleware
from fastapi import FastAPI
app = FastAPI()
app.add_midleware(MSATaskMiddleware)

# OR enable MSASetting for this feature (settings.task_middleware)

写处理程序

指定触发的任务应该如何工作。

async def handler():
    await asyncio.sleep(5)
    print('Works!')

函数中的触发任务

from msaSignal import initiate_task
@app.get("/")
async def endpoint():
    await initiate_task(handler,some_data="test value")
    return {"status":"Success"}

许可协议

• msaSignal基于MIT开源和免费使用，可免费用于商业用途，但请在某处显示/列出有关 msaSignal 的版权信息。

如何创建文档

我们使用 mkdocs 和 mkdocsstring。代码引用和导航条目实际上是由触发的 Python 脚本 /docs/gen_ref_pages.py 在执行mkdocs serve或build执行时创建的。

PDF 创建选项的要求安装：

PDF Export 主要使用 weasyprint，如果您在这里遇到一些错误，请。检查那里的文档。安装是 msaSignal 的一部分，所以应该没问题。

我们现在可以使用以下方法测试和查看我们的文档：

mkdocs serve

构建静态站点：

mkdocs build

构建和发布

建造：

python setup.py sdist

发布到 pypi：

twine upload dist/*