class_docstr: Interface around the redis for putting tasks to the queue
  
  Attributes:
      db (Redis): database object
      stream
      cache
      pubsub
class_docstr_tok: ["Interface","around","the","redis","for","putting","tasks","to","the","queue","Attributes",":","db","(","Redis",")",":","database","object","stream","cache","pubsub"]
docstr: Add tasks to the queue
  
  Args:
      task_type (ML_TASK_TYPE): type of the task
      model_id (int): model identifier
      payload (dict): lightweight model data that will be added to stream message
      dataframe (DataFrame): dataframe will be transfered via regular redis storage
  
  Returns:
      Task: object representing the task
docstr_tok: ["Add","tasks","to","the","queue","Args",":","task_type","(","ML_TASK_TYPE",")",":","type","of","the","task","model_id","(","int",")",":","model","identifier","payload","(","dict",")",":","lightweight","model","data","that","will","be","added","to","stream","message","dataframe","(","DataFrame",")",":","dataframe","will","be","transfered","via","regular","redis","storage","Returns",":","Task",":","object","representing","the","task"]
code: def apply_async(
      self,
      task_type: ML_TASK_TYPE,
      model_id: int,
      payload: dict,
      dataframe: DataFrame = None,
  ) -> Task:
      """Add tasks to the queue
  
      Args:
          task_type (ML_TASK_TYPE): type of the task
          model_id (int): model identifier
          payload (dict): lightweight model data that will be added to stream message
          dataframe (DataFrame): dataframe will be transfered via regular redis storage
  
      Returns:
          Task: object representing the task
      """
      try:
          payload = pickle.dumps(payload, protocol=5)
          redis_key = RedisKey.new()
          message = {
              "task_type": task_type.value,
              "company_id": ""
              if ctx.company_id is None
              else ctx.company_id,  # None can not be dumped
              "model_id": model_id,
              "payload": payload,
              "redis_key": redis_key.base,
          }
  
          self.wait_redis_ping()
          if dataframe is not None:
              self.cache.set(
                  redis_key.dataframe, to_bytes(dataframe), 180
              )
          self.cache.set(redis_key.status, ML_TASK_STATUS.WAITING, 180)
  
          self.stream.add(message)
          return Task(self.db, redis_key)
      except ConnectionError:
          logger.error("Cant send message to redis: connect failed")
          raise
  
file_path: mindsdb/mindsdb/utilities/ml_task_queue/producer.py
imports: import pickle
  from walrus import Database
  from pandas import DataFrame
  from mindsdb.utilities.context import context
  from mindsdb.utilities.config import Config
  from mindsdb.utilities.ml_task_queue.utils import RedisKey, to_bytes
  from mindsdb.utilities.ml_task_queue.task import Task
  from mindsdb.utilities.ml_task_queue.base import BaseRedisQueue
  from mindsdb.utilities.ml_task_queue.const import TASKS_STREAM_NAME, ML_TASK_TYPE, ML_TASK_STATUS
  from mindsdb.utilities import log
label_desc: Use image node_id 2 for calling the MLTaskProducer obj's underlying member method code with example usage: obj.apply_async(task_type, model_id, payload, dataframe) and returns: Task