依赖注入
蝙蝠侠想要了解 Robyn 中的依赖注入机制。Robyn 向他介绍了依赖注入的概念以及如何在应用程序中使用它。
Robyn 提供了两种依赖注入方式： 全局依赖注入和局部依赖注入。
应用程序级依赖注入
全局依赖注入用于将依赖项注入到整个应用程序中，注入的依赖项对所有请求都可用。
py
from robyn import Robyn
app = Robyn(__file__)
GLOBAL_DEPENDENCY = "GLOBAL DEPENDENCY"
app.inject_global(GLOBAL_DEPENDENCY=GLOBAL_DEPENDENCY)
@app.get("/sync/global_di")
def sync_global_di(request, global_dependencies):
return global_dependencies["GLOBAL_DEPENDENCY"]py
from robyn import Robyn, ALLOW_CORS
app = Robyn(__file__)
GLOBAL_DEPENDENCY = "GLOBAL DEPENDENCY"
app.inject_global(GLOBAL_DEPENDENCY=GLOBAL_DEPENDENCY)
@app.get("/sync/global_di")
def sync_global_di(request, global_dependencies):
return global_dependencies["GLOBAL_DEPENDENCY"]路由器级依赖注入
局部依赖注入用于将依赖项注入到特定路由器中，注入的依赖项仅对该路由器的请求有效。
py
from robyn import Robyn
app = Robyn(__file__)
ROUTER_DEPENDENCY = "ROUTER DEPENDENCY"
app.inject(ROUTER_DEPENDENCY=ROUTER_DEPENDENCY)
@app.get("/sync/global_di")
def sync_global_di(r, router_dependencies):
return router_dependencies["ROUTER_DEPENDENCY"]py
from robyn import Robyn, ALLOW_CORS, Request
app = Robyn(__file__)
ROUTER_DEPENDENCY = "ROUTER DEPENDENCY"
app.inject(ROUTER_DEPENDENCY=ROUTER_DEPENDENCY)
@app.get("/sync/global_di")
def sync_global_di(r: Request, router_dependencies):
return router_dependencies["ROUTER_DEPENDENCY"]注意：router_dependencies 和 global_dependencies 是保留参数名，必须按照这些名称使用。参数的顺序不重要，但这两个参数必须位于 request 参数之后。
WebSocket 依赖注入
WebSocket 支持与 HTTP 路由相同的依赖注入系统。global_dependencies 和 router_dependencies 参数可以在主处理程序、on_connect 和 on_close 回调中使用。
py
from robyn import Robyn
import logging
app = Robyn(__file__)
app.inject_global(logger=logging.getLogger(__name__))
app.inject(cache=RedisCache())
@app.websocket("/chat")
async def chat(websocket, global_dependencies=None, router_dependencies=None):
logger = global_dependencies.get("logger")
cache = router_dependencies.get("cache")
logger.info(f"新连接: {websocket.id}")
while True:
message = await websocket.receive_text()
cache.set(f"ws_{websocket.id}", message)
await websocket.broadcast(f"用户 {websocket.id}: {message}")
@chat.on_connect
async def on_connect(websocket, global_dependencies=None):
logger = global_dependencies.get("logger")
logger.info(f"客户端已连接: {websocket.id}")
return "已连接"py
from robyn import Robyn
import logging
app = Robyn(__file__)
app.inject_global(logger=logging.getLogger(__name__))
app.inject(cache=RedisCache())
@app.websocket("/chat")
async def chat(websocket, global_dependencies=None, router_dependencies=None):
logger = global_dependencies.get("logger")
cache = router_dependencies.get("cache")
logger.info(f"新连接: {websocket.id}")
while True:
message = await websocket.receive_text()
cache.set(f"ws_{websocket.id}", message)
await websocket.broadcast(f"用户 {websocket.id}: {message}")
@chat.on_connect
async def on_connect(websocket, global_dependencies=None):
logger = global_dependencies.get("logger")
logger.info(f"客户端已连接: {websocket.id}")
return "已连接"下一步
由于蝙蝠侠对黑暗面非常熟悉，他对异常处理产生了兴趣。
Robyn 随即向他介绍了异常处理的概念，以及如何通过异常处理来应对应用程序中的各种错误情况。