Source code for bentoml._internal.service.loader

from __future__ import annotations

import importlib
import logging
import os
import sys
import typing as t
from typing import TYPE_CHECKING

import fs
from simple_di import Provide
from simple_di import inject

from ...exceptions import BentoMLException
from ...exceptions import ImportServiceError
from ...exceptions import NotFound
from ..bento import Bento
from ..bento.bento import BENTO_PROJECT_DIR_NAME
from ..bento.bento import BENTO_YAML_FILENAME
from ..bento.bento import DEFAULT_BENTO_BUILD_FILE
from ..bento.build_config import BentoBuildConfig
from ..configuration import BENTOML_VERSION
from ..configuration.containers import BentoMLContainer
from ..models import ModelStore
from ..tag import Tag
from .service import on_load_bento

    from ..bento import BentoStore
    from .service import Service

logger = logging.getLogger(__name__)

def import_service(
    svc_import_path: str,
    working_dir: t.Optional[str] = None,
    standalone_load: bool = False,
    model_store: ModelStore = Provide[BentoMLContainer.model_store],
) -> Service:
    """Import a Service instance from source code, by providing the svc_import_path
    which represents the module where the Service instance is created and optionally
    what attribute can be used to access this Service instance in that module

    Example usage:
        # When multiple service defined in the same module

        # Find svc by Python module name or file path

        # When there's only one Service instance in the target module, the attributes
        # part in the svc_import_path can be omitted
    from bentoml import Service

    prev_cwd = None
    sys_path_modified = False
    prev_cwd = os.getcwd()
    global_model_store = BentoMLContainer.model_store.get()

    def recover_standalone_env_change():
        # Reset to previous cwd

        if working_dir is not None:
            working_dir = os.path.realpath(os.path.expanduser(working_dir))
            # Set cwd(current working directory) to the Bento's project directory,
            # which allows user code to read files using relative path
            working_dir = os.getcwd()

        if working_dir not in sys.path:
            sys.path.insert(0, working_dir)
            sys_path_modified = True

        if model_store is not global_model_store:

            'Importing service "%s" from working dir: "%s"',

        import_path, _, attrs_str = svc_import_path.partition(":")
        if not import_path:
            raise ImportServiceError(
                f'Invalid import target "{svc_import_path}", must format as '
                '"<module>:<attribute>" or "<module>'

        if os.path.isfile(import_path):
            import_path = os.path.realpath(import_path)
            # Importing from a module file path:
            if not import_path.startswith(working_dir):
                raise ImportServiceError(
                    f'Module "{import_path}" not found in working directory "{working_dir}"'

            file_name, ext = os.path.splitext(import_path)
            if ext != ".py":
                raise ImportServiceError(
                    f'Invalid module extension "{ext}" in target "{svc_import_path}",'
                    ' the only extension acceptable here is ".py"'

            # move up until no longer in a python package or in the working dir
            module_name_parts: t.List[str] = []
            path = file_name
            while True:
                path, name = os.path.split(path)
                if (
                    not os.path.exists(os.path.join(path, ""))
                    or path == working_dir
            module_name = ".".join(module_name_parts[::-1])
            # Importing by module name:
            module_name = import_path

        # Import the service using the Bento's own model store
            module = importlib.import_module(module_name, package=working_dir)
        except ImportError as e:
            raise ImportServiceError(f'Failed to import module "{module_name}": {e}')
        if not standalone_load:

        if attrs_str:
            instance = module
                for attr_str in attrs_str.split("."):
                    instance = getattr(instance, attr_str)
            except AttributeError:
                raise ImportServiceError(
                    f'Attribute "{attrs_str}" not found in module "{module_name}".'
            instances = [
                (k, v) for k, v in module.__dict__.items() if isinstance(v, Service)

            if len(instances) == 1:
                attrs_str = instances[0][0]
                instance = instances[0][1]
                raise ImportServiceError(
                    f'Multiple Service instances found in module "{module_name}", use'
                    '"<module>:<svc_variable_name>" to specify the service instance or'
                    "define only service instance per python module/file"

        assert isinstance(
            instance, Service
        ), f'import target "{module_name}:{attrs_str}" is not a bentoml.Service instance'

        # set import_str for retrieving the service import origin
        object.__setattr__(instance, "_import_str", f"{module_name}:{attrs_str}")
        return instance
    except ImportServiceError:
        if sys_path_modified and working_dir:
            # Undo changes to sys.path


def load_bento(
    bento: str | Tag | Bento,
    bento_store: "BentoStore" = Provide[BentoMLContainer.bento_store],
    standalone_load: bool = False,
) -> "Service":
    """Load a Service instance from a bento found in local bento store:

    Example usage:
    if isinstance(bento, (str, Tag)):
        bento = bento_store.get(bento)

        'Loading bento "%s" found in local store: %s',

    # not in validate as it's only really necessary when getting bentos from disk
        info_bentoml_version =
        if tuple(info_bentoml_version.split(".")) > tuple(BENTOML_VERSION.split(".")):
                "%s was built with newer version of BentoML, which does not match with current running BentoML version %s",
                "%s was built with BentoML version %s, which does not match the current BentoML version %s",
    return _load_bento(bento, standalone_load)

