Repository: bentoml/BentoML
Pull request #4419 (2024-01-18 09:28:38)

fix: refactor cli & sdk for deployment
## What does this PR address?

<!--
Thanks for sending a pull request!

Congrats for making it this far! Here's a 🍱 for you. There are still a few steps ahead.

Please make sure to read the contribution guidelines, then fill out the blanks below before requesting a code review.

Name your Pull Request with one of the following prefixes, e.g. "feat: add support for PyTorch", to indicate the type of changes proposed. This is based on the [Conventional Commits specification](https://www.conventionalcommits.org/en/v1.0.0/#summary).
  - feat: (new feature for the user, not a new feature for build script)
  - fix: (bug fix for the user, not a fix to a build script)
  - docs: (changes to the documentation)
  - style: (formatting, missing semicolons, etc; no production code change)
  - refactor: (refactoring production code, eg. renaming a variable)
  - perf: (code changes that improve performance)
  - test: (adding missing tests, refactoring tests; no production code change)
  - chore: (updating grunt tasks etc; no production code change)
  - build: (changes that affect the build system or external dependencies)
  - ci: (changes to configuration files and scripts)
  - revert: (reverts a previous commit)

Describe your changes in detail. Attach screenshots here if appropriate.

Once you're done with this, someone from BentoML team or community member will help review your PR (see "Who can help review?" section for potential reviewers.). If no one has reviewed your PR after a week have passed, don't hesitate to post a new comment and ping @-the same person. Notifications sometimes get lost 🥲.
-->

<!-- Remove if not applicable -->

Fixes #(issue)

## Before submitting:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
<!--- If you plan to update documentation or tests in follow-up, please note -->

- [ ] Does the Pull Request follow [Conventional Commits specification](https://www.conventionalcommits.org/en/v1.0.0/#summary) naming? Here are [GitHub's
      guide](https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/proposing-changes-to-your-work-with-pull-requests/creating-a-pull-request) on how to create a pull request.
- [ ] Does the code follow BentoML's code style, `pre-commit run -a` script has passed ([instructions](https://github.com/bentoml/BentoML/blob/main/DEVELOPMENT.md#style-check-auto-formatting-type-checking))?
- [ ] Did you read through [contribution guidelines](https://github.com/bentoml/BentoML/blob/main/CONTRIBUTING.md#ways-to-contribute) and follow [development guidelines](https://github.com/bentoml/BentoML/blob/main/DEVELOPMENT.md#start-developing)?
- [ ] Did your changes require updates to the documentation? Have you updated
      those accordingly? Here are [documentation guidelines](https://github.com/bentoml/BentoML/tree/main/docs) and [tips on writting docs](https://github.com/bentoml/BentoML/tree/main/docs#writing-documentation).
- [ ] Did you write tests to cover your changes?



Patch:
diff --git a/src/bentoml/_internal/cloud/base.py b/src/bentoml/_internal/cloud/base.py
index c8cdd162..ab8ac0b1 100644
--- a/src/bentoml/_internal/cloud/base.py
+++ b/src/bentoml/_internal/cloud/base.py
@@ -58,8 +58,7 @@ class CallbackIOWrapper(io.BytesIO):
         return res
 
 
-class CloudClient(ABC):
-    # Moved atrributes to __init__ because otherwise it will keep all the log when running SDK.
+class Spinner(ABC):
     def __init__(self):
         self.log_progress = Progress(TextColumn("{task.description}"))
 
@@ -96,6 +95,12 @@ class CloudClient(ABC):
             self.spinner_progress.stop_task(task_id)
             self.spinner_progress.update(task_id, visible=False)
 
+
+class CloudClient(ABC):
+    # Moved atrributes to __init__ because otherwise it will keep all the log when running SDK.
+    def __init__(self):
+        self.spinner = Spinner()
+
     @abstractmethod
     def push_model(
         self,
diff --git a/src/bentoml/_internal/cloud/bentocloud.py b/src/bentoml/_internal/cloud/bentocloud.py
index 04472d98..cfce7098 100644
--- a/src/bentoml/_internal/cloud/bentocloud.py
+++ b/src/bentoml/_internal/cloud/bentocloud.py
@@ -67,8 +67,8 @@ class BentoCloudClient(CloudClient):
         threads: int = 10,
         context: str | None = None,
     ):
-        with Live(self.progress_group):
-            upload_task_id = self.transmission_progress.add_task(
+        with Live(self.spinner.progress_group):
+            upload_task_id = self.spinner.transmission_progress.add_task(
                 f'Pushing Bento "{bento.tag}"', start=False, visible=False
             )
             self._do_push_bento(
@@ -100,7 +100,7 @@ class BentoCloudClient(CloudClient):
         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(
+                model_upload_task_id = self.spinner.transmission_progress.add_task(
                     f'Pushing model "{model.tag}"', start=False, visible=False
                 )
                 self._do_push_model(
@@ -113,16 +113,18 @@ class BentoCloudClient(CloudClient):
 
             futures: t.Iterator[None] = executor.map(push_model, models)
             list(futures)
-        with self.spin(text=f'Fetching Bento repository "{name}"'):
+        with self.spinner.spin(text=f'Fetching Bento repository "{name}"'):
             bento_repository = yatai_rest_client.v1.get_bento_repository(
                 bento_repository_name=name
             )
         if not bento_repository:
-            with self.spin(text=f'Bento repository "{name}" not found, creating now..'):
+            with self.spinner.spin(
+                text=f'Bento repository "{name}" not found, creating now..'
+            ):
                 bento_repository = yatai_rest_client.v1.create_bento_repository(
                     req=CreateBentoRepositorySchema(name=name, description="")
                 )
-        with self.spin(
+        with self.spinner.spin(
             text=f'Try fetching Bento "{bento.tag}" from remote Bento store..'
         ):
             remote_bento = yatai_rest_client.v1.get_bento(
@@ -133,7 +135,7 @@ class BentoCloudClient(CloudClient):
             and remote_bento
             and remote_bento.upload_status == BentoUploadStatus.SUCCESS
         ):
-            self.log_progress.add_task(
+            self.spinner.log_progress.add_task(
                 f'[bold blue]Push skipped: Bento "{bento.tag}" already exists in remote Bento store'
             )
             return
@@ -170,7 +172,7 @@ class BentoCloudClient(CloudClient):
             envs=info.envs,
         )
         if not remote_bento:
-            with self.spin(
+            with self.spinner.spin(
                 text=f'Registering Bento "{bento.tag}" with remote Bento store..'
             ):
                 remote_bento = yatai_rest_client.v1.create_bento(
@@ -184,7 +186,7 @@ class BentoCloudClient(CloudClient):
                     ),
                 )
         else:
-            with self.spin(text=f'Updating Bento "{bento.tag}"..'):
+            with self.spinner.spin(text=f'Updating Bento "{bento.tag}"..'):
                 remote_bento = yatai_rest_client.v1.update_bento(
                     bento_repository_name=bento_repository.name,
                     version=version,
@@ -200,7 +202,7 @@ class BentoCloudClient(CloudClient):
         if remote_bento.transmission_strategy is not None:
             transmission_strategy = remote_bento.transmission_strategy
         else:
-            with self.spin(
+            with self.spinner.spin(
                 text=f'Getting a presigned upload url for bento "{bento.tag}" ..'
             ):
                 remote_bento = yatai_rest_client.v1.presign_bento_upload_url(
@@ -212,10 +214,12 @@ class BentoCloudClient(CloudClient):
 
         def io_cb(x: int):
             with io_mutex:
-                self.transmission_progress.update(upload_task_id, advance=x)
+                self.spinner.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 self.spinner.spin(
+                text=f'Creating tar archive for bento "{bento.tag}"..'
+            ):
                 with tarfile.open(fileobj=tar_io, mode="w:") as tar:
 
                     def filter_(
@@ -230,17 +234,17 @@ class BentoCloudClient(CloudClient):
                     tar.add(bento.path, arcname="./", filter=filter_)
             tar_io.seek(0, 0)
 
-            with self.spin(text=f'Start uploading bento "{bento.tag}"..'):
+            with self.spinner.spin(text=f'Start uploading bento "{bento.tag}"..'):
                 yatai_rest_client.v1.start_upload_bento(
                     bento_repository_name=bento_repository.name, version=version
                 )
 
             file_size = tar_io.getbuffer().nbytes
 
-            self.transmission_progress.update(
+            self.spinner.transmission_progress.update(
                 upload_task_id, completed=0, total=file_size, visible=True
             )
-            self.transmission_progress.start_task(upload_task_id)
+            self.spinner.transmission_progress.start_task(upload_task_id)
 
             io_mutex = threading.Lock()
 
@@ -252,11 +256,11 @@ class BentoCloudClient(CloudClient):
                         data=tar_io,
                     )
                 except Exception as e:  # pylint: disable=broad-except
-                    self.log_progress.add_task(
+                    self.spinner.log_progress.add_task(
                         f'[bold red]Failed to upload bento "{bento.tag}"'
                     )
                     raise e
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold green]Successfully pushed bento "{bento.tag}"'
                 )
                 return
@@ -273,7 +277,7 @@ class BentoCloudClient(CloudClient):
                             reason=resp.text,
                         )
                 else:
-                    with self.spin(
+                    with self.spinner.spin(
                         text=f'Start multipart uploading Bento "{bento.tag}"...'
                     ):
                         remote_bento = (
@@ -294,7 +298,7 @@ class BentoCloudClient(CloudClient):
                     def chunk_upload(
                         upload_id: str, chunk_number: int
                     ) -> FinishUploadBentoSchema | tuple[str, int]:
-                        with self.spin(
+                        with self.spinner.spin(
                             text=f'({chunk_number}/{chunks_count}) Presign multipart upload url of Bento "{bento.tag}"...'
                         ):
                             remote_bento = (
@@ -307,7 +311,7 @@ class BentoCloudClient(CloudClient):
                                     ),
                                 )
                             )
-                        with self.spin(
+                        with self.spinner.spin(
                             text=f'({chunk_number}/{chunks_count}) Uploading chunk of Bento "{bento.tag}"...'
                         ):
                             chunk = (
@@ -364,7 +368,7 @@ class BentoCloudClient(CloudClient):
                                 )
                             )
 
-                    with self.spin(
+                    with self.spinner.spin(
                         text=f'Completing multipart upload of Bento "{bento.tag}"...'
                     ):
                         remote_bento = (
@@ -384,21 +388,23 @@ class BentoCloudClient(CloudClient):
                     reason=str(e),
                 )
             if finish_req.status is BentoUploadStatus.FAILED:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold red]Failed to upload Bento "{bento.tag}"'
                 )
-            with self.spin(text="Submitting upload status to remote Bento store"):
+            with self.spinner.spin(
+                text="Submitting upload status to remote Bento store"
+            ):
                 yatai_rest_client.v1.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(
+                self.spinner.log_progress.add_task(
                     f'[bold red]Failed pushing Bento "{bento.tag}": {finish_req.reason}'
                 )
             else:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold green]Successfully pushed Bento "{bento.tag}"'
                 )
 
@@ -411,8 +417,8 @@ class BentoCloudClient(CloudClient):
         context: str | None = None,
         bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     ) -> Bento:
-        with Live(self.progress_group):
-            download_task_id = self.transmission_progress.add_task(
+        with Live(self.spinner.progress_group):
+            download_task_id = self.spinner.transmission_progress.add_task(
                 f'Pulling bento "{tag}"', start=False, visible=False
             )
             return self._do_pull_bento(
@@ -437,7 +443,7 @@ class BentoCloudClient(CloudClient):
         try:
             bento = bento_store.get(tag)
             if not force:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold blue]Bento "{tag}" exists in local model store'
                 )
                 return bento
@@ -452,7 +458,7 @@ class BentoCloudClient(CloudClient):
 
         yatai_rest_client = get_rest_api_client(context)
 
-        with self.spin(text=f'Fetching bento "{_tag}"'):
+        with self.spinner.spin(text=f'Fetching bento "{_tag}"'):
             remote_bento = yatai_rest_client.v1.get_bento(
                 bento_repository_name=name, version=version
             )
@@ -467,8 +473,10 @@ class BentoCloudClient(CloudClient):
             ) as executor:
 
                 def pull_model(model_tag: Tag):
-                    model_download_task_id = self.transmission_progress.add_task(
-                        f'Pulling model "{model_tag}"', start=False, visible=False
+                    model_download_task_id = (
+                        self.spinner.transmission_progress.add_task(
+                            f'Pulling model "{model_tag}"', start=False, visible=False
+                        )
                     )
                     self._do_pull_model(
                         model_tag,
@@ -488,7 +496,7 @@ class BentoCloudClient(CloudClient):
             if remote_bento.transmission_strategy is not None:
                 transmission_strategy = remote_bento.transmission_strategy
             else:
-                with self.spin(
+                with self.spinner.spin(
                     text=f'Getting a presigned download url for bento "{_tag}"'
                 ):
                     remote_bento = yatai_rest_client.v1.presign_bento_download_url(
@@ -505,7 +513,7 @@ class BentoCloudClient(CloudClient):
                 )
             else:
                 if presigned_download_url is None:
-                    with self.spin(
+                    with self.spinner.spin(
                         text=f'Getting a presigned download url for bento "{_tag}"'
                     ):
                         remote_bento = yatai_rest_client.v1.presign_bento_download_url(
@@ -521,25 +529,25 @@ class BentoCloudClient(CloudClient):
                         )
                     total_size_in_bytes = int(response.headers.get("content-length", 0))
                     block_size = 1024  # 1 Kibibyte
-                    self.transmission_progress.update(
+                    self.spinner.transmission_progress.update(
                         download_task_id,
                         completed=0,
                         total=total_size_in_bytes,
                         visible=True,
                     )
-                    self.transmission_progress.start_task(download_task_id)
+                    self.spinner.transmission_progress.start_task(download_task_id)
                     for data in response.iter_bytes(block_size):
-                        self.transmission_progress.update(
+                        self.spinner.transmission_progress.update(
                             download_task_id, advance=len(data)
                         )
                         tar_file.write(data)
 
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold green]Finished downloading all bento "{_tag}" files'
                 )
                 tar_file.seek(0, 0)
                 tar = tarfile.open(fileobj=tar_file, mode="r")
-                with self.spin(text=f'Extracting bento "{_tag}" tar file'):
+                with self.spinner.spin(text=f'Extracting bento "{_tag}" tar file'):
                     with fs.open_fs("temp://") as temp_fs:
                         for member in tar.getmembers():
                             f = tar.extractfile(member)
@@ -551,7 +559,7 @@ class BentoCloudClient(CloudClient):
                             temp_fs.writebytes(member.name, f.read())
                         bento = Bento.from_fs(temp_fs)
                         for model_tag in remote_bento.manifest.models:
-                            with self.spin(
+                            with self.spinner.spin(
                                 text=f'Copying model "{model_tag}" to model store'
                             ):
                                 copy_model(
@@ -560,7 +568,7 @@ class BentoCloudClient(CloudClient):
                                     target_model_store=global_model_store,
                                 )
                         bento = bento.save(bento_store)
-                        self.log_progress.add_task(
+                        self.spinner.log_progress.add_task(
                             f'[bold green]Successfully pulled bento "{_tag}"'
                         )
                         return bento
@@ -574,8 +582,8 @@ class BentoCloudClient(CloudClient):
         threads: int = 10,
         context: str | None = None,
     ):
-        with Live(self.progress_group):
-            upload_task_id = self.transmission_progress.add_task(
+        with Live(self.spinner.progress_group):
+            upload_task_id = self.spinner.transmission_progress.add_task(
                 f'Pushing model "{model.tag}"', start=False, visible=False
             )
             self._do_push_model(
@@ -597,16 +605,18 @@ class BentoCloudClient(CloudClient):
         if version is None:
             raise BentoMLException(f'Model "{model.tag}" version cannot be None')
         info = model.info
-        with self.spin(text=f'Fetching model repository "{name}"'):
+        with self.spinner.spin(text=f'Fetching model repository "{name}"'):
             model_repository = yatai_rest_client.v1.get_model_repository(
                 model_repository_name=name
             )
         if not model_repository:
-            with self.spin(text=f'Model repository "{name}" not found, creating now..'):
+            with self.spinner.spin(
+                text=f'Model repository "{name}" not found, creating now..'
+            ):
                 model_repository = yatai_rest_client.v1.create_model_repository(
                     req=CreateModelRepositorySchema(name=name, description="")
                 )
-        with self.spin(
+        with self.spinner.spin(
             text=f'Try fetching model "{model.tag}" from remote model store..'
         ):
             remote_model = yatai_rest_client.v1.get_model(
@@ -617,7 +627,7 @@ class BentoCloudClient(CloudClient):
             and remote_model
             and remote_model.upload_status == ModelUploadStatus.SUCCESS
         ):
-            self.log_progress.add_task(
+            self.spinner.log_progress.add_task(
                 f'[bold blue]Model "{model.tag}" already exists in remote model store, skipping'
             )
             return
@@ -626,7 +636,7 @@ class BentoCloudClient(CloudClient):
                 LabelItemSchema(key=key, value=value)
                 for key, value in info.labels.items()
             ]
-            with self.spin(
+            with self.spinner.spin(
                 text=f'Registering model "{model.tag}" with remote model store..'
             ):
                 remote_model = yatai_rest_client.v1.create_model(
@@ -654,7 +664,7 @@ class BentoCloudClient(CloudClient):
         if remote_model.transmission_strategy is not None:
             transmission_strategy = remote_model.transmission_strategy
         else:
-            with self.spin(
+            with self.spinner.spin(
                 text=f'Getting a presigned upload url for Model "{model.tag}" ..'
             ):
                 remote_model = yatai_rest_client.v1.presign_model_upload_url(
@@ -668,25 +678,27 @@ class BentoCloudClient(CloudClient):
 
         def io_cb(x: int):
             with io_mutex:
-                self.transmission_progress.update(upload_task_id, advance=x)
+                self.spinner.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 model "{model.tag}"..'):
+            with self.spinner.spin(
+                text=f'Creating tar archive for model "{model.tag}"..'
+            ):
                 with tarfile.open(fileobj=tar_io, mode="w:") as tar:
                     tar.add(model.path, arcname="./")
             tar_io.seek(0, 0)
-            with self.spin(text=f'Start uploading model "{model.tag}"..'):
+            with self.spinner.spin(text=f'Start uploading model "{model.tag}"..'):
                 yatai_rest_client.v1.start_upload_model(
                     model_repository_name=model_repository.name, version=version
                 )
             file_size = tar_io.getbuffer().nbytes
-            self.transmission_progress.update(
+            self.spinner.transmission_progress.update(
                 upload_task_id,
                 description=f'Uploading model "{model.tag}"',
                 total=file_size,
                 visible=True,
             )
-            self.transmission_progress.start_task(upload_task_id)
+            self.spinner.transmission_progress.start_task(upload_task_id)
 
             if transmission_strategy == "proxy":
                 try:
@@ -696,11 +708,11 @@ class BentoCloudClient(CloudClient):
                         data=tar_io,
                     )
                 except Exception as e:  # pylint: disable=broad-except
-                    self.log_progress.add_task(
+                    self.spinner.log_progress.add_task(
                         f'[bold red]Failed to upload model "{model.tag}"'
                     )
                     raise e
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold green]Successfully pushed model "{model.tag}"'
                 )
                 return
@@ -717,7 +729,7 @@ class BentoCloudClient(CloudClient):
                             reason=resp.text,
                         )
                 else:
-                    with self.spin(
+                    with self.spinner.spin(
                         text=f'Start multipart uploading Model "{model.tag}"...'
                     ):
                         remote_model = (
@@ -738,7 +750,7 @@ class BentoCloudClient(CloudClient):
                     def chunk_upload(
                         upload_id: str, chunk_number: int
                     ) -> FinishUploadModelSchema | tuple[str, int]:
-                        with self.spin(
+                        with self.spinner.spin(
                             text=f'({chunk_number}/{chunks_count}) Presign multipart upload url of model "{model.tag}"...'
                         ):
                             remote_model = (
@@ -752,7 +764,7 @@ class BentoCloudClient(CloudClient):
                                 )
                             )
 
-                        with self.spin(
+                        with self.spinner.spin(
                             text=f'({chunk_number}/{chunks_count}) Uploading chunk of model "{model.tag}"...'
                         ):
                             chunk = (
@@ -809,7 +821,7 @@ class BentoCloudClient(CloudClient):
                                 )
                             )
 
-                    with self.spin(
+                    with self.spinner.spin(
                         text=f'Completing multipart upload of model "{model.tag}"...'
                     ):
                         remote_model = (
@@ -829,10 +841,12 @@ class BentoCloudClient(CloudClient):
                     reason=str(e),
                 )
             if finish_req.status is ModelUploadStatus.FAILED:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold red]Failed to upload model "{model.tag}"'
                 )
-            with self.spin(text="Submitting upload status to remote model store"):
+            with self.spinner.spin(
+                text="Submitting upload status to remote model store"
+            ):
                 yatai_rest_client.v1.finish_upload_model(
                     model_repository_name=model_repository.name,
                     version=version,
@@ -840,11 +854,11 @@ class BentoCloudClient(CloudClient):
                 )
 
             if finish_req.status != ModelUploadStatus.SUCCESS:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold red]Failed pushing model "{model.tag}" : {finish_req.reason}'
                 )
             else:
-                self.log_progress.add_task(
+                self.spinner.log_progress.add_task(
                     f'[bold green]Successfully pushed model "{model.tag}"'
                 )
 
@@ -858,8 +872,8 @@ class BentoCloudClient(CloudClient):
         model_store: ModelStore = Provide[BentoMLContainer.model_store],
         query: str | None = None,
     ) -> Model:
-        with Live(self.progress_group):
-            download_task_id = self.transmission_progress.add_task(
+        with Live(self.spinner.progress_group):
+            download_task_id = self.spinner.transmission_progress.add_task(
                 f'Pulling model "{tag}"', start=False, visible=False
             )
             return self._do_pull_model(
@@ -890,7 +904,7 @@ class BentoCloudClient(CloudClient):
         else:
             if _tag.version not in (None, "latest"):
                 if not force:
-                    self.log_progress.add_task(
+                    self.spinner.log_progress.add_task(
                         f'[bold blue]Model "{tag}" already exists locally, skipping'
                     )
                     return model
@@ -907,13 +921,13 @@ class BentoCloudClient(CloudClient):
                 )
             if model is not None:
                 if not force and latest_model.build_at < model.creation_time:
-                    self.log_progress.add_task(
+                    self.spinner.log_progress.add_task(
                         f'[bold blue]Newer version of model "{name}" exists locally, skipping'
                     )
                     return model
                 if model.tag.version == latest_model.version:
                     if not force:
-                        self.log_progress.add_task(
+                        self.spinner.log_progress.add_task(
                             f'[bold blue]Model "{model.tag}" already exists locally, skipping'
                         )
                         return model
@@ -925,7 +939,9 @@ class BentoCloudClient(CloudClient):
                 "`query` is ignored when model version is specified", UserWarning
             )
 
-        with self.spin(text=f'Getting a presigned download url for model "{_tag}"..'):
+        with self.spinner.spin(
+            text=f'Getting a presigned download url for model "{_tag}"..'
+        ):
             remote_model = yatai_rest_client.v1.presign_model_download_url(
                 name, version
             )
@@ -940,7 +956,9 @@ class BentoCloudClient(CloudClient):
         if remote_model.transmission_strategy is not None:
             transmission_strategy = remote_model.transmission_strategy
         else:
-            with self.spin(text=f'Getting a presigned download url for model "{_tag}"'):
+            with self.spinner.spin(
+                text=f'Getting a presigned download url for model "{_tag}"'
+            ):
                 remote_model = yatai_rest_client.v1.presign_model_download_url(
                     name, version
                 )
@@ -954,7 +972,7 @@ class BentoCloudClient(CloudClient):
             )
         else:
             if presigned_download_url is None:
-                with self.spin(
+                with self.spinner.spin(
                     text=f'Getting a presigned download url for model "{_tag}"'
                 ):
                     remote_model = yatai_rest_client.v1.presign_model_download_url(
@@ -971,25 +989,25 @@ class BentoCloudClient(CloudClient):
 
                 total_size_in_bytes = int(response.headers.get("content-length", 0))
                 block_size = 1024  # 1 Kibibyte
-                self.transmission_progress.update(
+                self.spinner.transmission_progress.update(
                     download_task_id,
                     description=f'Downloading model "{_tag}"',
                     total=total_size_in_bytes,
                     visible=True,
                 )
-                self.transmission_progress.start_task(download_task_id)
+                self.spinner.transmission_progress.start_task(download_task_id)
                 for data in response.iter_bytes(block_size):
-                    self.transmission_progress.update(
+                    self.spinner.transmission_progress.update(
                         download_task_id, advance=len(data)
                     )
                     tar_file.write(data)
 
-            self.log_progress.add_task(
+            self.spinner.log_progress.add_task(
                 f'[bold green]Finished downloading model "{_tag}" files'
             )
             tar_file.seek(0, 0)
             tar = tarfile.open(fileobj=tar_file, mode="r")
-            with self.spin(text=f'Extracting model "{_tag}" tar file'):
+            with self.spinner.spin(text=f'Extracting model "{_tag}" tar file'):
                 with fs.open_fs("temp://") as temp_fs:
                     for member in tar.getmembers():
                         f = tar.extractfile(member)
@@ -1000,7 +1018,7 @@ class BentoCloudClient(CloudClient):
                             temp_fs.makedirs(str(p.parent), recreate=True)
                         temp_fs.writebytes(member.name, f.read())
                     model = Model.from_fs(temp_fs).save(model_store)
-                    self.log_progress.add_task(
+                    self.spinner.log_progress.add_task(
                         f'[bold green]Successfully pulled model "{_tag}"'
                     )
                     return model
diff --git a/src/bentoml/_internal/cloud/client.py b/src/bentoml/_internal/cloud/client.py
index 84429f7e..f286a929 100644
--- a/src/bentoml/_internal/cloud/client.py
+++ b/src/bentoml/_internal/cloud/client.py
@@ -7,6 +7,7 @@ from urllib.parse import urljoin
 import httpx
 
 from ...exceptions import CloudRESTApiClientError
+from ...exceptions import NotFound
 from ..configuration import BENTOML_VERSION
 from .schemas.schemasv1 import BentoRepositorySchema
 from .schemas.schemasv1 import BentoSchema
@@ -28,6 +29,7 @@ from .schemas.schemasv1 import ModelSchema
 from .schemas.schemasv1 import ModelWithRepositoryListSchema
 from .schemas.schemasv1 import OrganizationSchema
 from .schemas.schemasv1 import PreSignMultipartUploadUrlSchema
+from .schemas.schemasv1 import ResourceInstanceSchema
 from .schemas.schemasv1 import UpdateBentoSchema
 from .schemas.schemasv1 import UpdateDeploymentSchema
 from .schemas.schemasv1 import UserSchema
@@ -556,7 +558,7 @@ class RestApiClientV2(BaseRestApiClient):
         name: str,
         update_schema: UpdateDeploymentSchemaV2,
         cluster: str | None = None,
-    ) -> DeploymentFullSchemaV2 | None:
+    ) -> DeploymentFullSchemaV2:
         url = urljoin(
             self.endpoint,
             f"/api/v2/deployments/{name}",
@@ -564,7 +566,7 @@ class RestApiClientV2(BaseRestApiClient):
         data = schema_to_json(update_schema)
         resp = self.session.put(url, content=data, params={"cluster": cluster})
         if self._is_not_found(resp):
-            return None
+            raise NotFound(f"Deployment {name} is not found: {resp.text}")
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
@@ -572,14 +574,14 @@ class RestApiClientV2(BaseRestApiClient):
         self,
         name: str,
         cluster: str | None = None,
-    ) -> DeploymentFullSchemaV2 | None:
+    ) -> DeploymentFullSchemaV2:
         url = urljoin(
             self.endpoint,
             f"/api/v2/deployments/{name}",
         )
         resp = self.session.get(url, params={"cluster": cluster})
         if self._is_not_found(resp):
-            return None
+            raise NotFound(f"Deployment {name} is not found: {resp.text}")
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
@@ -592,7 +594,7 @@ class RestApiClientV2(BaseRestApiClient):
         q: str | None = None,
         search: str | None = None,
         start: int | None = None,
-    ) -> DeploymentListSchemaV2 | None:
+    ) -> DeploymentListSchemaV2:
         url = urljoin(self.endpoint, "/api/v2/deployments")
         resp = self.session.get(
             url,
@@ -606,7 +608,7 @@ class RestApiClientV2(BaseRestApiClient):
             },
         )
         if self._is_not_found(resp):
-            return None
+            raise NotFound(f"Deployment not found: {resp.text}")
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentListSchemaV2)
 
@@ -614,14 +616,14 @@ class RestApiClientV2(BaseRestApiClient):
         self,
         name: str,
         cluster: str | None = None,
-    ) -> DeploymentFullSchemaV2 | None:
+    ) -> DeploymentFullSchemaV2:
         url = urljoin(
             self.endpoint,
             f"/api/v2/deployments/{name}/terminate",
         )
         resp = self.session.post(url, params={"cluster": cluster})
         if self._is_not_found(resp):
-            return None
+            raise NotFound(f"Deployment {name} is not found: {resp.text}")
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
@@ -629,21 +631,35 @@ class RestApiClientV2(BaseRestApiClient):
         self,
         name: str,
         cluster: str | None = None,
-    ) -> DeploymentFullSchemaV2 | None:
+    ) -> DeploymentFullSchemaV2:
         url = urljoin(
             self.endpoint,
             f"/api/v2/deployments/{name}",
         )
         resp = self.session.delete(url, params={"cluster": cluster})
         if self._is_not_found(resp):
-            return None
+            raise NotFound(f"Deployment {name} is not found: {resp.text}")
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
+    def list_instance_types(
+        self,
+        cluster: str | None = None,
+    ) -> list[ResourceInstanceSchema]:
+        url = urljoin(
+            self.endpoint,
+            "/api/v1/instance_types",
+        )
+        resp = self.session.get(url, params={"cluster": cluster})
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, list[ResourceInstanceSchema])
+
 
 class RestApiClient:
     def __init__(self, endpoint: str, api_token: str) -> None:
-        self.session = httpx.Client()
+        self.session = httpx.Client(timeout=60)
         self.session.headers.update(
             {
                 "X-YATAI-API-TOKEN": api_token,
diff --git a/src/bentoml/_internal/cloud/deployment.py b/src/bentoml/_internal/cloud/deployment.py
index ea16c28b..935028c1 100644
--- a/src/bentoml/_internal/cloud/deployment.py
+++ b/src/bentoml/_internal/cloud/deployment.py
@@ -3,13 +3,18 @@ from __future__ import annotations
 import logging
 import time
 import typing as t
+from os import path
 
 import attr
+import click
 import yaml
 from deepmerge.merger import Merger
+from rich.progress import TaskID
 from simple_di import Provide
 from simple_di import inject
 
+from bentoml._internal.cloud.base import Spinner
+
 if t.TYPE_CHECKING:
     from _bentoml_impl.client import AsyncHTTPClient
     from _bentoml_impl.client import SyncHTTPClient
@@ -19,12 +24,12 @@ if t.TYPE_CHECKING:
 
 from ...exceptions import BentoMLException
 from ...exceptions import NotFound
+from ..bento import Bento
 from ..configuration.containers import BentoMLContainer
 from ..tag import Tag
 from ..utils import bentoml_cattr
 from ..utils import resolve_user_filepath
 from .config import get_rest_api_client
-from .schemas.modelschemas import AccessControl
 from .schemas.modelschemas import DeploymentStatus
 from .schemas.modelschemas import DeploymentTargetHPAConf
 from .schemas.schemasv2 import CreateDeploymentSchema as CreateDeploymentSchemaV2
@@ -44,6 +49,188 @@ config_merger = Merger(
 )
 
 
+@attr.define
+class DeploymentConfigParameters:
+    name: str | None = None
+    path_context: str | None = None
+    context: str | None = None
+    bento: Tag | str | None = None
+    cluster: str | None = None
+    access_authorization: bool | None = None
+    scaling_min: int | None = None
+    scaling_max: int | None = None
+    instance_type: str | None = None
+    strategy: str | None = None
+    envs: t.List[dict[str, t.Any]] | None = None
+    extras: dict[str, t.Any] | None = None
+    config_dict: dict[str, t.Any] | None = None
+    config_file: str | t.TextIO | None = None
+    cli: bool = False
+    service_name: str | None = None
+    cfg_dict: dict[str, t.Any] | None = None
+    _param_config: dict[str, t.Any] | None = None
+
+    def verify(
+        self,
+    ):
+        deploy_by_param = (
+            self.name
+            or self.bento
+            or self.cluster
+            or self.access_authorization
+            or self.scaling_min
+            or self.scaling_max
+            or self.instance_type
+            or self.strategy
+            or self.envs
+            or self.extras
+        )
+
+        if (
+            self.config_dict
+            and self.config_file
+            or self.config_dict
+            and deploy_by_param
+            or self.config_file
+            and deploy_by_param
+        ):
+            raise BentoMLException(
+                "Configure a deployment can only use one of the following: config_dict, config_file, or the other parameters"
+            )
+
+        if deploy_by_param:
+            self.cfg_dict = {
+                k: v
+                for k, v in [
+                    ("name", self.name),
+                    ("bento", self.bento),
+                    ("cluster", self.cluster),
+                    ("access_authorization", self.access_authorization),
+                    ("envs", self.envs),
+                ]
+                if v is not None
+            }
+            # add service name
+            self._param_config = {
+                k: v
+                for k, v in [
+                    ("scaling_min", self.scaling_min),
+                    ("scaling_max", self.scaling_max),
+                    ("instance_type", self.instance_type),
+                    ("strategy", self.strategy),
+                    ("extras", self.extras),
+                ]
+                if v is not None
+            }
+        elif self.config_dict:
+            self.cfg_dict = self.config_dict
+        elif isinstance(self.config_file, str):
+            real_path = resolve_user_filepath(self.config_file, self.path_context)
+            try:
+                with open(real_path, "r") as file:
+                    self.cfg_dict = yaml.safe_load(file)
+            except FileNotFoundError:
+                raise ValueError(f"File not found: {real_path}")
+            except yaml.YAMLError as exc:
+                logger.error("Error while parsing YAML file: %s", exc)
+                raise
+            except Exception as e:
+                raise ValueError(
+                    f"An error occurred while reading the file: {real_path}\n{e}"
+                )
+        elif self.config_file:
+            try:
+                self.cfg_dict = yaml.safe_load(self.config_file)
+            except yaml.YAMLError as exc:
+                logger.error("Error while parsing YAML config-file stream: %s", exc)
+                raise
+        else:
+            raise BentoMLException(
+                "Must provide either config_dict, config_file, or the other parameters"
+            )
+
+        bento_name = self.cfg_dict.get("bento")
+        # determine if bento is a path or a name
+        if bento_name:
+            if isinstance(bento_name, str) and path.exists(bento_name):
+                # target is a path
+                if self.cli:
+                    click.echo(f"building bento from {bento_name} ...")
+                bento_obj = get_real_bento(
+                    project_path=bento_name,
+                    context=self.context,
+                )
+            else:
+                if self.cli:
+                    click.echo(f"using bento {bento_name}...")
+                bento_obj = get_real_bento(
+                    bento=str(bento_name),
+                    context=self.context,
+                )
+            self.cfg_dict["bento"] = bento_obj.tag
+            if self.service_name is None:
+                self.service_name = bento_obj.info.name
+
+    def get_name(self):
+        if self.cfg_dict is None:
+            raise BentoMLException(
+                "DeploymentConfigParameters.verify() must be called first"
+            )
+        return self.cfg_dict.get("name")
+
+    def get_cluster(self, pop: bool = True):
+        if self.cfg_dict is None:
+            raise BentoMLException(
+                "DeploymentConfigParameters.verify() must be called first"
+            )
+        if pop:
+            return self.cfg_dict.pop("cluster", None)
+        else:
+            return self.cfg_dict.get("cluster")
+
+    def get_config_dict(self, bento: str | None = None):
+        if self.cfg_dict is None:
+            raise BentoMLException(
+                "DeploymentConfigParameters.verify() must be called first"
+            )
+        if self.service_name is None:
+            if bento is None:
+                if self.cfg_dict.get("bento") is None:
+                    raise BentoMLException(
+                        "DeploymentConfigParameters.verify() must be called first"
+                    )
+                bento = self.cfg_dict.get("bento")
+            bento_obj = get_real_bento(
+                bento=bento,
+                context=self.context,
+            )
+            if len(bento_obj.info.services) > 0:
+                self.service_name = bento_obj.info.name
+            else:
+                # for compatibility
+                self.service_name = "apiserver"
+        if self._param_config is not None:
+            scaling_min = self._param_config.pop("scaling_min", None)
+            scaling_max = self._param_config.pop("scaling_max", None)
+            if scaling_min is not None or scaling_max is not None:
+                self._param_config["scaling"] = {
+                    "min_replicas": scaling_min,
+                    "max_replicas": scaling_max,
+                }
+                self._param_config["scaling"] = {
+                    k: v
+                    for k, v in self._param_config["scaling"].items()
+                    if v is not None
+                }
+
+            strategy = self._param_config.pop("strategy", None)
+            if strategy is not None:
+                self._param_config["deployment_strategy"] = strategy
+            self.cfg_dict["services"] = {self.service_name: self._param_config}
+
+        return self.cfg_dict
+
+
 def get_args_from_config(
     name: str | None = None,
     bento: Tag | str | None = None,
@@ -94,19 +281,19 @@ def get_args_from_config(
 
 
 @inject
-def get_real_bento_tag(
+def get_real_bento(
     project_path: str | None = None,
     bento: str | Tag | None = None,
     context: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
-) -> Tag:
+) -> Bento:
     if project_path:
         from bentoml.bentos import build_bentofile
 
         bento_obj = build_bentofile(build_ctx=project_path, _bento_store=_bento_store)
         _cloud_client.push_bento(bento=bento_obj, context=context)
-        return bento_obj.tag
+        return bento_obj
     elif bento:
         bento = Tag.from_taglike(bento)
         try:
@@ -117,11 +304,17 @@ def get_real_bento_tag(
                 raise e
             bento_obj = None
 
-        # try to push if bento exists, otherwise expects bentocloud to have it
-        if bento_obj:
+        if bento_obj is None:
+            # try to pull from bentocloud
+            try:
+                bento_obj = _cloud_client.pull_bento(context=context, tag=bento)
+            except NotFound as e:
+                raise e
+            return bento_obj
+        else:
+            # push to bentocloud
             _cloud_client.push_bento(bento=bento_obj, context=context)
-            bento = bento_obj.tag
-        return bento
+            return bento_obj
     else:
         raise BentoMLException(
             "Create a deployment needs a target; project path or bento is necessary"
@@ -130,11 +323,18 @@ def get_real_bento_tag(
 
 @attr.define
 class DeploymentConfig(CreateDeploymentSchemaV2):
-    def to_yaml(self):
-        return yaml.dump(bentoml_cattr.unstructure(self))
+    def to_yaml(self, with_meta: bool = True):
+        config_dict = self.to_dict(with_meta=with_meta)
+        return yaml.dump(config_dict, sort_keys=False)
 
-    def to_dict(self):
-        return bentoml_cattr.unstructure(self)
+    def to_dict(self, with_meta: bool = True):
+        config_dict = bentoml_cattr.unstructure(self)
+        if with_meta is False:
+            # delete name and bento
+            config_dict.pop("name", None)
+            config_dict.pop("bento", None)
+
+        return config_dict
 
 
 @attr.define
@@ -156,9 +356,6 @@ class DeploymentInfo:
     created_at: str
     created_by: str
     cluster: str
-    organization: str
-    distributed: bool
-    description: t.Optional[str]
     _context: t.Optional[str] = attr.field(alias="_context", repr=False)
     _schema: DeploymentSchema = attr.field(alias="_schema", repr=False)
     _urls: t.Optional[list[str]] = attr.field(alias="_urls", default=None, repr=False)
@@ -166,17 +363,18 @@ class DeploymentInfo:
     def to_dict(self) -> dict[str, t.Any]:
         return {
             "name": self.name,
+            "bento": self.get_bento(refetch=False),
             "cluster": self.cluster,
-            "description": self.description,
-            "organization": self.organization,
             "admin_console": self.admin_console,
             "created_at": self.created_at,
             "created_by": self.created_by,
-            "distributed": self.distributed,
-            "config": self.get_config(refetch=False).to_dict(),
+            "config": self.get_config(refetch=False).to_dict(with_meta=False),
             "status": self.get_status(refetch=False).to_dict(),
         }
 
+    def to_yaml(self):
+        return yaml.dump(self.to_dict(), sort_keys=False)
+
     def _refetch(self) -> None:
         res = Deployment.get(self.name, self.cluster, self._context)
         self._schema = res._schema
@@ -204,17 +402,9 @@ class DeploymentInfo:
         return DeploymentConfig(
             name=self.name,
             bento=self.get_bento(refetch=False),
-            distributed=self.distributed,
-            description=self.description,
             services=target.config.services,
-            instance_type=target.config.instance_type,
-            deployment_strategy=target.config.deployment_strategy,
-            scaling=target.config.scaling,
+            access_authorization=target.config.access_authorization,
             envs=target.config.envs,
-            extras=target.config.extras,
-            access_type=target.config.access_type,
-            bentoml_config_overrides=target.config.bentoml_config_overrides,
-            cold_start_timeout=target.config.cold_start_timeout,
         )
 
     def get_status(self, refetch: bool = True) -> DeploymentState:
@@ -264,19 +454,46 @@ class DeploymentInfo:
             raise BentoMLException("Deployment url is not ready")
         return AsyncHTTPClient(self._urls[0], media_type=media_type, token=token)
 
-    def wait_until_ready(self, timeout: int = 3600, check_interval: int = 30) -> None:
+    def wait_until_ready(
+        self,
+        timeout: int = 3600,
+        check_interval: int = 10,
+        spinner: Spinner | None = None,
+        spinner_task_id: TaskID | None = None,
+    ) -> None:
         start_time = time.time()
-        while time.time() - start_time < timeout:
-            status = self.get_status()
-            if status.status == DeploymentStatus.Running.value:
-                logger.info(
-                    f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Deployment '{self.name}' is ready."
+        if spinner is not None and spinner_task_id is not None:
+            while time.time() - start_time < timeout:
+                status = self.get_status()
+                spinner.spinner_progress.update(
+                    spinner_task_id,
+                    action=f"Waiting for deployment '{self.name}' to be ready. Current status: '{status.status}'.",
                 )
-                return
-            logger.info(
-                f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Waiting for deployment '{self.name}' to be ready. Current status: '{status}'."
+                if status.status == DeploymentStatus.Running.value:
+                    spinner.spinner_progress.update(
+                        spinner_task_id,
+                        action=f'[bold green] Deployment "{self.name}" is ready.[/bold green]',
+                    )
+                    spinner.spinner_progress.stop_task(spinner_task_id)
+                    return
+                time.sleep(check_interval)
+            spinner.spinner_progress.update(
+                spinner_task_id,
+                action=f'[bold red]Time out waiting for Deployment "{self.name}" ready.[/bold red]',
             )
-            time.sleep(check_interval)
+            spinner.spinner_progress.stop_task(spinner_task_id)
+        else:
+            while time.time() - start_time < timeout:
+                status = self.get_status()
+                if status.status == DeploymentStatus.Running.value:
+                    logger.info(
+                        f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Deployment '{self.name}' is ready."
+                    )
+                    return
+                logger.info(
+                    f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Waiting for deployment '{self.name}' to be ready. Current status: '{status.status}'."
+                )
+                time.sleep(check_interval)
 
         raise TimeoutError(
             f"Timed out waiting for deployment '{self.name}' to be ready."
@@ -297,9 +514,8 @@ class Deployment:
             scaling.max_replicas = max(scaling.min_replicas, 1)
         # one edge case:
         if scaling.min_replicas > scaling.max_replicas:
-            scaling.min_replicas = scaling.max_replicas
-            logger.warning(
-                "min scaling value is greater than max scaling value, setting min scaling to max scaling value"
+            raise BentoMLException(
+                "min scaling values must be less than or equal to max scaling values"
             )
         if scaling.min_replicas < 0:
             raise BentoMLException(
@@ -309,55 +525,14 @@ class Deployment:
             raise BentoMLException("max scaling values must be greater than 0")
         return scaling
 
-    @staticmethod
-    def _validate_input_on_distributed(
-        config_struct: UpdateDeploymentSchemaV2, distributed: bool
-    ) -> None:
-        if distributed:
-            if config_struct.instance_type is not None:
-                raise BentoMLException(
-                    "The 'instance_type' field is not allowed for distributed deployments. Please specify it per service in the services field."
-                )
-            if (
-                config_struct.scaling is not None
-                and config_struct.scaling != DeploymentTargetHPAConf()
-            ):
-                raise BentoMLException(
-                    "The 'scaling' field is not allowed for distributed deployments. Please specify it per service in the services field."
-                )
-            if config_struct.deployment_strategy is not None:
-                raise BentoMLException(
-                    "The 'deployment_strategy' field is not allowed for distributed deployments. Please specify it per service in the services field."
-                )
-            if config_struct.extras is not None:
-                raise BentoMLException(
-                    "The 'extras' field is not allowed for distributed deployments. Please specify it per service in the services field."
-                )
-            if config_struct.cold_start_timeout is not None:
-                raise BentoMLException(
-                    "The 'cold_start_timeout' field is not allowed for distributed deployments. Please specify it per service in the services field."
-                )
-        elif not distributed:
-            if config_struct.services != {}:
-                raise BentoMLException(
-                    "The 'services' field is only allowed for distributed deployments."
-                )
-
     @classmethod
     def _fix_and_validate_schema(
-        cls, config_struct: UpdateDeploymentSchemaV2, distributed: bool
+        cls,
+        config_struct: UpdateDeploymentSchemaV2,
     ):
-        cls._validate_input_on_distributed(config_struct, distributed)
         # fix scaling
-        if distributed:
-            if len(config_struct.services) == 0:
-                raise BentoMLException("The configuration for services is mandatory")
-            for _, svc in config_struct.services.items():
-                svc.scaling = cls._fix_scaling(svc.scaling)
-        else:
-            config_struct.scaling = cls._fix_scaling(config_struct.scaling)
-        if config_struct.access_type is None:
-            config_struct.access_type = AccessControl.PUBLIC
+        for _, svc in config_struct.services.items():
+            svc.scaling = cls._fix_scaling(svc.scaling)
 
     @staticmethod
     def _convert_schema_to_update_schema(_schema: DeploymentSchema) -> dict[str, t.Any]:
@@ -376,16 +551,11 @@ class Deployment:
             raise BentoMLException(f"Deployment {_schema.name} has no bento")
         update_schema = UpdateDeploymentSchemaV2(
             services=target_schema.config.services,
-            instance_type=target_schema.config.instance_type,
-            deployment_strategy=target_schema.config.deployment_strategy,
-            scaling=target_schema.config.scaling,
+            access_authorization=target_schema.config.access_authorization,
             envs=target_schema.config.envs,
-            extras=target_schema.config.extras,
-            access_type=target_schema.config.access_type,
-            bentoml_config_overrides=target_schema.config.bentoml_config_overrides,
             bento=target_schema.bento.repository.name + ":" + target_schema.bento.name,
-            cold_start_timeout=target_schema.config.cold_start_timeout,
         )
+        print("update_schema", update_schema)
         return bentoml_cattr.unstructure(update_schema)
 
     @staticmethod
@@ -416,9 +586,6 @@ class Deployment:
             created_at=res.created_at.strftime("%Y-%m-%d %H:%M:%S"),
             created_by=res.creator.name,
             cluster=cluster_display_name,
-            organization=res.cluster.organization_name,
-            distributed=res.distributed,
-            description=res.description,
             _schema=res,
             _context=context,
             _urls=urls,
@@ -434,100 +601,35 @@ class Deployment:
         cloud_rest_client = get_rest_api_client(context)
         if cluster is None:
             res_count = cloud_rest_client.v2.list_deployment(all=True, search=search)
-            if res_count is None:
-                raise BentoMLException("List deployments request failed")
             if res_count.total == 0:
                 return []
             res = cloud_rest_client.v2.list_deployment(
                 search=search, count=res_count.total, all=True
             )
-            if res is None:
-                raise BentoMLException("List deployments request failed")
         else:
             res_count = cloud_rest_client.v2.list_deployment(cluster, search=search)
-            if res_count is None:
-                raise NotFound(f"Cluster {cluster} is not found")
             if res_count.total == 0:
                 return []
             res = cloud_rest_client.v2.list_deployment(
                 cluster, search=search, count=res_count.total
             )
-            if res is None:
-                raise BentoMLException("List deployments request failed")
         return [cls._generate_deployment_info_(context, schema) for schema in res.items]
 
     @classmethod
     def create(
         cls,
-        bento: Tag | str | None = None,
-        access_type: str | None = None,
-        name: str | None = None,
-        cluster: str | None = None,
-        scaling_min: int | None = None,
-        scaling_max: int | None = None,
-        instance_type: str | None = None,
-        strategy: str | None = None,
-        envs: t.List[dict[str, t.Any]] | None = None,
-        extras: dict[str, t.Any] | None = None,
-        config_dict: dict[str, t.Any] | None = None,
-        config_file: str | t.TextIO | None = None,
-        path_context: str | None = None,
+        deployment_config_params: DeploymentConfigParameters,
         context: str | None = None,
     ) -> DeploymentInfo:
-        cloud_rest_client = get_rest_api_client(context)
-        dict: dict[str, t.Any] = {
-            "bento": str(bento),
-        }
-        if name:
-            dict["name"] = name
-        else:
-            # the cloud takes care of the name
-            dict["name"] = ""
-
-        if config_dict:
-            merging_dict = config_dict
-            pass
-        elif isinstance(config_file, str):
-            real_path = resolve_user_filepath(config_file, path_context)
-            try:
-                with open(real_path, "r") as file:
-                    merging_dict = yaml.safe_load(file)
-            except FileNotFoundError:
-                raise ValueError(f"File not found: {real_path}")
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML file: %s", exc)
-                raise
-            except Exception as e:
-                raise ValueError(
-                    f"An error occurred while reading the file: {real_path}\n{e}"
-                )
-        elif config_file is not None:
-            try:
-                merging_dict = yaml.safe_load(config_file)
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML config-file stream: %s", exc)
-                raise
-        else:
-            merging_dict = {
-                "scaling": {"min_replicas": scaling_min, "max_replicas": scaling_max},
-                "instance_type": instance_type,
-                "deployment_strategy": strategy,
-                "envs": envs,
-                "extras": extras,
-                "access_type": access_type,
-            }
-        dict.update(merging_dict)
+        cluster = deployment_config_params.get_cluster()
+        config_params = deployment_config_params.get_config_dict()
+        if config_params.get("bento") is None:
+            raise ValueError("bento is required")
 
-        if "distributed" not in dict:
-            dict["distributed"] = (
-                "services" in dict
-                and dict["services"] is not None
-                and dict["services"] != {}
-            )
-
-        config_struct = bentoml_cattr.structure(dict, CreateDeploymentSchemaV2)
-        cls._fix_and_validate_schema(config_struct, dict["distributed"])
+        config_struct = bentoml_cattr.structure(config_params, CreateDeploymentSchemaV2)
+        cls._fix_and_validate_schema(config_struct)
 
+        cloud_rest_client = get_rest_api_client(context)
         res = cloud_rest_client.v2.create_deployment(
             create_schema=config_struct, cluster=cluster
         )
@@ -539,164 +641,72 @@ class Deployment:
     @classmethod
     def update(
         cls,
-        name: str | None,
-        bento: Tag | str | None = None,
-        access_type: str | None = None,
-        cluster: str | None = None,
-        scaling_min: int | None = None,
-        scaling_max: int | None = None,
-        instance_type: str | None = None,
-        strategy: str | None = None,
-        envs: t.List[dict[str, t.Any]] | None = None,
-        extras: dict[str, t.Any] | None = None,
-        config_dict: dict[str, t.Any] | None = None,
-        config_file: str | t.TextIO | None = None,
-        path_context: str | None = None,
+        deployment_config_params: DeploymentConfigParameters,
         context: str | None = None,
     ) -> DeploymentInfo:
+        name = deployment_config_params.get_name()
         if name is None:
             raise ValueError("name is required")
+        cluster = deployment_config_params.get_cluster()
+
         cloud_rest_client = get_rest_api_client(context)
         deployment_schema = cloud_rest_client.v2.get_deployment(name, cluster)
-        if deployment_schema is None:
-            raise NotFound(f"deployment {name} is not found")
-
         orig_dict = cls._convert_schema_to_update_schema(deployment_schema)
-        distributed = deployment_schema.distributed
-        if bento:
-            orig_dict["bento"] = str(bento)
-
-        if config_dict:
-            merging_dict = config_dict
-            pass
-        elif isinstance(config_file, str):
-            real_path = resolve_user_filepath(config_file, path_context)
-            try:
-                with open(real_path, "r") as file:
-                    merging_dict = yaml.safe_load(file)
-            except FileNotFoundError:
-                raise ValueError(f"File not found: {real_path}")
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML file: %s", exc)
-                raise
-            except Exception as e:
-                raise ValueError(
-                    f"An error occurred while reading the file: {real_path}\n{e}"
-                )
-        elif config_file is not None:
-            try:
-                merging_dict = yaml.safe_load(config_file)
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML config-file stream: %s", exc)
-                raise
-
-        else:
-            merging_dict: dict[str, t.Any] = {"scaling": {}}
-            if scaling_min is not None:
-                merging_dict["scaling"]["min_replicas"] = scaling_min
-            if scaling_max is not None:
-                merging_dict["scaling"]["max_replicas"] = scaling_max
-            if instance_type is not None:
-                merging_dict["instance_type"] = instance_type
-
-            if strategy is not None:
-                merging_dict["deployment_strategy"] = strategy
-
-            if envs is not None:
-                merging_dict["envs"] = envs
-
-            if extras is not None:
-                merging_dict["extras"] = extras
 
-            if access_type is not None:
-                merging_dict["access_type"] = access_type
-
-        config_merger.merge(orig_dict, merging_dict)
+        config_params = deployment_config_params.get_config_dict(orig_dict.get("bento"))
 
+        print("1111111 orig: ", orig_dict)
+        print("1111111 config: ", config_params)
+        config_merger.merge(orig_dict, config_params)
+        print("2222 orig: ", orig_dict)
         config_struct = bentoml_cattr.structure(orig_dict, UpdateDeploymentSchemaV2)
 
-        cls._fix_and_validate_schema(config_struct, distributed)
+        cls._fix_and_validate_schema(config_struct)
 
         res = cloud_rest_client.v2.update_deployment(
             cluster=deployment_schema.cluster.host_cluster_display_name,
             name=name,
             update_schema=config_struct,
         )
-        if res is None:
-            raise NotFound(f"deployment {name} is not found")
         logger.debug("Deployment Schema: %s", config_struct)
         return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def apply(
         cls,
-        name: str | None = None,
-        bento: Tag | str | None = None,
-        cluster: str | None = None,
-        config_dict: dict[str, t.Any] | None = None,
-        config_file: t.TextIO | str | None = None,
-        path_context: str | None = None,
+        deployment_config_params: DeploymentConfigParameters,
         context: str | None = None,
     ) -> DeploymentInfo:
-        if name is None:
-            raise ValueError("name is required")
         cloud_rest_client = get_rest_api_client(context)
-        res = cloud_rest_client.v2.get_deployment(name, cluster)
-        if res is None:
-            if bento is not None:
+        name = deployment_config_params.get_name()
+        if name is not None:
+            try:
+                deployment_schema = cloud_rest_client.v2.get_deployment(
+                    name, deployment_config_params.get_cluster(pop=False)
+                )
+            except NotFound:
                 return cls.create(
-                    bento=bento,
-                    name=name,
-                    cluster=cluster,
-                    config_dict=config_dict,
-                    config_file=config_file,
-                    path_context=path_context,
+                    deployment_config_params=deployment_config_params,
                     context=context,
                 )
-            else:
-                raise ValueError("bento is required")
-        if bento is None:
-            bento = cls._convert_schema_to_bento(_schema=res)
-
-        schema_dict: dict[str, t.Any] = {"bento": str(bento)}
-        distributed = res.distributed
-
-        if isinstance(config_file, str):
-            real_path = resolve_user_filepath(config_file, path_context)
-            try:
-                with open(real_path, "r") as file:
-                    config_dict = yaml.safe_load(file)
-            except FileNotFoundError:
-                raise ValueError(f"File not found: {real_path}")
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML file: %s", exc)
-                raise
-            except Exception as e:
-                raise ValueError(
-                    f"An error occurred while reading the file: {real_path}\n{e}"
-                )
-        elif config_file is not None:
-            try:
-                config_dict = yaml.safe_load(config_file)
-            except yaml.YAMLError as exc:
-                logger.error("Error while parsing YAML config-file stream: %s", exc)
-                raise
-        if config_dict is None:
-            raise BentoMLException("Apply a deployment needs a configuration input")
+            # directly update the deployment with the schema, do not merge with the existing schema
+            config_struct = bentoml_cattr.structure(
+                deployment_config_params.get_config_dict(), UpdateDeploymentSchemaV2
+            )
+            cls._fix_and_validate_schema(config_struct)
 
-        schema_dict.update(config_dict)
-        config_struct = bentoml_cattr.structure(schema_dict, UpdateDeploymentSchemaV2)
-        cls._fix_and_validate_schema(config_struct, distributed)
+            res = cloud_rest_client.v2.update_deployment(
+                name=name,
+                update_schema=config_struct,
+                cluster=deployment_schema.cluster.host_cluster_display_name,
+            )
+            logger.debug("Deployment Schema: %s", config_struct)
+            return cls._generate_deployment_info_(context, res, res.urls)
 
-        res = cloud_rest_client.v2.update_deployment(
-            name=name,
-            update_schema=config_struct,
-            cluster=res.cluster.host_cluster_display_name,
+        return cls.create(
+            deployment_config_params=deployment_config_params,
+            context=context,
         )
-        if res is None:
-            raise NotFound(f"deployment {name} is not found")
-        logger.debug("Deployment Schema: %s", config_struct)
-        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def get(
@@ -707,8 +717,6 @@ class Deployment:
     ) -> DeploymentInfo:
         cloud_rest_client = get_rest_api_client(context)
         res = cloud_rest_client.v2.get_deployment(name, cluster)
-        if res is None:
-            raise NotFound(f"deployment {name} is not found")
         return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
@@ -720,8 +728,6 @@ class Deployment:
     ) -> DeploymentInfo:
         cloud_rest_client = get_rest_api_client(context)
         res = cloud_rest_client.v2.terminate_deployment(name, cluster)
-        if res is None:
-            raise NotFound(f"Deployment {name} is not found")
         return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
@@ -732,6 +738,39 @@ class Deployment:
         context: str | None = None,
     ) -> None:
         cloud_rest_client = get_rest_api_client(context)
-        res = cloud_rest_client.v2.delete_deployment(name, cluster)
-        if res is None:
-            raise NotFound(f"Deployment {name} is not found")
+        cloud_rest_client.v2.delete_deployment(name, cluster)
+
+    @classmethod
+    def list_instance_types(
+        cls,
+        cluster: str | None = None,
+        context: str | None = None,
+    ) -> list[InstanceTypeInfo]:
+        cloud_rest_client = get_rest_api_client(context)
+        res = cloud_rest_client.v2.list_instance_types(cluster)
+        return [
+            InstanceTypeInfo(
+                name=schema.display_name,
+                description=schema.description,
+                cpu=schema.config.resources.requests.cpu,
+                memory=schema.config.resources.requests.memory,
+                gpu=schema.config.resources.requests.gpu,
+                gpu_type=schema.config.gpu_config.type,
+                price=schema.config.price,
+            )
+            for schema in res
+        ]
+
+
+@attr.define
+class InstanceTypeInfo:
+    name: t.Optional[str] = None
+    price: t.Optional[str] = None
+    description: t.Optional[str] = None
+    cpu: t.Optional[str] = None
+    memory: t.Optional[str] = None
+    gpu: t.Optional[str] = None
+    gpu_type: t.Optional[str] = None
+
+    def to_dict(self):
+        return {k: v for k, v in attr.asdict(self).items() if v is not None and v != ""}
diff --git a/src/bentoml/_internal/cloud/schemas/modelschemas.py b/src/bentoml/_internal/cloud/schemas/modelschemas.py
index 79d28d7c..6fd10cd3 100644
--- a/src/bentoml/_internal/cloud/schemas/modelschemas.py
+++ b/src/bentoml/_internal/cloud/schemas/modelschemas.py
@@ -353,7 +353,6 @@ class DeploymentTargetConfig:
     )
     traffic_control: t.Optional[TrafficControlConfig] = attr.field(default=None)
     deployment_cold_start_wait_timeout: t.Optional[int] = attr.field(default=None)
-    bentoml_config_overrides: t.Optional[dict[str, t.Any]] = attr.field(default=None)
 
 
 @attr.define
@@ -378,6 +377,7 @@ class DeploymentServiceConfig:
     deployment_strategy: t.Optional[DeploymentStrategy] = attr.field(default=None)
     extras: t.Optional[ExtraDeploymentOverrides] = attr.field(default=None)
     cold_start_timeout: t.Optional[int] = attr.field(default=None)
+    config_overrides: t.Optional[t.Dict[str, t.Any]] = attr.field(factory=dict)
 
 
 class DeploymentStatus(Enum):
@@ -405,6 +405,13 @@ class ResourceInstanceConfigSchema:
     resources: DeploymentTargetResources
     price: str
     node_selectors: t.Optional[t.Dict[str, str]] = attr.field(factory=dict)
+    gpu_config: t.Optional[ResourceInstanceGPUConfigSchema] = attr.field(default=None)
+
+
+@attr.define
+class ResourceInstanceGPUConfigSchema:
+    type: str
+    memory: str
 
 
 class DeploymentRevisionStatus(Enum):
diff --git a/src/bentoml/_internal/cloud/schemas/schemasv1.py b/src/bentoml/_internal/cloud/schemas/schemasv1.py
index 77530792..4495c33e 100644
--- a/src/bentoml/_internal/cloud/schemas/schemasv1.py
+++ b/src/bentoml/_internal/cloud/schemas/schemasv1.py
@@ -257,7 +257,6 @@ class DeploymentSchema(ResourceSchema):
     cluster: ClusterSchema
     status: DeploymentStatus
     kube_namespace: str
-    distributed: bool = attr.field(default=False)
     latest_revision: t.Optional[DeploymentRevisionSchema] = attr.field(
         default=None
     )  # Delete returns no latest revision
@@ -328,8 +327,6 @@ class CreateDeploymentSchema(UpdateDeploymentSchema):
     __forbid_extra_keys__ = True
     name: str
     kube_namespace: str
-    # cluster: str
-    distributed: t.Optional[bool] = attr.field(default=False)
 
 
 @attr.define(kw_only=True)
diff --git a/src/bentoml/_internal/cloud/schemas/schemasv2.py b/src/bentoml/_internal/cloud/schemas/schemasv2.py
index 00cb7b1a..4f136ad6 100644
--- a/src/bentoml/_internal/cloud/schemas/schemasv2.py
+++ b/src/bentoml/_internal/cloud/schemas/schemasv2.py
@@ -4,7 +4,6 @@ import typing as t
 
 import attr
 
-from bentoml._internal.cloud.schemas.modelschemas import AccessControl
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentMode
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentRevisionStatus
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentServiceConfig
@@ -15,6 +14,8 @@ from bentoml._internal.cloud.schemas.schemasv1 import ClusterSchema
 from bentoml._internal.cloud.schemas.schemasv1 import ResourceSchema
 from bentoml._internal.cloud.schemas.schemasv1 import UserSchema
 
+from .modelschemas import EnvItemSchema
+
 
 @attr.define
 class DeploymentTargetSchema(ResourceSchema):
@@ -43,12 +44,12 @@ class DeploymentRevisionListSchema(BaseListSchema):
 
 
 @attr.define
-class DeploymentConfigSchema(DeploymentServiceConfig):
+class DeploymentConfigSchema:
     __omit_if_default__ = True
     __forbid_extra_keys__ = False
-    access_type: t.Optional[AccessControl] = attr.field(default=None)
+    access_authorization: bool = attr.field(default=False)
+    envs: t.Optional[t.List[t.Optional[EnvItemSchema]]] = attr.field(default=None)
     services: t.Dict[str, DeploymentServiceConfig] = attr.field(factory=dict)
-    bentoml_config_overrides: t.Dict[str, t.Any] = attr.field(factory=dict)
 
 
 @attr.define(kw_only=True)
@@ -56,16 +57,13 @@ class UpdateDeploymentSchema(DeploymentConfigSchema):
     __omit_if_default__ = True
     __forbid_extra_keys__ = False  # distributed, cluster and name need to be ignored
     bento: str
-    access_type: t.Optional[AccessControl] = attr.field(default=None)
-    description: t.Optional[str] = attr.field(default=None)
 
 
 @attr.define(kw_only=True)
 class CreateDeploymentSchema(UpdateDeploymentSchema):
     __omit_if_default__ = True
     __forbid_extra_keys__ = True
-    name: str
-    distributed: bool
+    name: t.Optional[str] = attr.field(default=None)
 
 
 @attr.define
@@ -78,8 +76,6 @@ class DeploymentSchema(ResourceSchema):
     cluster: ClusterSchema
     latest_revision: t.Optional[DeploymentRevisionSchema]
     mode: t.Optional[DeploymentMode] = attr.field(default=None)
-    distributed: bool = attr.field(default=False)
-    description: str = attr.field(default="")
 
 
 @attr.define
diff --git a/src/bentoml/_internal/configuration/__init__.py b/src/bentoml/_internal/configuration/__init__.py
index b3409016..8ece7700 100644
--- a/src/bentoml/_internal/configuration/__init__.py
+++ b/src/bentoml/_internal/configuration/__init__.py
@@ -110,7 +110,7 @@ def get_bentoml_override_config_json_from_env() -> dict[str, t.Any] | None:
     if CONFIG_OVERRIDE_JSON_ENV_VAR in os.environ:
         # User local config options for customizing bentoml
         values = os.environ.get(CONFIG_OVERRIDE_JSON_ENV_VAR, None)
-        if values is not None:
+        if values is not None and values != "":
             return json.loads(values)
 
     return None
diff --git a/src/bentoml/deployment.py b/src/bentoml/deployment.py
index afbd6f88..d30ce829 100644
--- a/src/bentoml/deployment.py
+++ b/src/bentoml/deployment.py
@@ -10,11 +10,9 @@ from simple_di import Provide
 from simple_di import inject
 
 from bentoml._internal.cloud.deployment import Deployment
+from bentoml._internal.cloud.deployment import DeploymentConfigParameters
 from bentoml._internal.cloud.deployment import DeploymentInfo
-from bentoml._internal.cloud.deployment import get_args_from_config
-from bentoml._internal.cloud.deployment import get_real_bento_tag
 from bentoml._internal.tag import Tag
-from bentoml.cloud import BentoCloudClient
 from bentoml.exceptions import BentoMLException
 
 from ._internal.configuration.containers import BentoMLContainer
@@ -23,26 +21,6 @@ if t.TYPE_CHECKING:
     from ._internal.bento import BentoStore
 
 
-@t.overload
-def create(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    cluster: str | None = ...,
-    access_type: str | None = ...,
-    scaling_min: int | None = ...,
-    scaling_max: int | None = ...,
-    instance_type: str | None = ...,
-    strategy: str | None = ...,
-    envs: t.List[dict[str, t.Any]] | None = ...,
-    extras: dict[str, t.Any] | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def create(
     name: str | None = ...,
@@ -52,7 +30,7 @@ def create(
     *,
     bento: Tag | str | None = ...,
     cluster: str | None = ...,
-    access_type: str | None = ...,
+    access_authorization: bool | None = ...,
     scaling_min: int | None = ...,
     scaling_max: int | None = ...,
     instance_type: str | None = ...,
@@ -76,19 +54,6 @@ def create(
     ...
 
 
-@t.overload
-def create(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    config_file: str | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def create(
     name: str | None = ...,
@@ -102,19 +67,6 @@ def create(
     ...
 
 
-@t.overload
-def create(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    config_dict: dict[str, t.Any] | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @inject
 def create(
     name: str | None = None,
@@ -122,10 +74,9 @@ def create(
     context: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
-    project_path: str | None = None,
     bento: Tag | str | None = None,
     cluster: str | None = None,
-    access_type: str | None = None,
+    access_authorization: bool | None = None,
     scaling_min: int | None = None,
     scaling_max: int | None = None,
     instance_type: str | None = None,
@@ -135,85 +86,34 @@ def create(
     config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
 ) -> DeploymentInfo:
-    deploy_by_param = (
-        access_type
-        or cluster
-        or scaling_min
-        or scaling_max
-        or instance_type
-        or strategy
-        or envs
-        or extras
-    )
-    if (
-        config_dict
-        and config_file
-        or config_dict
-        and deploy_by_param
-        or config_file
-        and deploy_by_param
-    ):
-        raise BentoMLException(
-            "Configure a deployment can only use one of the following: config_dict, config_file, or the other parameters"
-        )
-    deploy_name, bento_name, cluster_name = get_args_from_config(
-        bento=bento,
+    config_params = DeploymentConfigParameters(
         name=name,
-        cluster=cluster,
-        config_dict=config_dict,
-        config_file=config_file,
         path_context=path_context,
-    )
-
-    if bento_name and project_path:
-        raise BentoMLException("Only one of bento or project_path can be provided")
-    if bento_name is None and project_path is None:
-        raise BentoMLException("Either bento or project_path must be provided")
-    bento = get_real_bento_tag(
-        project_path=project_path,
-        bento=bento_name,
         context=context,
-        _cloud_client=BentoCloudClient(),
-    )
-
-    return Deployment.create(
         bento=bento,
-        access_type=access_type,
-        name=deploy_name,
-        cluster=cluster_name,
-        scaling_min=scaling_min,
+        cluster=cluster,
+        access_authorization=access_authorization,
         scaling_max=scaling_max,
+        scaling_min=scaling_min,
         instance_type=instance_type,
         strategy=strategy,
         envs=envs,
         extras=extras,
         config_dict=config_dict,
         config_file=config_file,
-        path_context=path_context,
+    )
+    try:
+        config_params.verify()
+    except BentoMLException as e:
+        raise BentoMLException(
+            f"Failed to create deployment due to invalid configuration: {e}"
+        )
+    return Deployment.create(
+        deployment_config_params=config_params,
         context=context,
     )
 
 
-@t.overload
-def update(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    cluster: str | None = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    access_type: str | None = ...,
-    scaling_min: int | None = ...,
-    scaling_max: int | None = ...,
-    instance_type: str | None = ...,
-    strategy: str | None = ...,
-    envs: t.List[dict[str, t.Any]] | None = ...,
-    extras: dict[str, t.Any] | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def update(
     name: str | None = ...,
@@ -223,7 +123,7 @@ def update(
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
-    access_type: str | None = ...,
+    access_authorization: bool | None = ...,
     scaling_min: int | None = ...,
     scaling_max: int | None = ...,
     instance_type: str | None = ...,
@@ -234,20 +134,6 @@ def update(
     ...
 
 
-@t.overload
-def update(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    cluster: str | None = None,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    config_file: str | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def update(
     name: str | None = ...,
@@ -262,20 +148,6 @@ def update(
     ...
 
 
-@t.overload
-def update(
-    name: str | None = ...,
-    path_context: str | None = ...,
-    context: str | None = ...,
-    cluster: str | None = None,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: str | None = ...,
-    config_dict: dict[str, t.Any] | None = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def update(
     name: str | None = ...,
@@ -298,9 +170,8 @@ def update(
     cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
-    project_path: str | None = None,
     bento: Tag | str | None = None,
-    access_type: str | None = None,
+    access_authorization: bool | None = None,
     scaling_min: int | None = None,
     scaling_max: int | None = None,
     instance_type: str | None = None,
@@ -310,76 +181,33 @@ def update(
     config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
 ) -> DeploymentInfo:
-    deploy_by_param = (
-        access_type
-        or scaling_min
-        or scaling_max
-        or instance_type
-        or strategy
-        or envs
-        or extras
-    )
-    if (
-        config_dict
-        and config_file
-        or config_dict
-        and deploy_by_param
-        or config_file
-        and deploy_by_param
-    ):
-        raise BentoMLException(
-            "Configure a deployment can only use one of the following: config_dict, config_file, or the other parameters"
-        )
-    deploy_name, bento_name, cluster_name = get_args_from_config(
-        bento=bento,
+    config_params = DeploymentConfigParameters(
         name=name,
-        cluster=cluster,
-        config_dict=config_dict,
-        config_file=config_file,
         path_context=path_context,
-    )
-    if bento_name and project_path:
-        raise BentoMLException("Only one of bento or project_path can be provided")
-    if bento_name is None and project_path is None:
-        bento = None
-    else:
-        bento = get_real_bento_tag(
-            project_path=project_path,
-            bento=bento_name,
-            context=context,
-            _cloud_client=BentoCloudClient(),
-        )
-
-    return Deployment.update(
+        context=context,
         bento=bento,
-        access_type=access_type,
-        name=deploy_name,
-        cluster=cluster_name,
-        scaling_min=scaling_min,
+        cluster=cluster,
+        access_authorization=access_authorization,
         scaling_max=scaling_max,
+        scaling_min=scaling_min,
         instance_type=instance_type,
         strategy=strategy,
         envs=envs,
         extras=extras,
         config_dict=config_dict,
         config_file=config_file,
-        path_context=path_context,
-        context=context,
     )
+    try:
+        config_params.verify()
+    except BentoMLException as e:
+        raise BentoMLException(
+            f"Failed to create deployment due to invalid configuration: {e}"
+        )
 
-
-@t.overload
-def apply(
-    name: str | None = ...,
-    cluster: t.Optional[str] = ...,
-    path_context: t.Optional[str] = ...,
-    context: t.Optional[str] = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: t.Optional[str] = ...,
-    config_dict: t.Optional[dict[str, t.Any]] = ...,
-) -> DeploymentInfo:
-    ...
+    return Deployment.update(
+        deployment_config_params=config_params,
+        context=context,
+    )
 
 
 @t.overload
@@ -396,20 +224,6 @@ def apply(
     ...
 
 
-@t.overload
-def apply(
-    name: str | None = ...,
-    cluster: t.Optional[str] = ...,
-    path_context: t.Optional[str] = ...,
-    context: t.Optional[str] = ...,
-    _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    *,
-    project_path: t.Optional[str] = ...,
-    config_file: t.Optional[str] = ...,
-) -> DeploymentInfo:
-    ...
-
-
 @t.overload
 def apply(
     name: str | None = ...,
@@ -432,43 +246,29 @@ def apply(
     context: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
-    project_path: str | None = None,
     bento: Tag | str | None = None,
     config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
 ) -> DeploymentInfo:
-    if config_dict and config_file:
-        raise BentoMLException(
-            "Configure a deployment can only use one of the following: config_dict, config_file"
-        )
-    deploy_name, bento_name, cluster_name = get_args_from_config(
-        bento=bento,
+    config_params = DeploymentConfigParameters(
         name=name,
+        path_context=path_context,
+        context=context,
+        bento=bento,
         cluster=cluster,
         config_dict=config_dict,
         config_file=config_file,
-        path_context=path_context,
     )
-    if bento_name and project_path:
-        raise BentoMLException("Only one of bento or project_path can be provided")
-    if bento_name is None and project_path is None:
-        bento = None
-    else:
-        bento = get_real_bento_tag(
-            project_path=project_path,
-            bento=bento_name,
-            context=context,
-            _cloud_client=BentoCloudClient(),
+    try:
+        config_params.verify()
+    except BentoMLException as e:
+        raise BentoMLException(
+            f"Failed to create deployment due to invalid configuration: {e}"
         )
 
     return Deployment.apply(
-        name=deploy_name,
-        bento=bento,
-        cluster=cluster_name,
+        deployment_config_params=config_params,
         context=context,
-        path_context=path_context,
-        config_dict=config_dict,
-        config_file=config_file,
     )
 
 
diff --git a/src/bentoml_cli/deployment.py b/src/bentoml_cli/deployment.py
index afd4a66b..1b5c4131 100644
--- a/src/bentoml_cli/deployment.py
+++ b/src/bentoml_cli/deployment.py
@@ -4,11 +4,14 @@ import typing as t
 
 import click
 import yaml
+from rich.live import Live
 from rich.syntax import Syntax
 
-from bentoml._internal.cloud.deployment import get_args_from_config
-from bentoml._internal.cloud.schemas.modelschemas import AccessControl
+from bentoml._internal.cloud.base import Spinner
+from bentoml._internal.cloud.deployment import Deployment
+from bentoml._internal.cloud.deployment import DeploymentConfigParameters
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentStrategy
+from bentoml.exceptions import BentoMLException
 
 if t.TYPE_CHECKING:
     TupleStrAny = tuple[str, ...]
@@ -21,17 +24,14 @@ else:
 def add_deployment_command(cli: click.Group) -> None:
     import json
 
-    import click_option_group as cog
     from rich.table import Table
 
-    from bentoml._internal.cloud.deployment import Deployment
-    from bentoml._internal.cloud.deployment import get_real_bento_tag
     from bentoml._internal.utils import rich_console as console
     from bentoml_cli.utils import BentoMLCommandGroup
 
     @cli.command()
     @click.argument(
-        "target",
+        "bento",
         type=click.STRING,
         required=False,
     )
@@ -47,11 +47,9 @@ def add_deployment_command(cli: click.Group) -> None:
         help="Name of the cluster",
     )
     @click.option(
-        "--access-type",
-        type=click.Choice(
-            [access_ctrl_type.value for access_ctrl_type in AccessControl]
-        ),
-        help="Type of access",
+        "--access-authorization",
+        type=click.BOOL,
+        help="Enable access authorization",
     )
     @click.option(
         "--scaling-min",
@@ -98,16 +96,16 @@ def add_deployment_command(cli: click.Group) -> None:
     @click.option(
         "--timeout",
         type=click.INT,
-        default=300,
+        default=1800,
         help="Timeout for deployment to be ready in seconds",
     )
     @click.pass_obj
     def deploy(
         shared_options: SharedOptions,
-        target: str | None,
+        bento: str | None,
         name: str | None,
         cluster: str | None,
-        access_type: str | None,
+        access_authorization: bool | None,
         scaling_min: int | None,
         scaling_max: int | None,
         instance_type: str | None,
@@ -120,50 +118,23 @@ def add_deployment_command(cli: click.Group) -> None:
         """Create a deployment on BentoCloud.
 
         \b
-        Create a deployment using parameters (standalone mode only), or using config yaml file.
+        Create a deployment using parameters, or using config yaml file.
         """
-        from os import path
-
-        deploy_name, bento_name, cluster_name = get_args_from_config(
-            name=name, bento=target, config_file=config_file, cluster=cluster
-        )
-        if bento_name is None:
-            raise click.BadParameter(
-                "please provide a target to deploy or a config file with bento name"
-            )
-
-        # determine if target is a path or a name
-        if isinstance(bento_name, str) and path.exists(bento_name):
-            # target is a path
-            click.echo(f"building bento from {target} ...")
-            bento_tag = get_real_bento_tag(project_path=target)
-        else:
-            click.echo(f"using bento {target}...")
-            bento_tag = get_real_bento_tag(bento=target)
-
-        deployment = Deployment.create(
-            bento=bento_tag,
-            name=deploy_name,
-            cluster=cluster_name,
-            access_type=access_type,
+        create_deployment(
+            context=shared_options.cloud_context,
+            bento=bento,
+            name=name,
+            cluster=cluster,
+            access_authorization=access_authorization,
             scaling_min=scaling_min,
             scaling_max=scaling_max,
             instance_type=instance_type,
             strategy=strategy,
-            envs=[
-                {"key": item.split("=")[0], "value": item.split("=")[1]} for item in env
-            ]
-            if env is not None
-            else None,
+            env=env,
             config_file=config_file,
-            context=shared_options.cloud_context,
+            wait=wait,
+            timeout=timeout,
         )
-        if wait:
-            deployment.wait_until_ready(timeout=timeout)
-            click.echo(
-                f"Deployment '{deployment.name}' created successfully in cluster '{deployment.cluster}'."
-            )
-        click.echo(f"To check the deployment, go to: {deployment.admin_console}.")
 
     output_option = click.option(
         "-o",
@@ -200,28 +171,20 @@ def add_deployment_command(cli: click.Group) -> None:
 
     @deployment_cli.command()
     @shared_decorator()
-    @cog.optgroup.group(cls=cog.MutuallyExclusiveOptionGroup, name="target options")
-    @cog.optgroup.option(
-        "--bento",
-        type=click.STRING,
-        help="Bento name",
-    )
-    @cog.optgroup.option(
-        "--project-path",
-        type=click.Path(exists=True),
-        help="Path to the project",
-    )
     @click.argument(
         "name",
         type=click.STRING,
-        required=True,
+        required=False,
     )
     @click.option(
-        "--access-type",
-        type=click.Choice(
-            [access_ctrl_type.value for access_ctrl_type in AccessControl]
-        ),
-        help="Type of access",
+        "--bento",
+        type=click.STRING,
+        help="Bento name or path to Bento project directory",
+    )
+    @click.option(
+        "--access-authorization",
+        type=click.BOOL,
+        help="Enable access authorization",
     )
     @click.option(
         "--scaling-min",
@@ -263,9 +226,8 @@ def add_deployment_command(cli: click.Group) -> None:
         shared_options: SharedOptions,
         name: str | None,
         cluster: str | None,
-        project_path: str | None,
         bento: str | None,
-        access_type: str | None,
+        access_authorization: bool | None,
         scaling_min: int | None,
         scaling_max: int | None,
         instance_type: str | None,
@@ -276,28 +238,17 @@ def add_deployment_command(cli: click.Group) -> None:
         """Update a deployment on BentoCloud.
 
         \b
-        A deployment can be updated using parameters (standalone mode only), or using config yaml file.
+        A deployment can be updated using parameters, or using config yaml file.
         You can also update bento by providing a project path or existing bento.
         """
-        deploy_name, bento_name, cluster_name = get_args_from_config(
-            name=name, bento=bento, cluster=cluster, config_file=config_file
-        )
-        if bento_name is None and project_path is None:
-            target = None
-        else:
-            target = get_real_bento_tag(
-                project_path=project_path,
-                bento=bento,
-                context=shared_options.cloud_context,
-            )
-
-        Deployment.update(
-            bento=target,
-            access_type=access_type,
-            name=deploy_name,
-            cluster=cluster_name,
-            scaling_min=scaling_min,
+        config_params = DeploymentConfigParameters(
+            name=name,
+            context=shared_options.cloud_context,
+            bento=bento,
+            cluster=cluster,
+            access_authorization=access_authorization,
             scaling_max=scaling_max,
+            scaling_min=scaling_min,
             instance_type=instance_type,
             strategy=strategy,
             envs=[
@@ -306,10 +257,19 @@ def add_deployment_command(cli: click.Group) -> None:
             if env is not None
             else None,
             config_file=config_file,
+        )
+        try:
+            config_params.verify()
+        except BentoMLException as e:
+            raise BentoMLException(
+                f"Failed to create deployment due to invalid configuration: {e}"
+            )
+        Deployment.update(
+            deployment_config_params=config_params,
             context=shared_options.cloud_context,
         )
 
-        click.echo(f"Deployment '{name}' updated successfully.")
+        click.echo(f"Deployment '{config_params.get_name()}' updated successfully.")
 
     @deployment_cli.command()
     @click.option(
@@ -328,27 +288,129 @@ def add_deployment_command(cli: click.Group) -> None:
         \b
         A deployment can be applied using config yaml file.
         """
-        deploy_name, bento_name, cluster_name = get_args_from_config(
-            name=None, bento=None, cluster=None, config_file=config_file
+        config_params = DeploymentConfigParameters(
+            context=shared_options.cloud_context,
+            config_file=config_file,
         )
-        if bento_name is None:
-            target = None
-        else:
-            target = get_real_bento_tag(
-                project_path=None,
-                bento=bento_name,
-                context=shared_options.cloud_context,
+        try:
+            config_params.verify()
+        except BentoMLException as e:
+            raise BentoMLException(
+                f"Failed to create deployment due to invalid configuration: {e}"
             )
-
         Deployment.apply(
-            bento=target,
-            name=deploy_name,
-            cluster=cluster_name,
-            config_file=config_file,
+            deployment_config_params=config_params,
             context=shared_options.cloud_context,
         )
 
-        click.echo(f"Deployment '{deploy_name}' applied successfully.")
+        click.echo(f"Deployment '{config_params.get_name()}' applied successfully.")
+
+    @deployment_cli.command()
+    @click.argument(
+        "bento",
+        type=click.STRING,
+        required=False,
+    )
+    @click.option(
+        "-n",
+        "--name",
+        type=click.STRING,
+        help="Deployment name",
+    )
+    @click.option(
+        "--cluster",
+        type=click.STRING,
+        help="Name of the cluster",
+    )
+    @click.option(
+        "--access-authorization",
+        type=click.BOOL,
+        help="Enable access authorization",
+    )
+    @click.option(
+        "--scaling-min",
+        type=click.INT,
+        help="Minimum scaling value",
+    )
+    @click.option(
+        "--scaling-max",
+        type=click.INT,
+        help="Maximum scaling value",
+    )
+    @click.option(
+        "--instance-type",
+        type=click.STRING,
+        help="Type of instance",
+    )
+    @click.option(
+        "--strategy",
+        type=click.Choice(
+            [deployment_strategy.value for deployment_strategy in DeploymentStrategy]
+        ),
+        help="Deployment strategy",
+    )
+    @click.option(
+        "--env",
+        type=click.STRING,
+        help="List of environment variables pass by --env key=value, --env ...",
+        multiple=True,
+    )
+    @click.option(
+        "-f",
+        "--config-file",
+        type=click.File(),
+        help="Configuration file path",
+        default=None,
+    )
+    @click.option(
+        "--wait/--no-wait",
+        type=click.BOOL,
+        is_flag=True,
+        help="Do not wait for deployment to be ready",
+        default=True,
+    )
+    @click.option(
+        "--timeout",
+        type=click.INT,
+        default=1800,
+        help="Timeout for deployment to be ready in seconds",
+    )
+    @click.pass_obj
+    def create(
+        shared_options: SharedOptions,
+        bento: str | None,
+        name: str | None,
+        cluster: str | None,
+        access_authorization: bool | None,
+        scaling_min: int | None,
+        scaling_max: int | None,
+        instance_type: str | None,
+        strategy: str | None,
+        env: tuple[str] | None,
+        config_file: str | t.TextIO | None,
+        wait: bool,
+        timeout: int,
+    ) -> None:
+        """Create a deployment on BentoCloud.
+
+        \b
+        Create a deployment using parameters, or using config yaml file.
+        """
+        create_deployment(
+            context=shared_options.cloud_context,
+            bento=bento,
+            name=name,
+            cluster=cluster,
+            access_authorization=access_authorization,
+            scaling_min=scaling_min,
+            scaling_max=scaling_max,
+            instance_type=instance_type,
+            strategy=strategy,
+            env=env,
+            config_file=config_file,
+            wait=wait,
+            timeout=timeout,
+        )
 
     @deployment_cli.command()
     @shared_decorator
@@ -456,3 +518,115 @@ def add_deployment_command(cli: click.Group) -> None:
         else:
             info = yaml.dump(res, indent=2, sort_keys=False)
             console.print(Syntax(info, "yaml", background_color="default"))
+
+    @deployment_cli.command()
+    @click.option(
+        "--cluster", type=click.STRING, default=None, help="Name of the cluster."
+    )
+    # @click.option(
+    #     "--search", type=click.STRING, default=None, help="Search for list request."
+    # )
+    @click.option(
+        "-o",
+        "--output",
+        help="Display the output of this command.",
+        type=click.Choice(["json", "yaml", "table"]),
+        default="table",
+    )
+    @click.pass_obj
+    def list_instance_types(  # type: ignore
+        shared_options: SharedOptions,
+        cluster: str | None,
+        # search: str | None,
+        output: t.Literal["json", "yaml", "table"],
+    ) -> None:
+        """List existing instance types in cluster on BentoCloud."""
+        d_list = Deployment.list_instance_types(
+            context=shared_options.cloud_context, cluster=cluster
+        )
+        res: list[dict[str, t.Any]] = [d.to_dict() for d in d_list]
+        if output == "table":
+            table = Table(box=None)
+            table.add_column("Name")
+            table.add_column("Price")
+            table.add_column("CPU")
+            table.add_column("Memory")
+            table.add_column("GPU")
+            table.add_column("GPU Type")
+            for info in d_list:
+                table.add_row(
+                    info.name,
+                    info.price,
+                    info.cpu,
+                    info.memory,
+                    info.gpu,
+                    info.gpu_type,
+                )
+            console.print(table)
+        elif output == "json":
+            info = json.dumps(res, indent=2, default=str)
+            console.print_json(info)
+        else:
+            info = yaml.dump(res, indent=2, sort_keys=False)
+            console.print(Syntax(info, "yaml", background_color="default"))
+
+
+def create_deployment(
+    context: str | None,
+    bento: str | None,
+    name: str | None,
+    cluster: str | None,
+    access_authorization: bool | None,
+    scaling_min: int | None,
+    scaling_max: int | None,
+    instance_type: str | None,
+    strategy: str | None,
+    env: tuple[str] | None,
+    config_file: str | t.TextIO | None,
+    wait: bool,
+    timeout: int,
+) -> None:
+    config_params = DeploymentConfigParameters(
+        name=name,
+        context=context,
+        bento=bento,
+        cluster=cluster,
+        access_authorization=access_authorization,
+        scaling_max=scaling_max,
+        scaling_min=scaling_min,
+        instance_type=instance_type,
+        strategy=strategy,
+        envs=[{"key": item.split("=")[0], "value": item.split("=")[1]} for item in env]
+        if env is not None
+        else None,
+        config_file=config_file,
+    )
+    try:
+        config_params.verify()
+    except BentoMLException as e:
+        raise BentoMLException(
+            f"Failed to create deployment due to invalid configuration: {e}"
+        )
+    spinner = Spinner()
+    with Live(spinner.progress_group):
+        task_id = spinner.spinner_progress.add_task(
+            "deploy", action="Deploying to BentoCloud"
+        )
+        deployment = Deployment.create(
+            deployment_config_params=config_params,
+            context=context,
+        )
+        spinner.log_progress.add_task(
+            f"[bold green]Successfully created deployment '{deployment.name}' in cluster '{deployment.cluster}'[/bold green]"
+        )
+        if wait:
+            spinner.spinner_progress.update(
+                task_id,
+                action="[bold blue]Waiting for deployment to be ready, you can use --no-wait to skip this process[/bold blue]",
+            )
+            deployment.wait_until_ready(
+                timeout=timeout, spinner_task_id=task_id, spinner=spinner
+            )
+        spinner.spinner_progress.stop_task(task_id)
+
+    click.echo(f"To check the deployment, go to: {deployment.admin_console}.")
diff --git a/src/bentoml_cli/start.py b/src/bentoml_cli/start.py
index e0ccdd96..47216fea 100644
--- a/src/bentoml_cli/start.py
+++ b/src/bentoml_cli/start.py
@@ -170,7 +170,7 @@ def add_start_command(cli: click.Group) -> None:
         svc = load(bento, working_dir=working_dir)
         if isinstance(svc, Service):
             # for <1.2 bentos
-            if not service_name:
+            if not service_name or service_name == svc.name:
                 from bentoml.start import start_http_server
 
                 for dep in depends or []:


Test patch:
diff --git a/tests/unit/_internal/cloud/test_deployment.py b/tests/unit/_internal/cloud/test_deployment.py
index 75abae07..bf985fef 100644
--- a/tests/unit/_internal/cloud/test_deployment.py
+++ b/tests/unit/_internal/cloud/test_deployment.py
@@ -9,7 +9,7 @@ import pytest
 
 from bentoml._internal.cloud.client import RestApiClient
 from bentoml._internal.cloud.deployment import Deployment
-from bentoml._internal.cloud.schemas.modelschemas import AccessControl
+from bentoml._internal.cloud.deployment import DeploymentConfigParameters
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentServiceConfig
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentStrategy
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentTargetHPAConf
@@ -58,7 +58,6 @@ class DummyUpdateSchema(UpdateDeploymentSchemaV2):
 def dummy_generate_deployment_schema(
     name: str,
     cluster: str | None,
-    distributed: bool,
     update_schema: UpdateDeploymentSchemaV2,
 ):
     user = UserSchema(name="", email="", first_name="", last_name="")
@@ -106,18 +105,14 @@ def dummy_generate_deployment_schema(
         ),
     )
     return DeploymentFullSchemaV2(
-        distributed=distributed,
         latest_revision=DeploymentRevisionSchemaV2(
             targets=[
                 DeploymentTargetSchemaV2(
                     bento=dummy_bento,
                     config=DeploymentConfigSchemaV2(
-                        access_type=update_schema.access_type,
+                        access_authorization=update_schema.access_authorization,
                         envs=update_schema.envs,
                         services=update_schema.services,
-                        scaling=update_schema.scaling,
-                        deployment_strategy=update_schema.deployment_strategy,
-                        bentoml_config_overrides=update_schema.bentoml_config_overrides,
                     ),
                     uid="",
                     created_at=datetime(2023, 5, 1),
@@ -170,10 +165,12 @@ def fixture_rest_client() -> RestApiClient:
     def dummy_create_deployment(
         create_schema: CreateDeploymentSchemaV2, cluster: str | None = None
     ):
+        if create_schema.name is None:
+            create_schema.name = "empty_name"
         if cluster is None:
             cluster = "default_display_name"
         return dummy_generate_deployment_schema(
-            create_schema.name, cluster, create_schema.distributed, create_schema
+            create_schema.name, cluster, create_schema
         )
 
     def dummy_update_deployment(
@@ -183,7 +180,7 @@ def fixture_rest_client() -> RestApiClient:
     ):
         if cluster is None:
             cluster = "default_display_name"
-        return dummy_generate_deployment_schema(name, cluster, False, update_schema)
+        return dummy_generate_deployment_schema(name, cluster, update_schema)
 
     def dummy_get_deployment(
         name: str,
@@ -195,7 +192,6 @@ def fixture_rest_client() -> RestApiClient:
             return dummy_generate_deployment_schema(
                 name,
                 cluster,
-                True,
                 UpdateDeploymentSchemaV2(
                     bento="abc:123",
                     envs=[EnvItemSchema(name="env_key", value="env_value")],
@@ -221,12 +217,17 @@ def fixture_rest_client() -> RestApiClient:
             return dummy_generate_deployment_schema(
                 name,
                 cluster,
-                False,
                 UpdateDeploymentSchemaV2(
                     bento="abc:123",
-                    access_type=AccessControl.PUBLIC,
-                    scaling=DeploymentTargetHPAConf(min_replicas=3, max_replicas=5),
-                    deployment_strategy=DeploymentStrategy.RollingUpdate,
+                    access_authorization=False,
+                    services={
+                        "irisclassifier": DeploymentServiceConfig(
+                            scaling=DeploymentTargetHPAConf(
+                                min_replicas=3, max_replicas=5
+                            ),
+                            deployment_strategy=DeploymentStrategy.RollingUpdate,
+                        )
+                    },
                     envs=[EnvItemSchema(name="env_key", value="env_value")],
                 ),
             )
@@ -263,14 +264,23 @@ def fixture_rest_client() -> RestApiClient:
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
 def test_create_deployment(mock_get_client: MagicMock, rest_client: RestApiClient):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.create(bento="abc:123")
+    deployment = Deployment.create(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "bento": "abc:123",
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
-    assert deployment.name == ""
-    assert deployment.distributed is False
+    assert deployment.name == "empty_name"
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=1, max_replicas=1)
+    assert config.access_authorization is False
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=1, max_replicas=1
+        )
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -279,25 +289,34 @@ def test_create_deployment_custom_standalone(
 ):
     mock_get_client.return_value = rest_client
     deployment = Deployment.create(
-        bento="abc:123",
-        name="custom-name",
-        scaling_min=2,
-        scaling_max=4,
-        access_type="private",
-        cluster="custom-cluster",
-        envs=[{"name": "env_key", "value": "env_value"}],
-        strategy="RollingUpdate",
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "bento": "abc:123",
+                "name": "custom-name",
+                "access_authorization": "private",
+                "cluster": "custom-cluster",
+                "envs": [{"name": "env_key", "value": "env_value"}],
+                "services": {
+                    "irisclassifier": {
+                        "deployment_strategy": "RollingUpdate",
+                        "scaling": {"min_replicas": 2, "max_replicas": 4},
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
     )
     # assert expected schema
     assert deployment.cluster == "custom-cluster"
     assert deployment.name == "custom-name"
-    assert deployment.distributed is False
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PRIVATE
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=2, max_replicas=4)
-
-    assert config.deployment_strategy == DeploymentStrategy.RollingUpdate
+    assert config.access_authorization is True
     assert config.envs == [EnvItemSchema(name="env_key", value="env_value")]
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=2, max_replicas=4
+        )
+        assert service.deployment_strategy == DeploymentStrategy.RollingUpdate
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -305,14 +324,28 @@ def test_create_deployment_scailing_only_min(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.create(bento="abc:123", scaling_min=3)
+    deployment = Deployment.create(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "bento": "abc:123",
+                "services": {
+                    "irisclassifier": {
+                        "scaling": {"min_replicas": 3},
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
-    assert deployment.name == ""
-    assert deployment.distributed is False
+    assert deployment.name == "empty_name"
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=3, max_replicas=3)
+    assert config.access_authorization is False
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=3, max_replicas=3
+        )
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -320,29 +353,28 @@ def test_create_deployment_scailing_only_max(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.create(bento="abc:123", scaling_max=3)
-    # assert expected schema
-    assert deployment.cluster == "default_display_name"
-    assert deployment.name == ""
-    assert deployment.distributed is False
-    config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=1, max_replicas=3)
-
-
-@patch("bentoml._internal.cloud.deployment.get_rest_api_client")
-def test_create_deployment_scailing_mismatch_min_max(
-    mock_get_client: MagicMock, rest_client: RestApiClient
-):
-    mock_get_client.return_value = rest_client
-    deployment = Deployment.create(bento="abc:123", scaling_min=3, scaling_max=2)
+    deployment = Deployment.create(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "bento": "abc:123",
+                "services": {
+                    "irisclassifier": {
+                        "scaling": {"max_replicas": 3},
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
-    assert deployment.name == ""
-    assert deployment.distributed is False
+    assert deployment.name == "empty_name"
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=2, max_replicas=2)
+    assert config.access_authorization is False
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=1, max_replicas=3
+        )
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -350,65 +382,75 @@ def test_create_deployment_config_dict(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    config_dict = {
-        "services": {
-            "irisclassifier": {"scaling": {"max_replicas": 2, "min_replicas": 1}},
-            "preprocessing": {"scaling": {"max_replicas": 2}},
-        },
-        "envs": [{"name": "env_key", "value": "env_value"}],
-        "bentoml_config_overrides": {
-            "irisclassifier": {
-                "resources": {
-                    "cpu": "300m",
-                    "memory": "500m",
+    deployment = Deployment.create(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "bento": "abc:123",
+                "services": {
+                    "irisclassifier": {
+                        "scaling": {"max_replicas": 2, "min_replicas": 1},
+                        "config_overrides": {
+                            "resources": {"cpu": "300m", "memory": "500m"}
+                        },
+                    },
+                    "preprocessing": {"scaling": {"max_replicas": 2}},
                 },
-            }
-        },
-    }
-    deployment = Deployment.create(bento="abc:123", config_dict=config_dict)
+                "envs": [{"name": "env_key", "value": "env_value"}],
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
-    assert deployment.name == ""
-    assert deployment.distributed
+    assert deployment.name == "empty_name"
     config = deployment.get_config(refetch=False)
     assert config.services == {
         "irisclassifier": DeploymentServiceConfig(
-            scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2)
+            scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2),
+            config_overrides={
+                "resources": {
+                    "cpu": "300m",
+                    "memory": "500m",
+                },
+            },
         ),
         "preprocessing": DeploymentServiceConfig(
             scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2)
         ),
     }
-    assert config.bentoml_config_overrides == {
-        "irisclassifier": {
-            "resources": {
-                "cpu": "300m",
-                "memory": "500m",
-            },
-        }
-    }
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
 def test_update_deployment(mock_get_client: MagicMock, rest_client: RestApiClient):
     mock_get_client.return_value = rest_client
     deployment = Deployment.update(
-        name="test",
-        bento="abc:1234",
-        access_type="private",
-        envs=[{"name": "env_key2", "value": "env_value2"}],
-        strategy="Recreate",
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "name": "test",
+                "bento": "abc:1234",
+                "access_authorization": "private",
+                "envs": [{"name": "env_key2", "value": "env_value2"}],
+                "services": {
+                    "irisclassifier": {
+                        "deployment_strategy": "Recreate",
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
     )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
     assert deployment.get_bento(refetch=False) == "abc:1234"
     assert deployment.name == "test"
-    assert deployment.distributed is False
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PRIVATE
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=3, max_replicas=5)
-    assert config.deployment_strategy == DeploymentStrategy.Recreate
+    assert config.access_authorization is True
     assert config.envs == [EnvItemSchema(name="env_key2", value="env_value2")]
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=3, max_replicas=5
+        )
+        assert service.deployment_strategy == DeploymentStrategy.Recreate
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -416,16 +458,30 @@ def test_update_deployment_scaling_only_min(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.update(name="test", scaling_min=1)
+    deployment = Deployment.update(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "name": "test",
+                "services": {
+                    "irisclassifier": {
+                        "scaling": {"min_replicas": 1},
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
     assert deployment.name == "test"
-    assert deployment.distributed is False
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=1, max_replicas=5)
-    assert config.deployment_strategy == DeploymentStrategy.RollingUpdate
+    assert config.access_authorization is False
     assert config.envs == [EnvItemSchema(name="env_key", value="env_value")]
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=1, max_replicas=5
+        )
+        assert service.deployment_strategy == DeploymentStrategy.RollingUpdate
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -433,16 +489,30 @@ def test_update_deployment_scaling_only_max(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.update(name="test", scaling_max=3)
+    deployment = Deployment.update(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "name": "test",
+                "services": {
+                    "irisclassifier": {
+                        "scaling": {"max_replicas": 3},
+                    },
+                },
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
     assert deployment.name == "test"
-    assert deployment.distributed is False
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=3, max_replicas=3)
-    assert config.deployment_strategy == DeploymentStrategy.RollingUpdate
+    assert config.access_authorization is False
     assert config.envs == [EnvItemSchema(name="env_key", value="env_value")]
+    for service in config.services.values():
+        assert service.scaling == DeploymentTargetHPAConf(
+            min_replicas=3, max_replicas=3
+        )
+        assert service.deployment_strategy == DeploymentStrategy.RollingUpdate
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -450,16 +520,23 @@ def test_update_deployment_scaling_too_big_min(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    deployment = Deployment.update(name="test", scaling_min=10)
-    # assert expected schema
-    assert deployment.cluster == "default_display_name"
-    assert deployment.name == "test"
-    assert deployment.distributed is False
-    config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
-    assert config.scaling == DeploymentTargetHPAConf(min_replicas=5, max_replicas=5)
-    assert config.deployment_strategy == DeploymentStrategy.RollingUpdate
-    assert config.envs == [EnvItemSchema(name="env_key", value="env_value")]
+    try:
+        Deployment.update(
+            deployment_config_params=DeploymentConfigParameters(
+                cfg_dict={
+                    "name": "test",
+                    "services": {
+                        "irisclassifier": {"scaling": {"min_replicas": 10}},
+                    },
+                },
+                service_name="irisclassifier",
+            )
+        )
+    except Exception as e:
+        assert (
+            "min scaling values must be less than or equal to max scaling values"
+            in str(e)
+        )
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -467,18 +544,23 @@ def test_update_deployment_distributed(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    config_dict = {
-        "services": {
-            "irisclassifier": {"scaling": {"max_replicas": 50}},
-            "preprocessing": {"instance_type": "t3-large"},
-        }
-    }
-    deployment = Deployment.update(name="test-distributed", config_dict=config_dict)
+    deployment = Deployment.update(
+        deployment_config_params=DeploymentConfigParameters(
+            cfg_dict={
+                "name": "test-distributed",
+                "services": {
+                    "irisclassifier": {"scaling": {"max_replicas": 50}},
+                    "preprocessing": {"instance_type": "t3-large"},
+                },
+            },
+            service_name="irisclassifier",
+        )
+    )
     # assert expected schema
     assert deployment.cluster == "default_display_name"
     assert deployment.name == "test-distributed"
     config = deployment.get_config(refetch=False)
-    assert config.access_type == AccessControl.PUBLIC
+    assert config.access_authorization is False
     assert config.envs == [EnvItemSchema(name="env_key", value="env_value")]
     assert config.services == {
         "irisclassifier": DeploymentServiceConfig(