feat: Adding SDK support for deployment labels

src/bentoml/_internal/cloud/deployment.py

@@ -1178,7 +1178,11 @@ def _generate_deployment_info_(
         )
 
     def list(
-        self, cluster: str | None = None, search: str | None = None, dev: bool = False
+        self,
+        cluster: str | None = None,
+        search: str | None = None,
+        dev: bool = False,
+        q: str | None = None,
     ) -> list[Deployment]:
         """
         List all deployments in the cluster.
@@ -1195,7 +1199,7 @@ def list(
             res_count = cloud_rest_client.v2.list_deployment(all=True, search=search)
             if res_count.total == 0:
                 return []
-            q = "is:dev" if dev else None
+            q = "is:dev" if dev else q
             res = cloud_rest_client.v2.list_deployment(
                 search=search, count=res_count.total, all=True, q=q
             )

src/bentoml/_internal/cloud/schemas/schemasv2.py

@@ -12,6 +12,7 @@
 from bentoml._internal.cloud.schemas.schemasv1 import UserSchema
 
 from .modelschemas import EnvItemSchema
+from .modelschemas import LabelItemSchema
 
 
 @attr.define
@@ -46,6 +47,7 @@ class DeploymentConfigSchema:
     __forbid_extra_keys__ = False
     access_authorization: bool = attr.field(default=False)
     envs: t.Optional[t.List[EnvItemSchema]] = attr.field(default=None)
+    labels: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
     secrets: t.Optional[t.List[str]] = attr.field(default=None)
     services: t.Dict[str, DeploymentServiceConfig] = attr.field(factory=dict)
 

src/bentoml/deployment.py

@@ -13,6 +13,7 @@
 from ._internal.cloud.deployment import Deployment
 from ._internal.cloud.deployment import DeploymentConfigParameters
 from ._internal.cloud.schemas.modelschemas import EnvItemSchema
+from ._internal.cloud.schemas.modelschemas import LabelItemSchema
 from ._internal.configuration.containers import BentoMLContainer
 from ._internal.tag import Tag
 from .exceptions import BentoMLException
@@ -288,9 +289,26 @@ def list(
     cluster: str | None = None,
     search: str | None = None,
     dev: bool = False,
+    q: str | None = None,
+    labels: t.List[LabelItemSchema] | t.List[dict[str, t.Any]] | None = None,
     _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
 ) -> t.List[Deployment]:
-    return _cloud_client.deployment.list(cluster=cluster, search=search, dev=dev)
+    # Syntatic sugar to enable searching by `labels` argument
+    if labels is not None:
+        label_query = " ".join(
+            f"label:{d['key']}={d['value']}"
+            for d in (
+                label if isinstance(label, dict) else attr.asdict(label)
+                for label in labels
+            )
+        )
+
+        if q is not None:
+            q = f"{q} {label_query}"
+        else:
+            q = label_query
+
+    return _cloud_client.deployment.list(cluster=cluster, search=search, dev=dev, q=q)
 
 
 __all__ = ["create", "get", "update", "apply", "terminate", "delete", "list"]

src/bentoml_cli/deployment.py

@@ -744,14 +744,29 @@ def delete(  # type: ignore
     type=click.Choice(["json", "yaml", "table"]),
     default="table",
 )
+@click.option(
+    "--label",
+    "labels",
+    type=click.STRING,
+    multiple=True,
+    default=None,
+    help="Filter deployments by label(s).",
+    metavar="KEY=VALUE",
+)
 def list_command(  # type: ignore
     cluster: str | None,
     search: str | None,
+    labels: tuple[str, ...] | None,
     output: t.Literal["json", "yaml", "table"],
 ) -> None:
     """List existing deployments on BentoCloud."""
+    if labels is not None:
+        # For labels like ["env=prod", "team=infra"]
+        # This will output: "label:env=prod label:team=infra"
+        labels_query = " ".join(f"label:{label}" for label in labels)
+
     try:
-        d_list = bentoml.deployment.list(cluster=cluster, search=search)
+        d_list = bentoml.deployment.list(cluster=cluster, search=search, q=labels_query)
     except BentoMLException as e:
         raise_deployment_config_error(e, "list")
     res: list[dict[str, t.Any]] = [d.to_dict() for d in d_list]