Repository: bentoml/BentoML
Pull request #4383 (2024-01-03 15:02:26)

fix: refactor deployment v2 client and cli
## 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?


bojiang: ‌‌The unit test did not pass.

Patch:
diff --git a/src/bentoml/_internal/cloud/client.py b/src/bentoml/_internal/cloud/client.py
index 16e62eb6..84429f7e 100644
--- a/src/bentoml/_internal/cloud/client.py
+++ b/src/bentoml/_internal/cloud/client.py
@@ -421,9 +421,9 @@ class RestApiClientV1(BaseRestApiClient):
         return schema_from_json(resp.text, ModelWithRepositoryListSchema)
 
     def get_cluster_deployment_list(
-        self, cluster_name: str, **params: str | int | None
+        self, cluster: str, **params: str | int | None
     ) -> DeploymentListSchema | None:
-        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}/deployments")
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster}/deployments")
         resp = self.session.get(url, params=params)
         if self._is_not_found(resp):
             return None
@@ -441,19 +441,19 @@ class RestApiClientV1(BaseRestApiClient):
         return schema_from_json(resp.text, DeploymentListSchema)
 
     def create_deployment(
-        self, cluster_name: str, create_schema: CreateDeploymentSchemaV1
+        self, cluster: str, create_schema: CreateDeploymentSchemaV1
     ) -> DeploymentFullSchema | None:
-        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}/deployments")
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster}/deployments")
         resp = self.session.post(url, content=schema_to_json(create_schema))
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchema)
 
     def get_deployment(
-        self, cluster_name: str, kube_namespace: str, deployment_name: str
+        self, cluster: str, kube_namespace: str, name: str
     ) -> DeploymentFullSchema | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+            f"/api/v1/clusters/{cluster}/namespaces/{kube_namespace}/deployments/{name}",
         )
         resp = self.session.get(url)
         if self._is_not_found(resp):
@@ -463,14 +463,14 @@ class RestApiClientV1(BaseRestApiClient):
 
     def update_deployment(
         self,
-        cluster_name: str,
+        cluster: str,
         kube_namespace: str,
-        deployment_name: str,
+        name: str,
         update_schema: UpdateDeploymentSchema,
     ) -> DeploymentFullSchema | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+            f"/api/v1/clusters/{cluster}/namespaces/{kube_namespace}/deployments/{name}",
         )
         resp = self.session.patch(url, content=schema_to_json(update_schema))
         if self._is_not_found(resp):
@@ -479,11 +479,11 @@ class RestApiClientV1(BaseRestApiClient):
         return schema_from_json(resp.text, DeploymentFullSchema)
 
     def terminate_deployment(
-        self, cluster_name: str, kube_namespace: str, deployment_name: str
+        self, cluster: str, kube_namespace: str, name: str
     ) -> DeploymentFullSchema | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}/terminate",
+            f"/api/v1/clusters/{cluster}/namespaces/{kube_namespace}/deployments/{name}/terminate",
         )
         resp = self.session.post(url)
         if self._is_not_found(resp):
@@ -492,11 +492,11 @@ class RestApiClientV1(BaseRestApiClient):
         return schema_from_json(resp.text, DeploymentFullSchema)
 
     def delete_deployment(
-        self, cluster_name: str, kube_namespace: str, deployment_name: str
+        self, cluster: str, kube_namespace: str, name: str
     ) -> DeploymentFullSchema | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+            f"/api/v1/clusters/{cluster}/namespaces/{kube_namespace}/deployments/{name}",
         )
         resp = self.session.delete(url)
         if self._is_not_found(resp):
@@ -514,8 +514,8 @@ class RestApiClientV1(BaseRestApiClient):
         self._check_resp(resp)
         return schema_from_json(resp.text, ClusterListSchema)
 
-    def get_cluster(self, cluster_name: str) -> ClusterFullSchema | None:
-        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}")
+    def get_cluster(self, cluster: str) -> ClusterFullSchema | None:
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster}")
         resp = self.session.get(url)
         if self._is_not_found(resp):
             return None
@@ -540,40 +540,44 @@ class RestApiClientV1(BaseRestApiClient):
 
 class RestApiClientV2(BaseRestApiClient):
     def create_deployment(
-        self, create_schema: CreateDeploymentSchemaV2, cluster_name: str
+        self,
+        create_schema: CreateDeploymentSchemaV2,
+        cluster: str | None = None,
     ) -> DeploymentFullSchemaV2:
         url = urljoin(self.endpoint, "/api/v2/deployments")
         resp = self.session.post(
-            url, content=schema_to_json(create_schema), params={"cluster": cluster_name}
+            url, content=schema_to_json(create_schema), params={"cluster": cluster}
         )
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
     def update_deployment(
         self,
+        name: str,
         update_schema: UpdateDeploymentSchemaV2,
-        cluster_name: str,
-        deployment_name: str,
+        cluster: str | None = None,
     ) -> DeploymentFullSchemaV2 | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v2/deployments/{deployment_name}",
+            f"/api/v2/deployments/{name}",
         )
         data = schema_to_json(update_schema)
-        resp = self.session.put(url, content=data, params={"cluster": cluster_name})
+        resp = self.session.put(url, content=data, params={"cluster": cluster})
         if self._is_not_found(resp):
             return None
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
     def get_deployment(
-        self, cluster_name: str, deployment_name: str
+        self,
+        name: str,
+        cluster: str | None = None,
     ) -> DeploymentFullSchemaV2 | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v2/deployments/{deployment_name}",
+            f"/api/v2/deployments/{name}",
         )
-        resp = self.session.get(url, params={"cluster": cluster_name})
+        resp = self.session.get(url, params={"cluster": cluster})
         if self._is_not_found(resp):
             return None
         self._check_resp(resp)
