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.

Request

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.

Request

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"]

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 global_dependencies and router_dependencies parameters work in the main handler, on_connect, and on_close callbacks.

WebSocket DI

WebSocket

/chat

  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"New connection: {websocket.id}")

      while True:
          message = await websocket.receive_text()
          cache.set(f"ws_{websocket.id}", message)
          await websocket.broadcast(f"User {websocket.id}: {message}")

  @chat.on_connect
  async def on_connect(websocket, global_dependencies=None):
      logger = global_dependencies.get("logger")
      logger.info(f"Client connected: {websocket.id}")
      return "Connected"

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.

Exceptions