fix mypy format lint

src/zenml/integrations/bentoml/model_deployers/bentoml_model_deployer.py

@@ -34,6 +34,8 @@
 )
 from zenml.logger import get_logger
 from zenml.model_deployers import BaseModelDeployer, BaseModelDeployerFlavor
+from zenml.services.container.container_service import ContainerServiceStatus
+from zenml.services.local.local_service import LocalDaemonServiceStatus
 from zenml.services.service import BaseService, ServiceConfig
 from zenml.utils.io_utils import create_dir_recursive_if_not_exists
 
@@ -104,7 +106,7 @@ def local_path(self) -> str:
         return self._service_path
 
     @staticmethod
-    def get_model_server_info(  # type: ignore[override]
+    def get_model_server_info(
         service_instance: BaseService,
     ) -> Dict[str, Optional[str]]:
         """Return implementation specific information on the model server.
@@ -135,25 +137,36 @@ def get_model_server_info(  # type: ignore[override]
             )
 
         predictions_apis_urls = ""
-        if service_instance.prediction_apis_urls is not None:
+        if service_instance.prediction_apis_urls is not None:  # type: ignore
             predictions_apis_urls = ", ".join(
                 [
                     api
-                    for api in service_instance.prediction_apis_urls
+                    for api in service_instance.prediction_apis_urls  # type: ignore
                     if api is not None
                 ]
             )
 
+        service_config = service_instance.config
+        assert isinstance(
+            service_config,
+            (BentoMLLocalDeploymentConfig, BentoMLContainerDeploymentConfig),
+        )
+
+        service_status = service_instance.status
+        assert isinstance(
+            service_status, (ContainerServiceStatus, LocalDaemonServiceStatus)
+        )
+
         return {
             "HEALTH_CHECK_URL": service_instance.get_healthcheck_url(),
             "PREDICTION_URL": service_instance.get_prediction_url(),
-            "BENTO_TAG": service_instance.config.bento_tag,
-            "MODEL_NAME": service_instance.config.model_name,
-            "MODEL_URI": service_instance.config.model_uri,
-            "BENTO_URI": service_instance.config.bento_uri,
-            "SERVICE_PATH": service_instance.status.runtime_path,
-            "DAEMON_PID": str(service_instance.status.pid)
-            if hasattr(service_instance.status, "pid")
+            "BENTO_TAG": service_config.bento_tag,
+            "MODEL_NAME": service_config.model_name,
+            "MODEL_URI": service_config.model_uri,
+            "BENTO_URI": service_config.bento_uri,
+            "SERVICE_PATH": service_status.runtime_path,
+            "DAEMON_PID": str(service_status.pid)
+            if hasattr(service_status, "pid")
             else None,
             "PREDICTION_APIS_URLS": predictions_apis_urls,
         }
@@ -219,6 +232,15 @@ def _clean_up_existing_service(
         # stop the older service
         existing_service.stop(timeout=timeout, force=force)
 
+        # assert that the service is either a BentoMLLocalDeploymentService or a BentoMLContainerDeploymentService
+        if not isinstance(
+            existing_service,
+            (BentoMLLocalDeploymentService, BentoMLContainerDeploymentService),
+        ):
+            raise ValueError(
+                f"Unsupported service type: {type(existing_service)}"
+            )
+
         # delete the old configuration file
         if existing_service.status.runtime_path:
             shutil.rmtree(existing_service.status.runtime_path)
@@ -244,12 +266,17 @@ def _create_new_service(
         Raises:
             ValueError: If the service type is not supported.
         """
+        assert isinstance(
+            config,
+            (BentoMLLocalDeploymentConfig, BentoMLContainerDeploymentConfig),
+        )
         # set the root runtime path with the stack component's UUID
         config.root_runtime_path = self.local_path
         # create a new service for the new model
         # if the config is of type BentoMLLocalDeploymentConfig, create a
         # BentoMLLocalDeploymentService, otherwise create a
         # BentoMLContainerDeploymentService
+        service: BaseService
         if isinstance(config, BentoMLLocalDeploymentConfig):
             service = BentoMLLocalDeploymentService(uuid=id, config=config)
         elif isinstance(config, BentoMLContainerDeploymentConfig):

src/zenml/integrations/bentoml/services/bentoml_container_deployment.py

@@ -57,7 +57,7 @@ class BentoMLContainerDeploymentConfig(ContainerServiceConfig):
     bento_tag: str
     bento_uri: Optional[str] = None
     platform: Optional[str] = None
-    image: Optional[str] = None
+    image: str = ""
     image_tag: Optional[str] = None
     features: Optional[List[str]] = None
     file: Optional[str] = None
@@ -243,7 +243,7 @@ def _start_container(self) -> None:
 
     def _containerize_and_push_bento(self) -> None:
         """Containerize the bento and push it to the container registry.
-        
+
         Raises:
             Exception: If the bento containerization fails.
         """
@@ -259,11 +259,11 @@ def _containerize_and_push_bento(self) -> None:
                 image_name = (
                     f"{container_registry.config.uri}/{self.config.bento_tag}"
                 )
-                image_tag = (image_name,)
+                image_tag = (image_name,)  # type: ignore
                 self.config.image = image_name
             else:
                 # bentoml will use the bento tag as the name of the image
-                image_tag = (self.config.bento_tag,)
+                image_tag = (self.config.bento_tag,)  # type: ignore
                 self.config.image = self.config.bento_tag
         try:
             bentoml.container.build(
@@ -301,7 +301,7 @@ def provision(self) -> None:
 
     def run(self) -> None:
         """Start the service.
-        
+
         Raises:
             FileNotFoundError: If the bento file is not found.
             subprocess.CalledProcessError: If the bentoml serve command fails.

src/zenml/integrations/bentoml/services/bentoml_local_deployment.py

@@ -234,7 +234,7 @@ def run(self) -> None:
                 logger.info("Stopping BentoML prediction service...")
         else:
             # bentoml>=1.2
-            from _bentoml_impl.server import serve_http
+            from _bentoml_impl.server import serve_http  # type: ignore
 
             svc.inject_config()
             try:

src/zenml/integrations/bentoml/steps/bentoml_deployer.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Implementation of the BentoML model deployer pipeline step."""
 
-from typing import List, Literal, Optional, cast
+from typing import List, Literal, Optional, Tuple, cast
 
 import bentoml
 from bentoml._internal.bento import bento
@@ -32,7 +32,8 @@
     SSLBentoMLParametersConfig,
 )
 from zenml.logger import get_logger
-from zenml.services.service import BaseService
+from zenml.services.service import BaseService, ServiceConfig
+from zenml.services.service_type import ServiceType
 from zenml.utils import source_utils
 
 logger = get_logger(__name__)
@@ -115,7 +116,9 @@ def service_apis(bento_tag: str) -> List[str]:
         apis_paths = list(apis.keys())
         return apis_paths
 
-    def create_deployment_config(deployment_type):
+    def create_deployment_config(
+        deployment_type: Literal["local", "container"],
+    ) -> Tuple[ServiceConfig, ServiceType]:
         common_config = {
             "model_name": model_name,
             "bento_tag": str(bento.tag),
@@ -161,6 +164,7 @@ def create_deployment_config(deployment_type):
     )
 
     # Creating a new service with inactive state and status by default
+    service: Optional[BaseService] = None
     if existing_services:
         if deployment_type == "container":
             service = cast(
@@ -176,11 +180,13 @@ def create_deployment_config(deployment_type):
             f"'{step_name}' and pipeline '{pipeline_name}' for model "
             f"'{model_name}'..."
         )
+        assert service is not None
         if not service.is_running:
             service.start(timeout=timeout)
         return service
 
     # create a new model deployment and replace an old one if it exists
+    new_service: BaseService
     if deployment_type == "container":
         new_service = cast(
             BentoMLContainerDeploymentService,