@@ -581,7 +585,7 @@ class RestApiClientV2(BaseRestApiClient):
 
     def list_deployment(
         self,
-        cluster_name: str | None = None,
+        cluster: str | None = None,
         all: bool | None = None,
         # if both of the above is none, list default cluster's deployments
         count: int | None = None,
@@ -593,7 +597,7 @@ class RestApiClientV2(BaseRestApiClient):
         resp = self.session.get(
             url,
             params={
-                "cluster": cluster_name,
+                "cluster": cluster,
                 "all": all,
                 "count": count,
                 "q": q,
@@ -607,26 +611,30 @@ class RestApiClientV2(BaseRestApiClient):
         return schema_from_json(resp.text, DeploymentListSchemaV2)
 
     def terminate_deployment(
-        self, cluster_name: str, deployment_name: str
+        self,
+        name: str,
+        cluster: str | None = None,
     ) -> DeploymentFullSchemaV2 | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v2/deployments/{deployment_name}/terminate",
+            f"/api/v2/deployments/{name}/terminate",
         )
-        resp = self.session.post(url, params={"cluster": cluster_name})
+        resp = self.session.post(url, params={"cluster": cluster})
         if self._is_not_found(resp):
             return None
         self._check_resp(resp)
         return schema_from_json(resp.text, DeploymentFullSchemaV2)
 
     def delete_deployment(
-        self, cluster_name: str, deployment_name: str
+        self,
+        name: str,
+        cluster: str | None = None,
     ) -> DeploymentFullSchemaV2 | None:
         url = urljoin(
             self.endpoint,
-            f"/api/v2/deployments/{deployment_name}",
+            f"/api/v2/deployments/{name}",
         )
-        resp = self.session.delete(url, params={"cluster": cluster_name})
+        resp = self.session.delete(url, params={"cluster": cluster})
         if self._is_not_found(resp):
             return None
         self._check_resp(resp)
diff --git a/src/bentoml/_internal/cloud/deployment.py b/src/bentoml/_internal/cloud/deployment.py
index 1f4174dd..9aaaac63 100644
--- a/src/bentoml/_internal/cloud/deployment.py
+++ b/src/bentoml/_internal/cloud/deployment.py
@@ -8,7 +8,6 @@ import attr
 import yaml
 from deepmerge.merger import Merger
 from simple_di import Provide
-from simple_di import inject
 
 if t.TYPE_CHECKING:
     from _bentoml_impl.client import AsyncHTTPClient
@@ -29,6 +28,7 @@ from .schemas.modelschemas import DeploymentStatus
 from .schemas.modelschemas import DeploymentTargetHPAConf
 from .schemas.schemasv2 import CreateDeploymentSchema as CreateDeploymentSchemaV2
 from .schemas.schemasv2 import DeploymentSchema
+from .schemas.schemasv2 import DeploymentTargetSchema
 from .schemas.schemasv2 import UpdateDeploymentSchema as UpdateDeploymentSchemaV2
 
 logger = logging.getLogger(__name__)
@@ -43,7 +43,55 @@ config_merger = Merger(
 )
 
 
-@inject
+def get_args_from_config(
+    name: str | None = None,
+    bento: str | None = None,
+    cluster: str | None = None,
+    config_dict: dict[str, t.Any] | None = None,
+    config_file: str | t.TextIO | None = None,
+    path_context: str | None = None,
+):
+    file_dict: dict[str, t.Any] | None = None
+
+    if name is None and config_dict is not None and "name" in config_dict:
+        name = config_dict["name"]
+    if bento is None and config_dict is not None and "bento" in config_dict:
+        bento = config_dict["bento"]
+    if cluster is None and config_dict is not None and "cluster" in config_dict:
+        cluster = config_dict["cluster"]
+
+    if isinstance(config_file, str):
+        real_path = resolve_user_filepath(config_file, path_context)
+        try:
+            with open(real_path, "r") as file:
+                file_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:
+            file_dict = yaml.safe_load(config_file)
+        except yaml.YAMLError as exc:
+            logger.error("Error while parsing YAML config-file stream: %s", exc)
+            raise
+
+    if file_dict is not None:
+        if bento is None and "bento" in file_dict:
+            bento = file_dict["bento"]
+        if name is None and "name" in file_dict:
+            name = file_dict["name"]
+        if cluster is None and "cluster" in file_dict:
+            cluster = file_dict["cluster"]
+
+    return name, bento, cluster
+
+
 def get_real_bento_tag(
     project_path: str | None = None,
     bento: str | Tag | None = None,
@@ -79,28 +127,162 @@ def get_real_bento_tag(
 
 
 @attr.define
-class DeploymentInfo:
-    __omit_if_default__ = True
-    name: str
+class DeploymentConfig(CreateDeploymentSchemaV2):
+    def to_yaml(self):
+        return yaml.dump(bentoml_cattr.unstructure(self))
+
+    def to_dict(self):
+        return bentoml_cattr.unstructure(self)
+
+
+@attr.define
+class DeploymentState:
+    status: str
     created_at: str
-    bento: Tag
-    status: DeploymentStatus
-    admin_console: str
-    endpoint: t.Optional[str]
-    config: dict[str, t.Any]
+    updated_at: str
+    # no error message for now
+    # error_msg: str
 
-    def to_dict(self) -> t.Dict[str, t.Any]:
+    def to_dict(self) -> dict[str, t.Any]:
         return bentoml_cattr.unstructure(self)
 
 
 @attr.define
-class Deployment:
-    context: t.Optional[str]
-    cluster_name: str
+class DeploymentInfo:
     name: str
+    admin_console: str
+    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)
+    _urls: t.Optional[list[str]] = attr.field(alias="_urls", default=None, repr=False)
+
+    def to_dict(self) -> dict[str, t.Any]:
+        return {
+            "name": self.name,
+            "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(),
+            "status": self.get_status(refetch=False).to_dict(),
+        }
+
+    def _refetch(self) -> None:
+        res = Deployment.get(self.name, self.cluster, self._context)
+        self._schema = res._schema
+        self._urls = res._urls
+
+    def _refetch_target(self, refetch: bool) -> DeploymentTargetSchema:
+        if refetch:
+            self._refetch()
+        if self._schema.latest_revision is None:
+            raise BentoMLException(f"Deployment {self.name} has no latest revision")
+        if len(self._schema.latest_revision.targets) == 0:
+            raise BentoMLException(
+                f"Deployment {self.name} has no latest revision targets"
+            )
+        target = self._schema.latest_revision.targets[0]
+        if target is None:
+            raise BentoMLException(f"Deployment {self.name} has no target")
+        return target
+
+    def get_config(self, refetch: bool = True) -> DeploymentConfig:
+        target = self._refetch_target(refetch)
+        if target.config is None:
+            raise BentoMLException(f"Deployment {self.name} has no config")
+
+        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,
+            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:
+        if refetch:
+            self._refetch()
+        updated_at = self._schema.created_at.strftime("%Y-%m-%d %H:%M:%S")
+        if self._schema.updated_at is not None:
+            updated_at = self._schema.updated_at.strftime("%Y-%m-%d %H:%M:%S")
+        return DeploymentState(
+            status=self._schema.status.value,
+            created_at=self._schema.created_at.strftime("%Y-%m-%d %H:%M:%S"),
+            updated_at=updated_at,
+        )
+
+    def get_bento(self, refetch: bool = True) -> str:
+        target = self._refetch_target(refetch)
+        if target.bento is None:
+            raise BentoMLException(f"Deployment {self.name} has no bento")
+        return target.bento.name + ":" + target.bento.version
+
+    def get_client(
+        self,
+        media_type: str = "application/json",
+        token: str | None = None,
+    ) -> SyncHTTPClient:
+        from _bentoml_impl.client import SyncHTTPClient
+
+        self._refetch()
+        if self._schema.status != DeploymentStatus.Running:
+            raise BentoMLException(f"Deployment status is {self._schema.status}")
+        if self._urls is None or len(self._urls) != 1:
+            raise BentoMLException("Deployment url is not ready")
+
+        return SyncHTTPClient(self._urls[0], media_type=media_type, token=token)
 
+    def get_async_client(
+        self,
+        media_type: str = "application/json",
+        token: str | None = None,
+    ) -> AsyncHTTPClient:
+        from _bentoml_impl.client import AsyncHTTPClient
+
+        self._refetch()
+        if self._schema.status != DeploymentStatus.Running:
+            raise BentoMLException(f"Deployment status is {self._schema.status}")
+        if self._urls is None or len(self._urls) != 1:
+            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 = 300, check_interval: int = 5) -> 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."
+                )
+                return
+            logger.info(
+                f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Waiting for deployment '{self.name}' to be ready. Current status: '{status}'."
+            )
+            time.sleep(check_interval)
+
+        raise TimeoutError(
+            f"Timed out waiting for deployment '{self.name}' to be ready."
+        )
+
+
+@attr.define
+class Deployment:
     @staticmethod
     def _fix_scaling(
         scaling: DeploymentTargetHPAConf | None,
@@ -175,48 +357,21 @@ class Deployment:
         if config_struct.access_type is None:
             config_struct.access_type = AccessControl.PUBLIC
 
-    @classmethod
-    def _get_default_kube_namespace(
-        cls,
-        cluster_name: str,
-        context: str | None = None,
-    ) -> str:
-        cloud_rest_client = get_rest_api_client(context)
-        res = cloud_rest_client.v1.get_cluster(cluster_name)
-        if not res:
-            raise BentoMLException("Cannot get default kube namespace")
-        return res.config.default_deployment_kube_namespace
-
-    @classmethod
-    def _get_default_cluster(cls, context: str | None = None) -> str:
-        cloud_rest_client = get_rest_api_client(context)
-        res = cloud_rest_client.v1.get_cluster_list(params={"count": 1})
-        if not res:
-            raise BentoMLException("Failed to get list of clusters.")
-        if not res.items:
-            raise BentoMLException("Cannot get default clusters.")
-        return res.items[0].name
-
-    def _refetch(self) -> None:
-        cloud_rest_client = get_rest_api_client(self.context)
-        res = cloud_rest_client.v2.get_deployment(self.cluster_name, self.name)
-        if res is None:
-            raise NotFound(f"deployment {self.name} is not found")
-        self._schema = res
-        self._urls = res.urls
-
-    def _conver_schema_to_update_schema(self) -> dict[str, t.Any]:
-        if self._schema.latest_revision is None:
+    @staticmethod
+    def _convert_schema_to_update_schema(_schema: DeploymentSchema) -> dict[str, t.Any]:
+        if _schema.latest_revision is None:
+            raise BentoMLException(f"Deployment {_schema.name} has no latest revision")
+        if len(_schema.latest_revision.targets) == 0:
             raise BentoMLException(
-                f"Deployment {self._schema.name} has no latest revision"
+                f"Deployment {_schema.name} has no latest revision targets"
             )
-        target_schema = self._schema.latest_revision.targets[0]
+        target_schema = _schema.latest_revision.targets[0]
         if target_schema is None:
-            raise BentoMLException(f"Deployment {self._schema.name} has no target")
+            raise BentoMLException(f"Deployment {_schema.name} has no target")
         if target_schema.config is None:
-            raise BentoMLException(f"Deployment {self._schema.name} has no config")
+            raise BentoMLException(f"Deployment {_schema.name} has no config")
         if target_schema.bento is None:
-            raise BentoMLException(f"Deployment {self._schema.name} has no bento")
+            raise BentoMLException(f"Deployment {_schema.name} has no bento")
         update_schema = UpdateDeploymentSchemaV2(
             services=target_schema.config.services,
             instance_type=target_schema.config.instance_type,
@@ -231,110 +386,51 @@ class Deployment:
         )
         return bentoml_cattr.unstructure(update_schema)
 
-    def _conver_schema_to_bento(self) -> Tag:
-        if self._schema.latest_revision is None:
-            raise BentoMLException(
-                f"Deployment {self._schema.name} has no latest revision"
-            )
-        target_schema = self._schema.latest_revision.targets[0]
+    @staticmethod
+    def _convert_schema_to_bento(_schema: DeploymentSchema) -> Tag:
+        if _schema.latest_revision is None:
+            raise BentoMLException(f"Deployment {_schema.name} has no latest revision")
+        target_schema = _schema.latest_revision.targets[0]
         if target_schema is None:
-            raise BentoMLException(f"Deployment {self._schema.name} has no target")
+            raise BentoMLException(f"Deployment {_schema.name} has no target")
         if target_schema.bento is None:
-            raise BentoMLException(f"Deployment {self._schema.name} has no bento")
+            raise BentoMLException(f"Deployment {_schema.name} has no bento")
         return Tag.from_taglike(
             target_schema.bento.repository.name + ":" + target_schema.bento.name
         )
 
-    @property
-    def info(self) -> DeploymentInfo:
-        schema = self._conver_schema_to_update_schema()
-        del schema["bento"]
+    @staticmethod
+    def _generate_deployment_info_(
+        context: str | None, res: DeploymentSchema, urls: list[str] | None = None
+    ) -> DeploymentInfo:
+        client = get_rest_api_client(context)
+        cluster_display_name = res.cluster.host_cluster_display_name
+        if cluster_display_name is None:
+            cluster_display_name = res.cluster.name
         return DeploymentInfo(
-            name=self.name,
-            bento=self._conver_schema_to_bento(),
-            status=self._schema.status,
-            admin_console=self.get_bento_cloud_url(),
-            endpoint=self._urls[0] if self._urls else None,
-            config=schema,
-            created_at=self._schema.created_at.strftime("%Y-%m-%d %H:%M:%S"),
-        )
-
-    def get_config(self) -> dict[str, t.Any]:
-        self._refetch()
-        res = self._conver_schema_to_update_schema()
-        # bento should not be in the deployment config
-        del res["bento"]
-        return res
-
-    def get_bento(self) -> str:
-        self._refetch()
-        return str(self._conver_schema_to_bento())
-
-    def get_status(self) -> str:
-        self._refetch()
-        return self._schema.status.value
-
-    def get_client(
-        self,
-        is_async: bool = False,
-        media_type: str = "application/json",
-        token: str | None = None,
-    ) -> SyncHTTPClient:
-        from _bentoml_impl.client import SyncHTTPClient
-
-        self._refetch()
-        if self._schema.status != DeploymentStatus.Running:
-            raise BentoMLException(f"Deployment status is {self._schema.status}")
-        if self._urls is None or len(self._urls) != 1:
-            raise BentoMLException("Deployment url is not ready")
-        return SyncHTTPClient(self._urls[0], media_type=media_type, token=token)
-
-    def get_bento_cloud_url(self) -> str:
-        client = get_rest_api_client(self.context)
-        namespace = self._get_default_kube_namespace(self.cluster_name, self.context)
-        return f"{client.v1.endpoint}/clusters/{self.cluster_name}/namespaces/{namespace}/deployments/{self.name}"
-
-    def get_async_client(
-        self,
-        media_type: str = "application/json",
-        token: str | None = None,
-    ) -> AsyncHTTPClient:
-        from _bentoml_impl.client import AsyncHTTPClient
-
-        self._refetch()
-        if self._schema.status != DeploymentStatus.Running:
-            raise BentoMLException(f"Deployment status is {self._schema.status}")
-        if self._urls is None or len(self._urls) != 1:
-            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 = 300, check_interval: int = 5) -> None:
-        start_time = time.time()
-        while time.time() - start_time < timeout:
-            status = self.get_status()
-            if 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}'."
-            )
-            time.sleep(check_interval)
-
-        raise TimeoutError(
-            f"Timed out waiting for deployment '{self.name}' to be ready."
+            name=res.name,
+            # TODO: update this after the url in the frontend is fixed
+            admin_console=f"{client.v1.endpoint}/clusters/{res.cluster.name}/namespaces/{res.kube_namespace}/deployments/{res.name}",
+            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,
         )
 
     @classmethod
     def list(
         cls,
         context: str | None = None,
-        cluster_name: str | None = None,
+        cluster: str | None = None,
         search: str | None = None,
-    ) -> list[Deployment]:
+    ) -> list[DeploymentInfo]:
         cloud_rest_client = get_rest_api_client(context)