def load_bento_dir(path: str, standalone_load: bool = False) -> "Service":
    """Load a Service instance from a bento directory

    Example usage:
    bento_fs = fs.open_fs(path)
    bento = Bento.from_fs(bento_fs)
        'Loading bento "%s" from directory: %s',
    return _load_bento(bento, standalone_load)

def _load_bento(bento: Bento, standalone_load: bool) -> "Service":
    # Use Bento's user project path as working directory when importing the service
    working_dir = bento._fs.getsyspath(BENTO_PROJECT_DIR_NAME)

    model_store = BentoMLContainer.model_store.get()
    # read from bento's local model store if it exists and is not empty
    # This is the case when running in a container
    local_model_store = bento._model_store
    if local_model_store is not None and len(local_model_store.list()) > 0:
        model_store = local_model_store

    # Read the model aliases
    resolved_model_aliases = {m.alias: str(m.tag) for m in if m.alias}

    svc = import_service(,
    on_load_bento(svc, bento)
    return svc

[docs]def load( bento_identifier: str | Tag | Bento, working_dir: t.Optional[str] = None, standalone_load: bool = False, ) -> "Service": """Load a Service instance by the bento_identifier Args: bento_identifier: target Service to import or Bento to load working_dir: when importing from service, set the working_dir standalone_load: treat target Service as standalone. This will change global current working directory and global model store. Returns: The loaded :obj:`bentoml.Service` instance. The argument ``bento_identifier`` can be one of the following forms: * Tag pointing to a Bento in local Bento store under `BENTOML_HOME/bentos` * File path to a Bento directory * "import_str" for loading a service instance from the `working_dir` Example load from Bento usage: .. code-block:: python # load from local bento store load("FraudDetector:latest") load("FraudDetector:4tht2icroji6zput") # load from bento directory load("~/bentoml/bentos/iris_classifier/4tht2icroji6zput") Example load from working directory by "import_str" usage: .. code-block:: python # When multiple service defined in the same module load("fraud_detector:svc_a") load("fraud_detector:svc_b") # Find svc by Python module name or file path load("fraud_detector:svc") load("") load("") load("./def/abc/") # When there's only one Service instance in the target module, the attributes # part in the svc_import_path can be omitted load("") load("fraud_detector") Limitations when `standalone_load=False`: * Models used in the Service being imported, if not accessed during module import, must be presented in the global model store * Files required for the Service to run, if not accessed during module import, must be presented in the current working directory """ if isinstance(bento_identifier, (Bento, Tag)): # Load from local BentoStore return load_bento(bento_identifier) if os.path.isdir(os.path.expanduser(bento_identifier)): bento_path = os.path.abspath(os.path.expanduser(bento_identifier)) if os.path.isfile( os.path.expanduser(os.path.join(bento_path, BENTO_YAML_FILENAME)) ): # Loading from path to a built Bento try: svc = load_bento_dir(bento_path, standalone_load=standalone_load) except ImportServiceError as e: raise BentoMLException( f"Failed loading Bento from directory {bento_path}: {e}" )"Service loaded from Bento directory: %s", svc) elif os.path.isfile( os.path.expanduser(os.path.join(bento_path, DEFAULT_BENTO_BUILD_FILE)) ): # Loading from path to a project directory containing bentofile.yaml try: with open( os.path.join(bento_path, DEFAULT_BENTO_BUILD_FILE), "r", encoding="utf-8", ) as f: build_config = BentoBuildConfig.from_yaml(f) assert ( build_config.service ), '"service" field in "bentofile.yaml" is required for loading the service, e.g. "service:"' BentoMLContainer.model_aliases.set(build_config.model_aliases) svc = import_service( build_config.service, working_dir=bento_path, standalone_load=standalone_load, ) except ImportServiceError as e: raise BentoMLException( f"Failed loading Bento from directory {bento_path}: {e}" ) logger.debug("'%s' loaded from '%s': %s",, bento_path, svc) else: raise BentoMLException( f"Failed loading service from path {bento_path}. When loading from a path, it must be either a Bento containing bento.yaml or a project directory containing bentofile.yaml" ) else: try: # Loading from service definition file, e.g. "" svc = import_service( bento_identifier, working_dir=working_dir, standalone_load=standalone_load, ) logger.debug("'%s' imported from source: %s",, svc) except ImportServiceError as e1: try: # Loading from local bento store by tag, e.g. "iris_classifier:latest" svc = load_bento(bento_identifier, standalone_load=standalone_load) logger.debug("'%s' loaded from Bento store: %s",, svc) except (NotFound, ImportServiceError) as e2: raise BentoMLException( f"Failed to load bento or import service '{bento_identifier}'.\n" f"If you are attempting to import bento in local store: '{e1}'.\n" f"If you are importing by python module path: '{e2}'." ) return svc