Dependency Injection
Batman wanted to learn about dependency injection in Robyn. Robyn introduced him to the concept of dependency injection and how it can be used in Robyn.
Robyn has two types of dependency injection: One is for the application level and the other is for the router level.
Application Level Dependency Injection
Application level dependency injection is used to inject dependencies into the application. These dependencies are available to all the requests.
# 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"]# 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"]Router Level Dependency Injection
Router level dependency injection is used to inject dependencies into the router. These dependencies are available to all the requests of that router.
# 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"]# 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
Note: router_dependencies, global_dependencies are reserved parameters and must be named as such. The order of the parameters does not matter among them. However, the router_dependencies and global_dependencies must only come after the request parameter.
WebSocket Dependency Injection
WebSockets support the same dependency injection system as HTTP routes. The same global_dependencies and router_dependencies parameters work identically in WebSocket handlers.
from robyn import Robyn
import logging
app = Robyn(__file__)
# Same dependency setup as HTTP routes
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):
# Access dependencies same as HTTP routes
logger = global_dependencies.get("logger")
database = global_dependencies.get("database")
cache = router_dependencies.get("cache")
auth = router_dependencies.get("auth_service")
logger.info(f"New chat connection: {websocket.id}")
while True:
message = await websocket.receive_text()
# Use injected dependencies
database.save_message(message, websocket.id)
await websocket.broadcast(f"User {websocket.id}: {message}")
@chat_handler.on_connect
async def on_connect(websocket, global_dependencies=None, router_dependencies=None):
# Connect and close handlers also support DI
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 "Unauthorized"
logger.info(f"Authenticated connection: {websocket.id}")
return "Connected"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()
# Use typed dependencies
if logger:
logger.info(f"Received data: {data}")
if cache:
cache.set(f"ws_{websocket.id}", data)
await websocket.send_json({"echo": data, "id": websocket.id})What's next?
Batman, being the familiar with the dark side wanted to know about Exceptions!
Robyn introduced him to the concept of exceptions and how he can use them to handle errors in his application.