-        if cluster_name is None:
+        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")
@@ -346,64 +442,54 @@ class Deployment:
             if res is None:
                 raise BentoMLException("List deployments request failed")
         else:
-            res_count = cloud_rest_client.v2.list_deployment(
-                cluster_name, search=search
-            )
+            res_count = cloud_rest_client.v2.list_deployment(cluster, search=search)
             if res_count is None:
-                raise NotFound(f"Cluster {cluster_name} is not found")
+                raise NotFound(f"Cluster {cluster} is not found")
             if res_count.total == 0:
                 return []
             res = cloud_rest_client.v2.list_deployment(
-                cluster_name, search=search, count=res_count.total
+                cluster, search=search, count=res_count.total
             )
             if res is None:
                 raise BentoMLException("List deployments request failed")
-        return [
-            Deployment(
-                name=schema.name,
-                context=context,
-                cluster_name=schema.cluster.name,
-                _schema=schema,
-            )
-            for schema in res.items
-        ]
+        return [cls._generate_deployment_info_(context, schema) for schema in res.items]
 
     @classmethod
     def create(
         cls,
-        bento: Tag,
+        bento: Tag | str | None = None,
         access_type: str | None = None,
         name: str | None = None,
-        cluster_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_dct: 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,
         context: str | None = None,
-    ) -> Deployment:
+    ) -> DeploymentInfo:
         cloud_rest_client = get_rest_api_client(context)
-        dct: dict[str, t.Any] = {
+        dict: dict[str, t.Any] = {
             "bento": str(bento),
         }
         if name:
-            dct["name"] = name
+            dict["name"] = name
         else:
             # the cloud takes care of the name
-            dct["name"] = ""
+            dict["name"] = ""
 
-        if config_dct:
-            merging_dct = config_dct
+        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_dct = yaml.safe_load(file)
+                    merging_dict = yaml.safe_load(file)
             except FileNotFoundError:
                 raise ValueError(f"File not found: {real_path}")
             except yaml.YAMLError as exc:
@@ -415,83 +501,77 @@ class Deployment:
                 )
         elif config_file is not None:
             try:
-                merging_dct = yaml.safe_load(config_file)
+                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_dct = {
+            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,
-                "cluster": cluster_name,
             }
-        dct.update(merging_dct)
-
-        # add cluster
-        if "cluster" not in dct or dct["cluster"] is None:
-            cluster_name = cls._get_default_cluster(context)
-            dct["cluster"] = cluster_name
-
-        if "distributed" not in dct:
-            dct["distributed"] = (
-                "services" in dct
-                and dct["services"] is not None
-                and dct["services"] != {}
+        dict.update(merging_dict)
+
+        if "distributed" not in dict:
+            dict["distributed"] = (
+                "services" in dict
+                and dict["services"] is not None
+                and dict["services"] != {}
             )
 
-        config_struct = bentoml_cattr.structure(dct, CreateDeploymentSchemaV2)
-        cls._fix_and_validate_schema(config_struct, dct["distributed"])
+        config_struct = bentoml_cattr.structure(dict, CreateDeploymentSchemaV2)
+        cls._fix_and_validate_schema(config_struct, dict["distributed"])
 
         res = cloud_rest_client.v2.create_deployment(
-            create_schema=config_struct, cluster_name=config_struct.cluster
+            create_schema=config_struct, cluster=cluster
         )
+
         logger.debug("Deployment Schema: %s", config_struct)
-        return Deployment(
-            context=context,
-            cluster_name=config_struct.cluster,
-            name=res.name,
-            _schema=res,
-        )
+
+        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def update(
         cls,
-        name: str,
+        name: str | None,
         bento: Tag | str | None = None,
         access_type: str | None = None,
-        cluster_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_dct: 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,
         context: str | None = None,
-    ) -> Deployment:
-        deployment = Deployment.get(
-            name=name, context=context, cluster_name=cluster_name
-        )
-        orig_dct = deployment._conver_schema_to_update_schema()
-        distributed = deployment._schema.distributed
+    ) -> DeploymentInfo:
+        if name is None:
+            raise ValueError("name is required")
         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_dct["bento"] = str(bento)
+            orig_dict["bento"] = str(bento)
 
-        if config_dct:
-            merging_dct = config_dct
+        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_dct = yaml.safe_load(file)
+                    merging_dict = yaml.safe_load(file)
             except FileNotFoundError:
                 raise ValueError(f"File not found: {real_path}")
             except yaml.YAMLError as exc:
@@ -503,90 +583,87 @@ class Deployment:
                 )
         elif config_file is not None:
             try:
-                merging_dct = yaml.safe_load(config_file)
+                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_dct: dict[str, t.Any] = {"scaling": {}}
+            merging_dict: dict[str, t.Any] = {"scaling": {}}
             if scaling_min is not None:
-                merging_dct["scaling"]["min_replicas"] = scaling_min
+                merging_dict["scaling"]["min_replicas"] = scaling_min
             if scaling_max is not None:
-                merging_dct["scaling"]["max_replicas"] = scaling_max
+                merging_dict["scaling"]["max_replicas"] = scaling_max
             if instance_type is not None:
-                merging_dct["instance_type"] = instance_type
+                merging_dict["instance_type"] = instance_type
 
             if strategy is not None:
-                merging_dct["deployment_strategy"] = strategy
+                merging_dict["deployment_strategy"] = strategy
 
             if envs is not None:
-                merging_dct["envs"] = envs
+                merging_dict["envs"] = envs
 
             if extras is not None:
-                merging_dct["extras"] = extras
+                merging_dict["extras"] = extras
 
             if access_type is not None:
-                merging_dct["access_type"] = access_type
+                merging_dict["access_type"] = access_type
 
-        config_merger.merge(orig_dct, merging_dct)
+        config_merger.merge(orig_dict, merging_dict)
 
-        config_struct = bentoml_cattr.structure(orig_dct, UpdateDeploymentSchemaV2)
+        config_struct = bentoml_cattr.structure(orig_dict, UpdateDeploymentSchemaV2)
 
         cls._fix_and_validate_schema(config_struct, distributed)
 
         res = cloud_rest_client.v2.update_deployment(
-            cluster_name=deployment.cluster_name,
-            deployment_name=name,
+            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)
-        deployment._schema = res
-        deployment._urls = res.urls
-        return deployment
+        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def apply(
         cls,
-        name: str,
-        bento: Tag | None = None,
-        cluster_name: str | None = None,
-        config_dct: dict[str, t.Any] | None = None,
-        config_file: str | None = None,
+        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,
         context: str | None = None,
-    ) -> Deployment:
-        try:
-            deployment = Deployment.get(
-                name=name, context=context, cluster_name=cluster_name
-            )
-        except NotFound as e:
+    ) -> 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:
                 return cls.create(
                     bento=bento,
                     name=name,
-                    cluster_name=cluster_name,
-                    config_dct=config_dct,
+                    cluster=cluster,
+                    config_dict=config_dict,
                     config_file=config_file,
                     path_context=path_context,
                     context=context,
                 )
             else:
-                raise e
-        cloud_rest_client = get_rest_api_client(context)
+                raise NotFound(f"deployment {name} is not found")
         if bento is None:
-            bento = deployment._conver_schema_to_bento()
+            bento = cls._convert_schema_to_bento(_schema=res)
 
-        schema_dct: dict[str, t.Any] = {"bento": str(bento)}
-        distributed = deployment._schema.distributed
+        schema_dict: dict[str, t.Any] = {"bento": str(bento)}
+        distributed = res.distributed
 
-        if config_file:
+        if isinstance(config_file, str):
             real_path = resolve_user_filepath(config_file, path_context)
             try:
                 with open(real_path, "r") as file:
-                    config_dct = yaml.safe_load(file)
+                    config_dict = yaml.safe_load(file)
             except FileNotFoundError:
                 raise ValueError(f"File not found: {real_path}")
             except yaml.YAMLError as exc:
@@ -596,79 +673,63 @@ class Deployment:
                 raise ValueError(
                     f"An error occurred while reading the file: {real_path}\n{e}"
                 )
-        if config_dct is None:
+        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")
 
