faststream 目前自己包含了一个asgi 的实现，可以快速实现api 能力，同时也可以与其他web 框架集成，以下是一个简单试用
参考代码
- demo.py
from faststream.redis import RedisBroker
from faststream.asgi import AsgiFastStream,get,AsgiResponse
from typing import Any
from pydantic import BaseModel
class Msg(BaseModel):
name: str
age: int
broker = RedisBroker("redis://localhost:6379",validate=True)
@get
async def liveness_ping(scope):
return AsgiResponse(b"is liveness ping", status_code=200)
app = AsgiFastStream(broker,asgi_routes=[
("/",liveness_ping),
],
asyncapi_path="/docs",
)
@broker.publisher("response")
@broker.subscriber("test")
async def msg(msg: Msg):
print("test",msg)
return {"name":"dalong","age":18}
@broker.subscriber("response")
async def msgv2(msg: Any):
print("response",msg)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
- fastapi 集成
一种是基于asgi 兼容，还有就是基于扩展的router 实现
from contextlib import asynccontextmanager
from fastapi import FastAPI
from faststream import FastStream
from faststream.nats import NatsBroker
from faststream.asgi import make_ping_asgi, make_asyncapi_asgi
broker = NatsBroker()
@asynccontextmanager
async def start_broker(app):
"""Start the broker with the app."""
async with broker:
await broker.start()
yield
app = FastAPI(lifespan=start_broker)
app.mount("/health", make_ping_asgi(broker, timeout=5.0))
app.mount("/asyncapi", make_asyncapi_asgi(FastStream(broker)))
基于router实现
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from faststream.redis.fastapi import RedisRouter, Logger
from typing import Any
router = RedisRouter("redis://localhost:6379")
class Incoming(BaseModel):
m: dict
def call():
return True
@router.subscriber("test")
@router.publisher("response")
async def hello(m: Incoming, logger: Logger, d=Depends(call)):
logger.info(m)
return {"name":"dalong","age":18}
@router.subscriber("response")
async def appdemo(m: Any,logger: Logger):
logger.info(m)
return {"response": "Hello, response!"}
@router.get("/")
async def hello_http():
return "Hello, HTTP!"
@router.post("/send")
async def hello_http(msg: Incoming):
await router.broker.publish(msg,"test")
return "Hello, HTTP!"
app = FastAPI(lifespan=router.lifespan_context)
app.include_router(router)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
参考资料
https://faststream.airt.ai/latest/getting-started/asgi/
标签：faststream,asgi,app,broker,实现,import,router From： https://www.cnblogs.com/rongfengliang/p/18350351