class_docstr: An extension for the scheduler to manage Variables
  
  This adds the following routes to the scheduler
  
  *  variable-set
  *  variable-get
  *  variable-delete
class_docstr_tok: ["An","extension","for","the","scheduler","to","manage","Variables","This","adds","the","following","routes","to","the","scheduler","*","variable-set","*","variable-get","*","variable-delete"]
code: def __init__(self, scheduler):
      self.scheduler = scheduler
      self.variables = dict()
      self.waiting = defaultdict(set)
      self.waiting_conditions = defaultdict(asyncio.Condition)
      self.started = asyncio.Condition()
  
      self.scheduler.handlers.update(
          {"variable_set": self.set, "variable_get": self.get}
      )
  
      self.scheduler.stream_handlers[
          "variable-future-release"
      ] = self.future_release
      self.scheduler.stream_handlers["variable_delete"] = self.delete
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import uuid
  from collections import defaultdict
  from contextlib import suppress
  from tlz import merge
  from dask.utils import parse_timedelta
  from distributed.client import Future
  from distributed.metrics import time
  from distributed.utils import TimeoutError, log_errors, wait_for
  from distributed.worker import get_client
label_desc: Use image node_id 1 to create a new VariableExtension object  with example: obj = VariableExtension(scheduler)