-        schema_dct.update(config_dct)
-        config_struct = bentoml_cattr.structure(schema_dct, UpdateDeploymentSchemaV2)
+        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(
-            deployment_name=name,
+            name=name,
             update_schema=config_struct,
-            cluster_name=deployment.cluster_name,
+            cluster=res.cluster.host_cluster_display_name,
         )
         if res is None:
             raise NotFound(f"deployment {name} is not found")
         logger.debug("Deployment Schema: %s", config_struct)
-        deployment._schema = res
-        deployment._urls = res.urls
-        return deployment
+        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def get(
         cls,
         name: str,
+        cluster: str | None = None,
         context: str | None = None,
-        cluster_name: str | None = None,
-    ) -> Deployment:
-        if cluster_name is None:
-            cluster_name = cls._get_default_cluster(context)
+    ) -> DeploymentInfo:
         cloud_rest_client = get_rest_api_client(context)
-        res = cloud_rest_client.v2.get_deployment(cluster_name, name)
+        res = cloud_rest_client.v2.get_deployment(name, cluster)
         if res is None:
             raise NotFound(f"deployment {name} is not found")
-
-        deployment = Deployment(
-            context=context,
-            cluster_name=cluster_name,
-            name=name,
-            _schema=res,
-            _urls=res.urls,
-        )
-        return deployment
+        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def terminate(
         cls,
         name: str,
+        cluster: str | None = None,
         context: str | None = None,
-        cluster_name: str | None = None,
-    ) -> Deployment:
+    ) -> DeploymentInfo:
         cloud_rest_client = get_rest_api_client(context)
-        if cluster_name is None:
-            cluster_name = cls._get_default_cluster(context)
-        res = cloud_rest_client.v2.terminate_deployment(cluster_name, name)
+        res = cloud_rest_client.v2.terminate_deployment(name, cluster)
         if res is None:
             raise NotFound(f"Deployment {name} is not found")
-        return Deployment(
-            name=name,
-            cluster_name=cluster_name,
-            context=context,
-            _schema=res,
-            _urls=res.urls,
-        )
+        return cls._generate_deployment_info_(context, res, res.urls)
 
     @classmethod
     def delete(
         cls,
         name: str,
+        cluster: str | None = None,
         context: str | None = None,
-        cluster_name: str | None = None,
     ) -> None:
         cloud_rest_client = get_rest_api_client(context)
-        if cluster_name is None:
-            cluster_name = cls._get_default_cluster(context)
-        res = cloud_rest_client.v2.delete_deployment(cluster_name, name)
+        res = cloud_rest_client.v2.delete_deployment(name, cluster)
         if res is None:
             raise NotFound(f"Deployment {name} is not found")
diff --git a/src/bentoml/_internal/cloud/schemas/modelschemas.py b/src/bentoml/_internal/cloud/schemas/modelschemas.py
index 2a2dfdc0..3a3e9852 100644
--- a/src/bentoml/_internal/cloud/schemas/modelschemas.py
+++ b/src/bentoml/_internal/cloud/schemas/modelschemas.py
@@ -195,6 +195,12 @@ class LabelItemSchema:
     value: str
 
 
+@attr.define
+class EnvItemSchema:
+    name: str
+    value: str
+
+
 class HPAMetricType(Enum):
     MEMORY = "memory"
     CPU = "cpu"
@@ -367,7 +373,7 @@ class DeploymentServiceConfig:
     __forbid_extra_keys__ = True
     instance_type: t.Optional[str] = attr.field(default=None)
     scaling: t.Optional[DeploymentTargetHPAConf] = attr.field(default=None)
-    envs: t.Optional[t.List[t.Optional[LabelItemSchema]]] = attr.field(default=None)
+    envs: t.Optional[t.List[t.Optional[EnvItemSchema]]] = attr.field(default=None)
     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)
diff --git a/src/bentoml/_internal/cloud/schemas/schemasv1.py b/src/bentoml/_internal/cloud/schemas/schemasv1.py
index 79c4d2ec..77530792 100644
--- a/src/bentoml/_internal/cloud/schemas/schemasv1.py
+++ b/src/bentoml/_internal/cloud/schemas/schemasv1.py
@@ -71,6 +71,8 @@ class OrganizationListSchema(BaseListSchema):
 @attr.define
 class ClusterSchema(ResourceSchema):
     description: str
+    host_cluster_display_name: str | None
+    organization_name: str
     creator: UserSchema
 
 
diff --git a/src/bentoml/_internal/cloud/schemas/schemasv2.py b/src/bentoml/_internal/cloud/schemas/schemasv2.py
index d81a9e21..00cb7b1a 100644
--- a/src/bentoml/_internal/cloud/schemas/schemasv2.py
+++ b/src/bentoml/_internal/cloud/schemas/schemasv2.py
@@ -19,17 +19,10 @@ from bentoml._internal.cloud.schemas.schemasv1 import UserSchema
 @attr.define
 class DeploymentTargetSchema(ResourceSchema):
     creator: t.Optional[UserSchema]
-    config: t.Optional[DeploymentTargetConfig]
+    config: t.Optional[DeploymentConfigSchema]
     bento: t.Optional[BentoWithRepositorySchema]
-
-
-@attr.define
-class DeploymentTargetConfig(DeploymentServiceConfig):
     kube_resource_uid: t.Optional[str] = attr.field(default=None)
     kube_resource_version: t.Optional[str] = attr.field(default=None)
-    services: t.Dict[str, DeploymentServiceConfig] = attr.field(factory=dict)
-    access_type: t.Optional[AccessControl] = attr.field(default=None)
-    bentoml_config_overrides: t.Dict[str, t.Optional[t.Any]] = attr.field(factory=dict)
 
 
 @attr.define
@@ -49,15 +42,22 @@ class DeploymentRevisionListSchema(BaseListSchema):
     items: t.List[t.Optional[DeploymentRevisionSchema]]
 
 
+@attr.define
+class DeploymentConfigSchema(DeploymentServiceConfig):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = False
+    access_type: t.Optional[AccessControl] = 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)
-class UpdateDeploymentSchema(DeploymentServiceConfig):
+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)
-    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)
@@ -65,7 +65,6 @@ class CreateDeploymentSchema(UpdateDeploymentSchema):
     __omit_if_default__ = True
     __forbid_extra_keys__ = True
     name: str
-    cluster: str
     distributed: bool
 
 
@@ -75,11 +74,12 @@ class DeploymentSchema(ResourceSchema):
     __forbid_extra_keys__ = True
     status: DeploymentStatus
     kube_namespace: str
-    creator: t.Optional[UserSchema]
-    cluster: t.Optional[ClusterSchema]
+    creator: UserSchema
+    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/deployment.py b/src/bentoml/deployment.py
index 22fd456f..b400ce50 100644
--- a/src/bentoml/deployment.py
+++ b/src/bentoml/deployment.py
@@ -10,6 +10,8 @@ from simple_di import Provide
 from simple_di import inject
 
 from bentoml._internal.cloud.deployment import Deployment
