code: def _do_push_bento(
      self,
      bento: Bento,
      upload_task_id: TaskID,
      *,
      force: bool = False,
      threads: int = 10,
      context: str | None = None,
      model_store: ModelStore = Provide[BentoMLContainer.model_store],
  ):
      yatai_rest_client = get_rest_api_client(context)
      name = bento.tag.name
      version = bento.tag.version
      if version is None:
          raise BentoMLException(
              f"Bento {bento.tag} version cannot be None"
          )
      info = bento.info
      model_tags = [m.tag for m in info.models]
      local_model_store = bento._model_store  # type: ignore  # using internal BentoML API
      if (
          local_model_store is not None
          and len(local_model_store.list()) > 0
      ):
          model_store = local_model_store
      models = (model_store.get(name) for name in model_tags)
      with ThreadPoolExecutor(
          max_workers=max(len(model_tags), 1)
      ) as executor:
  
          def push_model(model: Model) -> None:
              model_upload_task_id = (
                  self.transmission_progress.add_task(
                      f'Pushing model "{model.tag}"',
                      start=False,
                      visible=False,
                  )
              )
              self._do_push_model(
                  model,
                  model_upload_task_id,
                  force=force,
                  threads=threads,
                  context=context,
              )
  
          futures: t.Iterator[None] = executor.map(push_model, models)
          list(futures)
      with self.spin(text=f'Fetching Bento repository "{name}"'):
          bento_repository = yatai_rest_client.get_bento_repository(
              bento_repository_name=name
          )
      if not bento_repository:
          with self.spin(
              text=f'Bento repository "{name}" not found, creating now..'
          ):
              bento_repository = (
                  yatai_rest_client.create_bento_repository(
                      req=CreateBentoRepositorySchema(
                          name=name, description=""
                      )
                  )
              )
      with self.spin(
          text=f'Try fetching Bento "{bento.tag}" from remote Bento store..'
      ):
          remote_bento = yatai_rest_client.get_bento(
              bento_repository_name=name, version=version
          )
      if (
          not force
          and remote_bento
          and remote_bento.upload_status == BentoUploadStatus.SUCCESS
      ):
          self.log_progress.add_task(
              f'[bold blue]Push failed: Bento "{bento.tag}" already exists in remote Bento store'
          )
          return
      labels: list[LabelItemSchema] = [
          LabelItemSchema(key=key, value=value)
          for key, value in info.labels.items()
      ]
      apis: dict[str, BentoApiSchema] = {}
      models = [str(m.tag) for m in info.models]
      runners = [
          BentoRunnerSchema(
              name=r.name,
              runnable_type=r.runnable_type,
              models=r.models,
              resource_config=BentoRunnerResourceSchema(
                  cpu=r.resource_config.get("cpu"),
                  nvidia_gpu=r.resource_config.get("nvidia.com/gpu"),
                  custom_resources=r.resource_config.get(
                      "custom_resources"
                  ),
              )
              if r.resource_config
              else None,
          )
          for r in info.runners
      ]
      manifest = BentoManifestSchema(
          service=info.service,
          bentoml_version=info.bentoml_version,
          apis=apis,
          models=models,
          runners=runners,
          size_bytes=bento.total_size(),
      )
      if not remote_bento:
          with self.spin(
              text=f'Registering Bento "{bento.tag}" with remote Bento store..'
          ):
              remote_bento = yatai_rest_client.create_bento(
                  bento_repository_name=bento_repository.name,
                  req=CreateBentoSchema(
                      description="",
                      version=version,
                      build_at=info.creation_time,
                      manifest=manifest,
                      labels=labels,
                  ),
              )
      else:
          with self.spin(text=f'Updating Bento "{bento.tag}"..'):
              remote_bento = yatai_rest_client.update_bento(
                  bento_repository_name=bento_repository.name,
                  version=version,
                  req=UpdateBentoSchema(
                      manifest=manifest,
                      labels=labels,
                  ),
              )
  
      transmission_strategy: TransmissionStrategy = "proxy"
      presigned_upload_url: str | None = None
  
      if remote_bento.transmission_strategy is not None:
          transmission_strategy = remote_bento.transmission_strategy
      else:
          with self.spin(
              text=f'Getting a presigned upload url for bento "{bento.tag}" ..'
          ):
              remote_bento = yatai_rest_client.presign_bento_upload_url(
                  bento_repository_name=bento_repository.name,
                  version=version,
              )
              if remote_bento.presigned_upload_url:
                  transmission_strategy = "presigned_url"
                  presigned_upload_url = (
                      remote_bento.presigned_upload_url
                  )
  
      def io_cb(x: int):
          with io_mutex:
              self.transmission_progress.update(
                  upload_task_id, advance=x
              )
  
      with CallbackIOWrapper(read_cb=io_cb) as tar_io:
          with self.spin(
              text=f'Creating tar archive for bento "{bento.tag}"..'
          ):
              with tarfile.open(fileobj=tar_io, mode="w:") as tar:
  
                  def filter_(
                      tar_info: tarfile.TarInfo,
                  ) -> t.Optional[tarfile.TarInfo]:
                      if (
                          tar_info.path == "./models"
                          or tar_info.path.startswith("./models/")
                      ):
                          return None
                      return tar_info
  
                  tar.add(bento.path, arcname="./", filter=filter_)
          tar_io.seek(0, 0)
  
          with self.spin(text=f'Start uploading bento "{bento.tag}"..'):
              yatai_rest_client.start_upload_bento(
                  bento_repository_name=bento_repository.name,
                  version=version,
              )
  
          file_size = tar_io.getbuffer().nbytes
  
          self.transmission_progress.update(
              upload_task_id, completed=0, total=file_size, visible=True
          )
          self.transmission_progress.start_task(upload_task_id)
  
          io_mutex = threading.Lock()
  
          if transmission_strategy == "proxy":
              try:
                  yatai_rest_client.upload_bento(
                      bento_repository_name=bento_repository.name,
                      version=version,
                      data=tar_io,
                  )
              except Exception as e:  # pylint: disable=broad-except
                  self.log_progress.add_task(
                      f'[bold red]Failed to upload bento "{bento.tag}"'
                  )
                  raise e
              self.log_progress.add_task(
                  f'[bold green]Successfully pushed bento "{bento.tag}"'
              )
              return
          finish_req = FinishUploadBentoSchema(
              status=BentoUploadStatus.SUCCESS,
              reason="",
          )
          try:
              if presigned_upload_url is not None:
                  resp = requests.put(presigned_upload_url, data=tar_io)
                  if resp.status_code != 200:
                      finish_req = FinishUploadBentoSchema(
                          status=BentoUploadStatus.FAILED,
                          reason=resp.text,
                      )
              else:
                  with self.spin(
                      text=f'Start multipart uploading Bento "{bento.tag}"...'
                  ):
                      remote_bento = yatai_rest_client.start_bento_multipart_upload(
                          bento_repository_name=bento_repository.name,
                          version=version,
                      )
                      if not remote_bento.upload_id:
                          raise BentoMLException(
                              f'Failed to start multipart upload for Bento "{bento.tag}", upload_id is empty'
                          )
  
                      upload_id: str = remote_bento.upload_id
  
                  chunks_count = file_size // FILE_CHUNK_SIZE + 1
  
                  def chunk_upload(
                      upload_id: str, chunk_number: int
                  ) -> FinishUploadBentoSchema | tuple[str, int]:
                      with self.spin(
                          text=f'({chunk_number}/{chunks_count}) Presign multipart upload url of Bento "{bento.tag}"...'
                      ):
                          remote_bento = yatai_rest_client.presign_bento_multipart_upload_url(
                              bento_repository_name=bento_repository.name,
                              version=version,
                              req=PreSignMultipartUploadUrlSchema(
                                  upload_id=upload_id,
                                  part_number=chunk_number,
                              ),
                          )
                      with self.spin(
                          text=f'({chunk_number}/{chunks_count}) Uploading chunk of Bento "{bento.tag}"...'
                      ):
                          chunk = (
                              tar_io.getbuffer()[
                                  (chunk_number - 1)
                                  * FILE_CHUNK_SIZE : chunk_number
                                  * FILE_CHUNK_SIZE
                              ]
                              if chunk_number < chunks_count
                              else tar_io.getbuffer()[
                                  (chunk_number - 1) * FILE_CHUNK_SIZE :
                              ]
                          )
  
                          with CallbackIOWrapper(
                              chunk, read_cb=io_cb
                          ) as chunk_io:
                              resp = requests.put(
                                  remote_bento.presigned_upload_url,
                                  data=chunk_io,
                              )
                              if resp.status_code != 200:
                                  return FinishUploadBentoSchema(
                                      status=BentoUploadStatus.FAILED,
                                      reason=resp.text,
                                  )
                              return resp.headers["ETag"], chunk_number
  
                  futures_: list[
                      Future[FinishUploadBentoSchema | tuple[str, int]]
                  ] = []
  
                  with ThreadPoolExecutor(
                      max_workers=min(max(chunks_count, 1), threads)
                  ) as executor:
                      for i in range(1, chunks_count + 1):
                          future = executor.submit(
                              chunk_upload,
                              upload_id,
                              i,
                          )
                          futures_.append(future)
  
                  parts: list[CompletePartSchema] = []
  
                  for future in futures_:
                      result = future.result()
                      if isinstance(result, FinishUploadBentoSchema):
                          finish_req = result
                          break
                      else:
                          etag, chunk_number = result
                          parts.append(
                              CompletePartSchema(
                                  part_number=chunk_number,
                                  etag=etag,
                              )
                          )
  
                  with self.spin(
                      text=f'Completing multipart upload of Bento "{bento.tag}"...'
                  ):
                      remote_bento = yatai_rest_client.complete_bento_multipart_upload(
                          bento_repository_name=bento_repository.name,
                          version=version,
                          req=CompleteMultipartUploadSchema(
                              upload_id=upload_id,
                              parts=parts,
                          ),
                      )
  
          except Exception as e:  # pylint: disable=broad-except
              finish_req = FinishUploadBentoSchema(
                  status=BentoUploadStatus.FAILED,
                  reason=str(e),
              )
          if finish_req.status is BentoUploadStatus.FAILED:
              self.log_progress.add_task(
                  f'[bold red]Failed to upload Bento "{bento.tag}"'
              )
          with self.spin(
              text="Submitting upload status to remote Bento store"
          ):
              yatai_rest_client.finish_upload_bento(
                  bento_repository_name=bento_repository.name,
                  version=version,
                  req=finish_req,
              )
          if finish_req.status != BentoUploadStatus.SUCCESS:
              self.log_progress.add_task(
                  f'[bold red]Failed pushing Bento "{bento.tag}": {finish_req.reason}'
              )
          else:
              self.log_progress.add_task(
                  f'[bold green]Successfully pushed Bento "{bento.tag}"'
              )
  
