依赖注入
Batman 想要了解 Robyn 中的依赖注入。Robyn 向他介绍了依赖注入的概念以及如何在 Robyn 中使用它。
Robyn 有两种类型的依赖注入：一种是应用程序级别，另一种是路由器级别。
应用程序级依赖注入
应用程序级依赖注入用于将依赖项注入应用程序。这些依赖项可供所有请求使用。
py
# GET /hello_world
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
# GET /hello_world
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
# GET /hello_world
from robyn import Robyn, ALLOW_CORS
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): # r is the request object
return router_dependencies["ROUTER_DEPENDENCY"]py
# GET /hello_world
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"]TIP
router_dependencies、global_dependencies 是保留参数，必须这样命名。这些参数的顺序无关紧要。
但是，router_dependencies 和 global_dependencies 必须位于请求参数之后。
WebSocket 依赖注入
WebSocket 支持与 HTTP 路由相同的依赖注入系统。相同的 global_dependencies 和 router_dependencies 参数在 WebSocket 处理程序中的工作方式完全一致。
python
from robyn import Robyn
import logging
app = Robyn(__file__)
# 与 HTTP 路由相同的依赖配置
app.inject_global(
logger=logging.getLogger(__name__),
database=DatabaseConnection()
)
app.inject(
cache=RedisCache(),
auth_service=JWTAuthService()
)
@app.websocket("/chat")
async def chat_handler(websocket, global_dependencies=None, router_dependencies=None):
# 访问依赖的方式与 HTTP 路由相同
logger = global_dependencies.get("logger")
database = global_dependencies.get("database")
cache = router_dependencies.get("cache")
auth = router_dependencies.get("auth_service")
logger.info(f"新的聊天连接: {websocket.id}")
while True:
message = await websocket.receive_text()
# 使用注入的依赖
database.save_message(message, websocket.id)
await websocket.broadcast(f"用户 {websocket.id}: {message}")
@chat_handler.on_connect
async def on_connect(websocket, global_dependencies=None, router_dependencies=None):
# 连接和关闭处理程序也支持依赖注入
logger = global_dependencies.get("logger")
auth = router_dependencies.get("auth_service")
if not auth.verify_token(websocket.query_params.get("token")):
await websocket.close()
return "未授权"
logger.info(f"已认证连接: {websocket.id}")
return "已连接"python
from robyn import Robyn, WebSocketAdapter
from typing import Optional, Dict, Any
app = Robyn(__file__)
app.inject_global(logger=logging.getLogger(__name__))
app.inject(cache=RedisCache())
@app.websocket("/api/ws")
async def api_handler(
websocket: WebSocketAdapter,
global_dependencies: Optional[Dict[str, Any]] = None,
router_dependencies: Optional[Dict[str, Any]] = None
):
logger = global_dependencies.get("logger") if global_dependencies else None
cache = router_dependencies.get("cache") if router_dependencies else None
while True:
data = await websocket.receive_json()
# 使用类型化依赖
if logger:
logger.info(f"接收到数据: {data}")
if cache:
cache.set(f"ws_{websocket.id}", data)
await websocket.send_json({"echo": data, "id": websocket.id})下一步是什么？
蝙蝠侠熟悉黑暗面，他想知道异常！
Robyn 向他介绍了异常的概念以及如何使用异常来处理应用程序中的错误。