+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
@@ -29,7 +31,7 @@ def create(
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = ...,
-    cluster_name: str | None = ...,
+    cluster: str | None = ...,
     access_type: str | None = ...,
     scaling_min: int | None = ...,
     scaling_max: int | None = ...,
@@ -37,7 +39,7 @@ def create(
     strategy: str | None = ...,
     envs: t.List[dict[str, t.Any]] | None = ...,
     extras: dict[str, t.Any] | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
@@ -49,7 +51,7 @@ def create(
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
-    cluster_name: str | None = ...,
+    cluster: str | None = ...,
     access_type: str | None = ...,
     scaling_min: int | None = ...,
     scaling_max: int | None = ...,
@@ -57,7 +59,7 @@ def create(
     strategy: str | None = ...,
     envs: t.List[dict[str, t.Any]] | None = ...,
     extras: dict[str, t.Any] | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
@@ -70,7 +72,7 @@ def create(
     *,
     bento: Tag | str | None = ...,
     config_file: str | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
@@ -83,7 +85,7 @@ def create(
     *,
     project_path: str | None = ...,
     config_file: str | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
@@ -95,8 +97,8 @@ def create(
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
-    config_dct: dict[str, t.Any] | None = ...,
-) -> Deployment:
+    config_dict: dict[str, t.Any] | None = ...,
+) -> DeploymentInfo:
     ...
 
 
@@ -108,8 +110,8 @@ def create(
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = ...,
-    config_dct: dict[str, t.Any] | None = ...,
-) -> Deployment:
+    config_dict: dict[str, t.Any] | None = ...,
+) -> DeploymentInfo:
     ...
 
 
@@ -122,7 +124,7 @@ def create(
     *,
     project_path: str | None = None,
     bento: Tag | str | None = None,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     access_type: str | None = None,
     scaling_min: int | None = None,
     scaling_max: int | None = None,
@@ -130,12 +132,12 @@ def create(
     strategy: str | None = None,
     envs: t.List[dict[str, t.Any]] | None = None,
     extras: dict[str, t.Any] | None = None,
-    config_dct: dict[str, t.Any] | None = None,
+    config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
-) -> Deployment:
+) -> DeploymentInfo:
     deploy_by_param = (
         access_type
-        or cluster_name
+        or cluster
         or scaling_min
         or scaling_max
         or instance_type
@@ -144,23 +146,31 @@ def create(
         or extras
     )
     if (
-        config_dct
+        config_dict
         and config_file
-        or config_dct
+        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_dct, config_file, or the other parameters"
+            "Configure a deployment can only use one of the following: config_dict, config_file, or the other parameters"
         )
-    if bento and project_path:
+    deploy_name, bento_name, cluster_name = get_args_from_config(
+        bento=str(bento),
+        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 is None and project_path is None:
+    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,
+        bento=bento_name,
         context=context,
         _cloud_client=BentoCloudClient(),
     )
@@ -168,15 +178,15 @@ def create(
     return Deployment.create(
         bento=bento,
         access_type=access_type,
-        name=name,
-        cluster_name=cluster_name,
+        name=deploy_name,
+        cluster=cluster_name,
         scaling_min=scaling_min,
         scaling_max=scaling_max,
         instance_type=instance_type,
         strategy=strategy,
         envs=envs,
         extras=extras,
-        config_dct=config_dct,
+        config_dict=config_dict,
         config_file=config_file,
         path_context=path_context,
         context=context,
@@ -185,10 +195,10 @@ def create(
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = ...,
+    cluster: str | None = ...,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = ...,
@@ -199,16 +209,16 @@ def update(
     strategy: str | None = ...,
     envs: t.List[dict[str, t.Any]] | None = ...,
     extras: dict[str, t.Any] | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = ...,
+    cluster: str | None = ...,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
@@ -219,72 +229,72 @@ def update(
     strategy: str | None = ...,
     envs: t.List[dict[str, t.Any]] | None = ...,
     extras: dict[str, t.Any] | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = ...,
     config_file: str | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
     config_file: str | None = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = ...,
-    config_dct: dict[str, t.Any] | None = ...,
-) -> Deployment:
+    config_dict: dict[str, t.Any] | None = ...,
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def update(
-    name: str,
+    name: str | None = ...,
     path_context: str | None = ...,
     context: str | None = ...,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: Tag | str | None = ...,
-    config_dct: dict[str, t.Any] | None = ...,
-) -> Deployment:
+    config_dict: dict[str, t.Any] | None = ...,
+) -> DeploymentInfo:
     ...
 
 
 @inject
 def update(
-    name: str,
+    name: str | None = None,
     path_context: str | None = None,
     context: str | None = None,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = None,
@@ -296,9 +306,9 @@ def update(
     strategy: str | None = None,
     envs: t.List[dict[str, t.Any]] | None = None,
     extras: dict[str, t.Any] | None = None,
-    config_dct: dict[str, t.Any] | None = None,
+    config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
-) -> Deployment:
+) -> DeploymentInfo:
     deploy_by_param = (
         access_type
         or scaling_min
@@ -309,24 +319,32 @@ def update(
         or extras
     )
     if (
-        config_dct
+        config_dict
         and config_file
-        or config_dct
+        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_dct, config_file, or the other parameters"
+            "Configure a deployment can only use one of the following: config_dict, config_file, or the other parameters"
         )
-    if bento and project_path:
+    deploy_name, bento_name, cluster_name = get_args_from_config(
+        bento=str(bento),
+        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 is None and project_path is None:
-        bento = None
+    if bento_name is None and project_path is None:
+        bento_name = None
     else:
         bento = get_real_bento_tag(
             project_path=project_path,
-            bento=bento,
+            bento=bento_name,
             context=context,
             _cloud_client=BentoCloudClient(),
         )
@@ -334,15 +352,15 @@ def update(
     return Deployment.update(
         bento=bento,
         access_type=access_type,
-        name=name,
-        cluster_name=cluster_name,
+        name=deploy_name,
+        cluster=cluster_name,
         scaling_min=scaling_min,
         scaling_max=scaling_max,
         instance_type=instance_type,
         strategy=strategy,
         envs=envs,
         extras=extras,
-        config_dct=config_dct,
+        config_dict=config_dict,
         config_file=config_file,
         path_context=path_context,
         context=context,
@@ -351,96 +369,104 @@ def update(
 
 @t.overload
 def apply(
-    name: str,
-    cluster_name: t.Optional[str] = ...,
+    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_dct: t.Optional[dict[str, t.Any]] = ...,
-) -> Deployment:
+    config_dict: t.Optional[dict[str, t.Any]] = ...,
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def apply(
-    name: str,
-    cluster_name: t.Optional[str] = ...,
+    name: str | None = ...,
+    cluster: t.Optional[str] = ...,
     path_context: t.Optional[str] = ...,
     context: t.Optional[str] = ...,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: t.Optional[t.Union[Tag, str]] = ...,
-    config_dct: t.Optional[dict[str, t.Any]] = ...,
-) -> Deployment:
+    config_dict: t.Optional[dict[str, t.Any]] = ...,
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def apply(
-    name: str,
-    cluster_name: t.Optional[str] = ...,
+    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] = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @t.overload
 def apply(
-    name: str,
-    cluster_name: t.Optional[str] = ...,
+    name: str | None = ...,
+    cluster: t.Optional[str] = ...,
     path_context: t.Optional[str] = ...,
     context: t.Optional[str] = ...,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     bento: t.Optional[t.Union[Tag, str]] = ...,
     config_file: t.Optional[str] = ...,
-) -> Deployment:
+) -> DeploymentInfo:
     ...
 
 
 @inject
 def apply(
-    name: str,
-    cluster_name: str | None = None,
+    name: str | None = None,
+    cluster: str | None = None,
     path_context: str | None = None,
     context: str | None = None,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     *,
     project_path: str | None = None,
     bento: Tag | str | None = None,
-    config_dct: dict[str, t.Any] | None = None,
+    config_dict: dict[str, t.Any] | None = None,
     config_file: str | None = None,
-) -> Deployment:
-    if config_dct and config_file:
+) -> DeploymentInfo:
+    if config_dict and config_file:
         raise BentoMLException(
-            "Configure a deployment can only use one of the following: config_dct, config_file"
+            "Configure a deployment can only use one of the following: config_dict, config_file"
         )
-    if bento and project_path:
+    deploy_name, bento_name, cluster_name = get_args_from_config(
+        bento=str(bento),
+        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 is None and project_path is None:
-        bento = None
+    if bento_name is None and project_path is None:
+        bento_name = None
     else:
         bento = get_real_bento_tag(
             project_path=project_path,
-            bento=bento,
+            bento=bento_name,
             context=context,
             _cloud_client=BentoCloudClient(),
         )
 
     return Deployment.apply(
-        name=name,
-        bento=bento,
-        cluster_name=cluster_name,
+        name=deploy_name,
+        bento=bento_name,
+        cluster=cluster_name,
         context=context,
         path_context=path_context,
-        config_dct=config_dct,
+        config_dict=config_dict,
         config_file=config_file,
     )
 
@@ -448,45 +474,45 @@ def apply(
 def get(
     name: str,
     context: str | None = None,
-    cluster_name: str | None = None,
-) -> Deployment:
+    cluster: str | None = None,
+) -> DeploymentInfo:
     return Deployment.get(
         name=name,
         context=context,
-        cluster_name=cluster_name,
+        cluster=cluster,
     )
 
 
 def terminate(
     name: str,
     context: str | None = None,
-    cluster_name: str | None = None,
-) -> Deployment:
+    cluster: str | None = None,
+) -> DeploymentInfo:
     return Deployment.terminate(
         name=name,
         context=context,
-        cluster_name=cluster_name,
+        cluster=cluster,
     )
 
 
 def delete(
     name: str,
     context: str | None = None,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
 ) -> None:
     Deployment.delete(
         name=name,
         context=context,
-        cluster_name=cluster_name,
+        cluster=cluster,
     )
 
 
 def list(
     context: str | None = None,
-    cluster_name: str | None = None,
+    cluster: str | None = None,
     search: str | None = None,
-) -> t.List[Deployment]:
-    return Deployment.list(context=context, cluster_name=cluster_name, search=search)
+) -> t.List[DeploymentInfo]:
+    return Deployment.list(context=context, cluster=cluster, search=search)
 
 
 __all__ = ["create", "get", "update", "apply", "terminate", "delete", "list"]
diff --git a/src/bentoml_cli/deployment.py b/src/bentoml_cli/deployment.py
index 226cc7c0..da9d33cf 100644
--- a/src/bentoml_cli/deployment.py
+++ b/src/bentoml_cli/deployment.py
@@ -6,6 +6,7 @@ import click
 import yaml
 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.schemas.modelschemas import DeploymentStrategy
 
@@ -32,7 +33,7 @@ def add_deployment_command(cli: click.Group) -> None:
     @click.argument(
         "target",
         type=click.STRING,
-        required=True,
+        required=False,
     )
     @click.option(
         "-n",
@@ -81,6 +82,7 @@ def add_deployment_command(cli: click.Group) -> None:
         multiple=True,
     )
     @click.option(
+        "-f",
         "--config-file",
         type=click.File(),
         help="Configuration file path",
@@ -102,7 +104,7 @@ def add_deployment_command(cli: click.Group) -> None:
     @click.pass_obj
     def deploy(
         shared_options: SharedOptions,
-        target: str,
+        target: str | None,
         name: str | None,
         cluster: str | None,
         access_type: str | None,
@@ -111,7 +113,7 @@ def add_deployment_command(cli: click.Group) -> None:
         instance_type: str | None,
         strategy: str | None,
         env: tuple[str] | None,
-        config_file: t.TextIO | None,
+        config_file: str | t.TextIO | None,
         wait: bool,
         timeout: int,
     ) -> None:
@@ -122,8 +124,16 @@ def add_deployment_command(cli: click.Group) -> None:
         """
         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 path.exists(target):
+        if path.exists(bento_name):
             # target is a path
             click.echo(f"building bento from {target} ...")
             bento_tag = get_real_bento_tag(project_path=target)
@@ -133,8 +143,8 @@ def add_deployment_command(cli: click.Group) -> None:
 
         deployment = Deployment.create(
             bento=bento_tag,
-            name=name,
-            cluster_name=cluster,
+            name=deploy_name,
+            cluster=cluster_name,
             access_type=access_type,
             scaling_min=scaling_min,
             scaling_max=scaling_max,
@@ -151,11 +161,9 @@ def add_deployment_command(cli: click.Group) -> None:
         if wait:
             deployment.wait_until_ready(timeout=timeout)
             click.echo(
-                f"Deployment '{deployment.name}' created successfully in cluster '{deployment.cluster_name}'."
+                f"Deployment '{deployment.name}' created successfully in cluster '{deployment.cluster}'."
             )
-        click.echo(
-            f"To check the deployment, go to: {deployment.get_bento_cloud_url()}."
-        )
+        click.echo(f"To check the deployment, go to: {deployment.admin_console}.")
 
     output_option = click.option(
         "-o",
@@ -166,17 +174,10 @@ def add_deployment_command(cli: click.Group) -> None:
     )
 
     def shared_decorator(
-        f: t.Callable[..., t.Any] | None = None,
-        *,
-        required_deployment_name: bool = True,
+        f: t.Callable[..., t.Any] | None = None
     ) -> t.Callable[..., t.Any]:
         def decorate(f: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
             options = [
-                click.argument(
-                    "name",
-                    type=click.STRING,
-                    required=required_deployment_name,
-                ),
                 click.option(
                     "--cluster",
                     type=click.STRING,
@@ -198,7 +199,7 @@ def add_deployment_command(cli: click.Group) -> None:
         """Deployment Subcommands Groups"""
 
     @deployment_cli.command()
-    @shared_decorator(required_deployment_name=True)
+    @shared_decorator()
     @cog.optgroup.group(cls=cog.MutuallyExclusiveOptionGroup, name="target options")
     @cog.optgroup.option(
         "--bento",
@@ -246,6 +247,7 @@ def add_deployment_command(cli: click.Group) -> None:
         multiple=True,
     )
     @click.option(
+        "-f",
         "--config-file",
         type=click.File(),
         help="Configuration file path, mututally exclusive with other config options",
@@ -254,7 +256,7 @@ def add_deployment_command(cli: click.Group) -> None:
     @click.pass_obj
     def update(  # type: ignore
         shared_options: SharedOptions,
-        name: str,
+        name: str | None,
         cluster: str | None,
         project_path: str | None,
         bento: str | None,
@@ -272,7 +274,10 @@ def add_deployment_command(cli: click.Group) -> None:
         A deployment can be updated using parameters (standalone mode only), or using config yaml file.
         You can also update bento by providing a project path or existing bento.
         """
-        if bento is None and project_path is None:
+        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(
@@ -284,8 +289,8 @@ def add_deployment_command(cli: click.Group) -> None:
         Deployment.update(
             bento=target,
             access_type=access_type,
-            name=name,
-            cluster_name=cluster,
+            name=deploy_name,
+            cluster=cluster_name,
             scaling_min=scaling_min,
             scaling_max=scaling_max,
             instance_type=instance_type,
@@ -301,8 +306,59 @@ def add_deployment_command(cli: click.Group) -> None:
 
         click.echo(f"Deployment '{name}' updated successfully.")
 
+    @deployment_cli.command()
+    @cog.optgroup.group(cls=cog.MutuallyExclusiveOptionGroup, name="target options")
+    @click.option(
+        "-f",
+        "--config-file",
+        type=click.File(),
+        help="Configuration file path, mututally exclusive with other config options",
+        default=None,
+    )
+    @click.pass_obj
+    def apply(  # type: ignore
+        shared_options: SharedOptions,
+        name: str | None,
+        cluster: str | None,
+        project_path: str | None,
+        bento: str | None,
+        config_file: str | t.TextIO | None,
+    ) -> None:
+        """Update a deployment on BentoCloud.
+
+        \b
+        A deployment can be updated using parameters (standalone mode only), 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.apply(
+            bento=target,
+            name=deploy_name,
+            cluster=cluster_name,
+            config_file=config_file,
+            context=shared_options.cloud_context,
+        )
+
+        click.echo(f"Deployment '{name}' updated successfully.")
+
     @deployment_cli.command()
     @shared_decorator
+    @click.argument(
+        "name",
+        type=click.STRING,
+        required=True,
+    )
     @output_option
     @click.pass_obj
     def get(  # type: ignore
@@ -312,18 +368,21 @@ def add_deployment_command(cli: click.Group) -> None:
         output: t.Literal["json", "default"],
     ) -> None:
         """Get a deployment on BentoCloud."""
-        d = Deployment.get(
-            name, context=shared_options.cloud_context, cluster_name=cluster
-        )
+        d = Deployment.get(name, context=shared_options.cloud_context, cluster=cluster)
         if output == "json":
-            info = json.dumps(d.info.to_dict(), indent=2, default=str)
+            info = json.dumps(d.to_dict(), indent=2, default=str)
             console.print_json(info)
         else:
-            info = yaml.dump(d.info.to_dict(), indent=2, sort_keys=False)
+            info = yaml.dump(d.to_dict(), indent=2, sort_keys=False)
             console.print(Syntax(info, "yaml", background_color="default"))
 
     @deployment_cli.command()
     @shared_decorator
+    @click.argument(
+        "name",
+        type=click.STRING,
+        required=True,
+    )
     @click.pass_obj
     def terminate(  # type: ignore
         shared_options: SharedOptions,
@@ -332,7 +391,7 @@ def add_deployment_command(cli: click.Group) -> None:
     ) -> None:
         """Terminate a deployment on BentoCloud."""
         Deployment.terminate(
-            name, context=shared_options.cloud_context, cluster_name=cluster
+            name, context=shared_options.cloud_context, cluster=cluster
         )
         click.echo(f"Deployment '{name}' terminated successfully.")
 
@@ -345,9 +404,7 @@ def add_deployment_command(cli: click.Group) -> None:
         cluster: str | None,
     ) -> None:
         """Delete a deployment on BentoCloud."""
-        Deployment.delete(
-            name, context=shared_options.cloud_context, cluster_name=cluster
-        )
+        Deployment.delete(name, context=shared_options.cloud_context, cluster=cluster)
         click.echo(f"Deployment '{name}' deleted successfully.")
 
     @deployment_cli.command()
@@ -373,21 +430,21 @@ def add_deployment_command(cli: click.Group) -> None:
     ) -> None:
         """List existing deployments on BentoCloud."""
         d_list = Deployment.list(
-            context=shared_options.cloud_context, cluster_name=cluster, search=search
+            context=shared_options.cloud_context, cluster=cluster, search=search
         )
-        res: list[dict[str, t.Any]] = [d.info.to_dict() for d in d_list]
+        res: list[dict[str, t.Any]] = [d.to_dict() for d in d_list]
         if output == "table":
             table = Table(box=None)
             table.add_column("Deployment")
             table.add_column("created_at")
             table.add_column("Bento")
             table.add_column("Status")
-            for info in res:
+            for info in d_list:
                 table.add_row(
-                    info["name"],
-                    info["created_at"],
-                    info["bento"],
-                    info["status"],
+                    info.name,
+                    info.created_at,
+                    info.get_bento(refetch=False),
+                    info.get_status(refetch=False).status,
                 )
             console.print(table)
         elif output == "json":


Test patch:
diff --git a/tests/unit/_internal/cloud/test_deployment.py b/tests/unit/_internal/cloud/test_deployment.py
index 4bd5fd96..d79b1fc5 100644
--- a/tests/unit/_internal/cloud/test_deployment.py
+++ b/tests/unit/_internal/cloud/test_deployment.py
@@ -13,6 +13,7 @@ from bentoml._internal.cloud.schemas.modelschemas import AccessControl
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentServiceConfig
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentStrategy
 from bentoml._internal.cloud.schemas.modelschemas import DeploymentTargetHPAConf
+from bentoml._internal.cloud.schemas.modelschemas import EnvItemSchema
 from bentoml._internal.cloud.schemas.schemasv1 import BentoFullSchema
 from bentoml._internal.cloud.schemas.schemasv1 import BentoImageBuildStatus
 from bentoml._internal.cloud.schemas.schemasv1 import BentoManifestSchema
@@ -22,20 +23,19 @@ from bentoml._internal.cloud.schemas.schemasv1 import ClusterListSchema
 from bentoml._internal.cloud.schemas.schemasv1 import ClusterSchema
 from bentoml._internal.cloud.schemas.schemasv1 import DeploymentRevisionStatus
 from bentoml._internal.cloud.schemas.schemasv1 import DeploymentStatus
-from bentoml._internal.cloud.schemas.schemasv1 import LabelItemSchema
 from bentoml._internal.cloud.schemas.schemasv1 import ResourceType
 from bentoml._internal.cloud.schemas.schemasv1 import UserSchema
 from bentoml._internal.cloud.schemas.schemasv2 import (
     CreateDeploymentSchema as CreateDeploymentSchemaV2,
 )
 from bentoml._internal.cloud.schemas.schemasv2 import (
-    DeploymentFullSchema as DeploymentFullSchemaV2,
+    DeploymentConfigSchema as DeploymentConfigSchemaV2,
 )
 from bentoml._internal.cloud.schemas.schemasv2 import (
-    DeploymentRevisionSchema as DeploymentRevisionSchemaV2,
+    DeploymentFullSchema as DeploymentFullSchemaV2,
 )
 from bentoml._internal.cloud.schemas.schemasv2 import (
-    DeploymentTargetConfig as DeploymentTargetConfigV2,
+    DeploymentRevisionSchema as DeploymentRevisionSchemaV2,
 )
 from bentoml._internal.cloud.schemas.schemasv2 import (
     DeploymentTargetSchema as DeploymentTargetSchemaV2,
@@ -55,225 +55,186 @@ class DummyUpdateSchema(UpdateDeploymentSchemaV2):
     )  # place holder for urls that's assigned to deployment._urls
 
 
+def dummy_generate_deployment_schema(
+    name: str,
+    cluster: str | None,
+    distributed: bool,
+    update_schema: UpdateDeploymentSchemaV2,
+):
+    user = UserSchema(name="", email="", first_name="", last_name="")
+    bento = update_schema.bento.split(":")
+    if len(bento) == 2:
+        bento_name = bento[0]
+        bento_version = bento[1]
+    else:
+        bento_name = bento[0]
+        bento_version = ""
+    dummy_bento = BentoFullSchema(
+        uid="",
+        created_at=datetime(2023, 5, 25),
+        updated_at=None,
+        deleted_at=None,
+        name=bento_name,
+        resource_type=ResourceType.BENTO,
+        labels=[],
+        description="",
+        version=bento_version,
+        image_build_status=BentoImageBuildStatus.PENDING,
+        upload_status=BentoUploadStatus.SUCCESS,
+        upload_finished_reason="",
+        presigned_upload_url="",
+        presigned_download_url="",
+        manifest=BentoManifestSchema(
+            service="",
+            bentoml_version="",
+            size_bytes=0,
+            apis={},
+            models=["iris_clf:ddaex6h2vw6kwcvj"],
+        ),
+        build_at=datetime(2023, 5, 25),
+        repository=BentoRepositorySchema(
+            uid="",
+            created_at=datetime(2023, 5, 1),
+            updated_at=None,
+            deleted_at=None,
+            name=bento_name,
+            resource_type=ResourceType.BENTO_REPOSITORY,
+            labels=[],
+            description="",
+            latest_bento=None,
+        ),
+    )
+    return DeploymentFullSchemaV2(
+        distributed=distributed,
+        latest_revision=DeploymentRevisionSchemaV2(
+            targets=[
+                DeploymentTargetSchemaV2(
+                    bento=dummy_bento,
+                    config=DeploymentConfigSchemaV2(
+                        access_type=update_schema.access_type,
+                        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),
+                    updated_at=None,
+                    deleted_at=None,
+                    name=name,
+                    resource_type=ResourceType.DEPLOYMENT_REVISION,
+                    labels=[],
+                    creator=user,
+                )
+            ],
+            uid="",
+            created_at=datetime(2023, 5, 1),
+            updated_at=None,
+            deleted_at=None,
+            name=name,
+            resource_type=ResourceType.DEPLOYMENT_REVISION,
+            labels=[],
+            creator=user,
+            status=DeploymentRevisionStatus.ACTIVE,
+        ),
+        uid="",
+        created_at=datetime(2023, 5, 1),
+        updated_at=None,
+        deleted_at=None,
+        name=name,
+        resource_type=ResourceType.DEPLOYMENT_REVISION,
+        labels=[],
+        creator=user,
+        status=DeploymentStatus.Running,
+        cluster=ClusterSchema(
+            uid="",
+            name="default",
+            host_cluster_display_name=cluster,
+            organization_name="default_organization",
+            resource_type=ResourceType.CLUSTER,
+            labels=[],
+            description="",
+            creator=user,
+            created_at=datetime(2023, 5, 1),
+            updated_at=None,
+            deleted_at=None,
+        ),
+        kube_namespace="",
+    )
+
+
 @pytest.fixture(name="rest_client", scope="function")
 def fixture_rest_client() -> RestApiClient:
     def dummy_create_deployment(
-        create_schema: CreateDeploymentSchemaV2, cluster_name: str
+        create_schema: CreateDeploymentSchemaV2, cluster: str | None = None
     ):
-        return create_schema
+        if cluster is None:
+            cluster = "default_display_name"
+        return dummy_generate_deployment_schema(
+            create_schema.name, cluster, create_schema.distributed, create_schema
+        )
 
     def dummy_update_deployment(
-        update_schema: UpdateDeploymentSchemaV2, cluster_name: str, deployment_name: str
+        name: str,
+        update_schema: UpdateDeploymentSchemaV2,
+        cluster: str | None = None,
     ):
-        from bentoml._internal.utils import bentoml_cattr
-
-        return bentoml_cattr.structure(attr.asdict(update_schema), DummyUpdateSchema)
-
-    def dummy_get_deployment(cluster_name: str, deployment_name: str):
-        if deployment_name == "test-distributed":
-            return DeploymentFullSchemaV2(
-                distributed=True,
-                latest_revision=DeploymentRevisionSchemaV2(
-                    targets=[
-                        DeploymentTargetSchemaV2(
-                            bento=BentoFullSchema(
-                                uid="",
-                                created_at=datetime(2023, 5, 25),
-                                updated_at=None,
-                                deleted_at=None,
-                                name="123",
-                                resource_type=ResourceType.BENTO,
-                                labels=[],
-                                description="",
-                                version="",
-                                image_build_status=BentoImageBuildStatus.PENDING,
-                                upload_status=BentoUploadStatus.SUCCESS,
-                                upload_finished_reason="",
-                                presigned_upload_url="",
-                                presigned_download_url="",
-                                manifest=BentoManifestSchema(
-                                    service="",
-                                    bentoml_version="",
-                                    size_bytes=0,
-                                    apis={},
-                                    models=["iris_clf:ddaex6h2vw6kwcvj"],
-                                ),
-                                build_at=datetime(2023, 5, 25),
-                                repository=BentoRepositorySchema(
-                                    uid="",
-                                    created_at="",
-                                    updated_at=None,
-                                    deleted_at=None,
-                                    name="abc",
-                                    resource_type=ResourceType.BENTO_REPOSITORY,
-                                    labels=[],
-                                    description="",
-                                    latest_bento="",
-                                ),
+        if cluster is None:
+            cluster = "default_display_name"
+        return dummy_generate_deployment_schema(name, cluster, False, update_schema)
+
+    def dummy_get_deployment(
+        name: str,
+        cluster: str | None = None,
+    ):
+        if cluster is None:
+            cluster = "default_display_name"
+        if name == "test-distributed":
+            return dummy_generate_deployment_schema(
+                name,
+                cluster,
+                True,
+                UpdateDeploymentSchemaV2(
+                    bento="abc:123",
+                    envs=[EnvItemSchema(name="env_key", value="env_value")],
+                    services={
+                        "irisclassifier": DeploymentServiceConfig(
+                            instance_type="t3-small",
+                            scaling=DeploymentTargetHPAConf(
+                                min_replicas=1, max_replicas=1
                             ),
-                            config=DeploymentTargetConfigV2(
-                                access_type=AccessControl.PUBLIC,
-                                envs=[
-                                    LabelItemSchema(key="env_key", value="env_value")
-                                ],
-                                services={
-                                    "irisclassifier": DeploymentServiceConfig(
-                                        instance_type="t3-small",
-                                        scaling=DeploymentTargetHPAConf(
-                                            min_replicas=1, max_replicas=1
-                                        ),
-                                        deployment_strategy=DeploymentStrategy.RollingUpdate,
-                                    ),
-                                    "preprocessing": DeploymentServiceConfig(
-                                        instance_type="t3-small",
-                                        scaling=DeploymentTargetHPAConf(
-                                            min_replicas=1, max_replicas=1
-                                        ),
-                                        deployment_strategy=DeploymentStrategy.RollingUpdate,
-                                    ),
-                                },
+                            deployment_strategy=DeploymentStrategy.RollingUpdate,
+                        ),
+                        "preprocessing": DeploymentServiceConfig(
+                            instance_type="t3-small",
+                            scaling=DeploymentTargetHPAConf(
+                                min_replicas=1, max_replicas=1
                             ),
-                            uid="",
-                            created_at=datetime(2023, 5, 1),
-                            updated_at=None,
-                            deleted_at=None,
-                            name="",
-                            resource_type=ResourceType.DEPLOYMENT_REVISION,
-                            labels=[],
-                            creator=user,
-                        )
-                    ],
-                    uid="",
-                    created_at=datetime(2023, 5, 1),
-                    updated_at=None,
-                    deleted_at=None,
-                    name="test=xxx",
-                    resource_type=ResourceType.DEPLOYMENT_REVISION,
-                    labels=[],
-                    creator=user,
-                    status=DeploymentRevisionStatus.ACTIVE,
+                            deployment_strategy=DeploymentStrategy.RollingUpdate,
+                        ),
+                    },
                 ),
-                uid="",
-                created_at=datetime(2023, 5, 1),
-                updated_at=None,
-                deleted_at=None,
-                name="test=xxx",
-                resource_type=ResourceType.DEPLOYMENT_REVISION,
-                labels=[],
-                creator=user,
-                status=DeploymentStatus.Running,
-                cluster=ClusterSchema(
-                    uid="",
-                    name="default",
-                    resource_type=ResourceType.CLUSTER,
-                    labels=[],
-                    description="",
-                    creator=user,
-                    created_at=datetime(2023, 5, 1),
-                    updated_at=None,
-                    deleted_at=None,
-                ),
-                kube_namespace="",
             )
-
         else:
-            return DeploymentFullSchemaV2(
-                distributed=False,
-                latest_revision=DeploymentRevisionSchemaV2(
-                    targets=[
-                        DeploymentTargetSchemaV2(
-                            bento=BentoFullSchema(
-                                uid="",
-                                created_at=datetime(2023, 5, 25),
-                                updated_at=None,
-                                deleted_at=None,
-                                name="123",
-                                resource_type=ResourceType.BENTO,
-                                labels=[],
-                                description="",
-                                version="",
-                                image_build_status=BentoImageBuildStatus.PENDING,
-                                upload_status=BentoUploadStatus.SUCCESS,
-                                upload_finished_reason="",
-                                presigned_upload_url="",
-                                presigned_download_url="",
-                                manifest=BentoManifestSchema(
-                                    service="",
-                                    bentoml_version="",
-                                    size_bytes=0,
-                                    apis={},
-                                    models=["iris_clf:ddaex6h2vw6kwcvj"],
-                                ),
-                                build_at=datetime(2023, 5, 25),
-                                repository=BentoRepositorySchema(
-                                    uid="",
-                                    created_at="",
-                                    updated_at=None,
-                                    deleted_at=None,
-                                    name="abc",
-                                    resource_type=ResourceType.BENTO_REPOSITORY,
-                                    labels=[],
-                                    description="",
-                                    latest_bento="",
-                                ),
-                            ),
-                            config=DeploymentTargetConfigV2(
-                                access_type=AccessControl.PUBLIC,
-                                scaling=DeploymentTargetHPAConf(
-                                    min_replicas=3, max_replicas=5
-                                ),
-                                deployment_strategy=DeploymentStrategy.RollingUpdate,
-                                envs=[
-                                    LabelItemSchema(key="env_key", value="env_value")
-                                ],
-                            ),
-                            uid="",
-                            created_at=datetime(2023, 5, 1),
-                            updated_at=None,
-                            deleted_at=None,
-                            name="",
-                            resource_type=ResourceType.DEPLOYMENT_REVISION,
-                            labels=[],
-                            creator=user,
-                        )
-                    ],
-                    uid="",
-                    created_at=datetime(2023, 5, 1),
-                    updated_at=None,
-                    deleted_at=None,
-                    name="test=xxx",
-                    resource_type=ResourceType.DEPLOYMENT_REVISION,
-                    labels=[],
-                    creator=user,
-                    status=DeploymentRevisionStatus.ACTIVE,
-                ),
-                uid="",
-                created_at=datetime(2023, 5, 1),
-                updated_at=None,
-                deleted_at=None,
-                name="test=xxx",
-                resource_type=ResourceType.DEPLOYMENT_REVISION,
-                labels=[],
-                creator=user,
-                status=DeploymentStatus.Running,
-                cluster=ClusterSchema(
-                    uid="",
-                    name="default",
-                    resource_type=ResourceType.CLUSTER,
-                    labels=[],
-                    description="",
-                    creator=user,
-                    created_at=datetime(2023, 5, 1),
-                    updated_at=None,
-                    deleted_at=None,
+            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,
+                    envs=[EnvItemSchema(name="env_key", value="env_value")],
                 ),
-                kube_namespace="",
             )
 
     client = RestApiClient("", "")
     user = UserSchema(name="", email="", first_name="", last_name="")
     client.v2.create_deployment = dummy_create_deployment  # type: ignore
     client.v2.update_deployment = dummy_update_deployment  # type: ignore
+    client.v2.get_deployment = dummy_get_deployment  # type: ignore
     client.v1.get_cluster_list = lambda params: ClusterListSchema(
         start=0,
         count=0,
@@ -282,6 +243,8 @@ def fixture_rest_client() -> RestApiClient:
             ClusterSchema(
                 uid="",
                 name="default",
+                host_cluster_display_name="default_display_name",
+                organization_name="default_organization",
                 resource_type=ResourceType.CLUSTER,
                 labels=[],
                 description="",
@@ -293,8 +256,6 @@ def fixture_rest_client() -> RestApiClient:
         ],
     )  # type: ignore
 
-    client.v2.get_deployment = dummy_get_deployment
-
     return client
 
 
@@ -303,14 +264,12 @@ def test_create_deployment(mock_get_client: MagicMock, rest_client: RestApiClien
     mock_get_client.return_value = rest_client
     deployment = Deployment.create(bento="abc:123")
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=1),
-        bento="abc:123",
-        name="",
-        cluster="default",
-        access_type=AccessControl.PUBLIC,
-        distributed=False,
-    )
+    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=1)
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -324,21 +283,20 @@ def test_create_deployment_custom_standalone(
         scaling_min=2,
         scaling_max=4,
         access_type="private",
-        cluster_name="custom-cluster",
-        envs=[{"key": "env_key", "value": "env_value"}],
+        cluster="custom-cluster",
+        envs=[{"name": "env_key", "value": "env_value"}],
         strategy="RollingUpdate",
     )
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        bento="abc:123",
-        name="custom-name",
-        cluster="custom-cluster",
-        access_type=AccessControl.PRIVATE,
-        scaling=DeploymentTargetHPAConf(min_replicas=2, max_replicas=4),
-        distributed=False,
-        deployment_strategy=DeploymentStrategy.RollingUpdate,
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-    )
+    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.envs == [EnvItemSchema(name="env_key", value="env_value")]
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -348,14 +306,12 @@ def test_create_deployment_scailing_only_min(
     mock_get_client.return_value = rest_client
     deployment = Deployment.create(bento="abc:123", scaling_min=3)
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        bento="abc:123",
-        name="",
-        cluster="default",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=3, max_replicas=3),
-        distributed=False,
-    )
+    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=3, max_replicas=3)
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -365,14 +321,12 @@ def test_create_deployment_scailing_only_max(
     mock_get_client.return_value = rest_client
     deployment = Deployment.create(bento="abc:123", scaling_max=3)
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        bento="abc:123",
-        name="",
-        cluster="default",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=3),
-        distributed=False,
-    )
+    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")
@@ -382,27 +336,25 @@ def test_create_deployment_scailing_mismatch_min_max(
     mock_get_client.return_value = rest_client
     deployment = Deployment.create(bento="abc:123", scaling_min=3, scaling_max=2)
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        bento="abc:123",
-        name="",
-        cluster="default",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=2, max_replicas=2),
-        distributed=False,
-    )
+    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=2, max_replicas=2)
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
-def test_create_deployment_config_dct(
+def test_create_deployment_config_dict(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    config_dct = {
+    config_dict = {
         "services": {
             "irisclassifier": {"scaling": {"max_replicas": 2, "min_replicas": 1}},
             "preprocessing": {"scaling": {"max_replicas": 2}},
         },
-        "envs": [{"key": "env_key", "value": "env_value"}],
+        "envs": [{"name": "env_key", "value": "env_value"}],
         "bentoml_config_overrides": {
             "irisclassifier": {
                 "resources": {
@@ -412,32 +364,28 @@ def test_create_deployment_config_dct(
             }
         },
     }
-    deployment = Deployment.create(bento="abc:123", config_dct=config_dct)
+    deployment = Deployment.create(bento="abc:123", config_dict=config_dict)
     # assert expected schema
-    assert deployment._schema == CreateDeploymentSchemaV2(
-        bento="abc:123",
-        name="",
-        cluster="default",
-        access_type=AccessControl.PUBLIC,
-        distributed=True,
-        services={
-            "irisclassifier": DeploymentServiceConfig(
-                scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2)
-            ),
-            "preprocessing": DeploymentServiceConfig(
-                scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2)
-            ),
-        },
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-        bentoml_config_overrides={
-            "irisclassifier": {
-                "resources": {
-                    "cpu": "300m",
-                    "memory": "500m",
-                },
-            }
-        },
-    )
+    assert deployment.cluster == "default_display_name"
+    assert deployment.name == ""
+    assert deployment.distributed
+    config = deployment.get_config(refetch=False)
+    assert config.services == {
+        "irisclassifier": DeploymentServiceConfig(
+            scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=2)
+        ),
+        "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")
@@ -447,17 +395,19 @@ def test_update_deployment(mock_get_client: MagicMock, rest_client: RestApiClien
         name="test",
         bento="abc:1234",
         access_type="private",
-        envs=[{"key": "env_key2", "value": "env_value2"}],
+        envs=[{"name": "env_key2", "value": "env_value2"}],
         strategy="Recreate",
     )
     # assert expected schema
-    assert deployment._schema == DummyUpdateSchema(
-        bento="abc:1234",
-        access_type=AccessControl.PRIVATE,
-        scaling=DeploymentTargetHPAConf(min_replicas=3, max_replicas=5),
-        deployment_strategy=DeploymentStrategy.Recreate,
-        envs=[LabelItemSchema(key="env_key2", value="env_value2")],
-    )
+    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.envs == [EnvItemSchema(name="env_key2", value="env_value2")]
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -467,13 +417,14 @@ def test_update_deployment_scaling_only_min(
     mock_get_client.return_value = rest_client
     deployment = Deployment.update(name="test", scaling_min=1)
     # assert expected schema
-    assert deployment._schema == DummyUpdateSchema(
-        bento="abc:123",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=5),
-        deployment_strategy=DeploymentStrategy.RollingUpdate,
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-    )
+    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.envs == [EnvItemSchema(name="env_key", value="env_value")]
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -483,13 +434,14 @@ def test_update_deployment_scaling_only_max(
     mock_get_client.return_value = rest_client
     deployment = Deployment.update(name="test", scaling_max=3)
     # assert expected schema
-    assert deployment._schema == DummyUpdateSchema(
-        bento="abc:123",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=3, max_replicas=3),
-        deployment_strategy=DeploymentStrategy.RollingUpdate,
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-    )
+    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.envs == [EnvItemSchema(name="env_key", value="env_value")]
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -499,13 +451,14 @@ def test_update_deployment_scaling_too_big_min(
     mock_get_client.return_value = rest_client
     deployment = Deployment.update(name="test", scaling_min=10)
     # assert expected schema
-    assert deployment._schema == DummyUpdateSchema(
-        bento="abc:123",
-        access_type=AccessControl.PUBLIC,
-        scaling=DeploymentTargetHPAConf(min_replicas=5, max_replicas=5),
-        deployment_strategy=DeploymentStrategy.RollingUpdate,
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-    )
+    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")]
 
 
 @patch("bentoml._internal.cloud.deployment.get_rest_api_client")
@@ -513,28 +466,28 @@ def test_update_deployment_distributed(
     mock_get_client: MagicMock, rest_client: RestApiClient
 ):
     mock_get_client.return_value = rest_client
-    config_dct = {
+    config_dict = {
         "services": {
             "irisclassifier": {"scaling": {"max_replicas": 50}},
             "preprocessing": {"instance_type": "t3-large"},
         }
     }
-    deployment = Deployment.update(name="test-distributed", config_dct=config_dct)
+    deployment = Deployment.update(name="test-distributed", config_dict=config_dict)
     # assert expected schema
-    assert deployment._schema == DummyUpdateSchema(
-        bento="abc:123",
-        access_type=AccessControl.PUBLIC,
-        envs=[LabelItemSchema(key="env_key", value="env_value")],
-        services={
-            "irisclassifier": DeploymentServiceConfig(
-                instance_type="t3-small",
-                scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=50),
-                deployment_strategy=DeploymentStrategy.RollingUpdate,
-            ),
-            "preprocessing": DeploymentServiceConfig(
-                instance_type="t3-large",
-                scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=1),
-                deployment_strategy=DeploymentStrategy.RollingUpdate,
-            ),
-        },
-    )
+    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.envs == [EnvItemSchema(name="env_key", value="env_value")]
+    assert config.services == {
+        "irisclassifier": DeploymentServiceConfig(
+            instance_type="t3-small",
+            scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=50),
+            deployment_strategy=DeploymentStrategy.RollingUpdate,
+        ),
+        "preprocessing": DeploymentServiceConfig(
+            instance_type="t3-large",
+            scaling=DeploymentTargetHPAConf(min_replicas=1, max_replicas=1),
+            deployment_strategy=DeploymentStrategy.RollingUpdate,
+        ),
+    }