imports: from __future__ import annotations
  import tarfile
  import tempfile
  import threading
  import typing
  import warnings
  from concurrent.futures import ThreadPoolExecutor
  from pathlib import Path
  from tempfile import NamedTemporaryFile
  import fs
  import requests
  from rich.live import Live
  from simple_di import Provide
  from simple_di import inject
  from ...exceptions import BentoMLException
  from ...exceptions import NotFound
  from ..bento import Bento
  from ..bento import BentoStore
  from ..configuration.containers import BentoMLContainer
  from ..models import Model
  from ..models import ModelStore
  from ..models import copy_model
  from ..tag import Tag
  from ..utils import calc_dir_size
  from .base import FILE_CHUNK_SIZE
  from .base import CallbackIOWrapper
  from .base import CloudClient
  from .config import get_rest_api_client
  from .deployment import Deployment
  from .schemas import BentoApiSchema
  from .schemas import BentoManifestSchema
  from .schemas import BentoRunnerResourceSchema
  from .schemas import BentoRunnerSchema
  from .schemas import BentoUploadStatus
  from .schemas import CompleteMultipartUploadSchema
  from .schemas import CompletePartSchema
  from .schemas import CreateBentoRepositorySchema
  from .schemas import CreateBentoSchema
  from .schemas import CreateModelRepositorySchema
  from .schemas import CreateModelSchema
  from .schemas import FinishUploadBentoSchema
  from .schemas import FinishUploadModelSchema
  from .schemas import LabelItemSchema
  from .schemas import ModelManifestSchema
  from .schemas import ModelUploadStatus
  from .schemas import PreSignMultipartUploadUrlSchema
  from .schemas import TransmissionStrategy
  from .schemas import UpdateBentoSchema
label_desc: Use image node_id 3 for calling the BentoCloudClient obj's underlying member method code with example usage: obj._do_push_bento(bento, upload_task_id) and returns: tar_info, None, resp,  chunk_number, FinishUploadBentoSchema