faststream 自己的asgi 实现

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/