"""
User facing python APIs for building a OCI-complicant image.
"""
from __future__ import annotations
import logging
import os
import shutil
import sys
import typing as t
from simple_di import Provide
from simple_di import inject
from ._internal.configuration.containers import BentoMLContainer
from ._internal.container import build as _internal_build
from ._internal.container import (
construct_containerfile as _internal_construct_containerfile,
)
from ._internal.container import get_backend
from ._internal.container import health
from ._internal.container import register_backend
from .exceptions import BentoMLException
if t.TYPE_CHECKING:
from ._internal.bento import BentoStore
from ._internal.container import DefaultBuilder
from ._internal.container.base import ArgType
from ._internal.tag import Tag
from ._internal.types import PathType
logger = logging.getLogger(__name__)
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["docker"] = ...,
image_tag: tuple[str, ...] | None = ...,
features: t.Sequence[str] | None = ...,
*,
file: PathType | None = ...,
context_path: PathType | None = ...,
add_host: dict[str, str] | ArgType = ...,
build_arg: dict[str, str] | ArgType = ...,
cache_from: str | dict[str, str] | ArgType = ...,
disable_content_trust: t.Literal[True, False] = ...,
iidfile: PathType | None = ...,
isolation: t.Literal["default", "process", "hyperv"] | None = ...,
label: dict[str, str] | ArgType = ...,
network: str | None = ...,
no_cache: t.Literal[True, False] = ...,
output: str | dict[str, str] | ArgType = ...,
platform: str | ArgType = ...,
progress: t.Literal["auto", "tty", "plain"] = ...,
pull: t.Literal[True, False] = ...,
quiet: t.Literal[True, False] = ...,
secret: str | dict[str, str] | ArgType = ...,
ssh: str | ArgType = ...,
target: str | ArgType = ...,
**kwargs: t.Any,
) -> t.Any: ...
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["buildctl"] = ...,
image_tag: tuple[str] | None = ...,
features: t.Sequence[str] | None = ...,
*,
file: PathType | None = ...,
context_path: PathType | None = ...,
output: str | dict[str, str] | ArgType = ...,
progress: t.Literal["auto", "tty", "plain"] | ArgType = ...,
trace: PathType | None = ...,
local: dict[str, str] | ArgType = ...,
frontend: str | None = ...,
no_cache: t.Literal[True, False] = ...,
export_cache: str | dict[str, str] | ArgType = ...,
import_cache: str | dict[str, str] | ArgType = ...,
secret: str | dict[str, str] | ArgType = ...,
allow: str | ArgType = ...,
ssh: str | ArgType = ...,
metadata_file: PathType | None = ...,
opt: tuple[str, ...] | dict[str, str | tuple[str, ...]] | None = ...,
**kwargs: t.Any,
) -> t.Any: ...
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["buildx"] = ...,
image_tag: tuple[str] | None = ...,
features: t.Sequence[str] | None = ...,
*,
file: PathType | None = ...,
tag: tuple[str] | None = ...,
context_path: PathType = ...,
add_host: dict[str, str] | ArgType = ...,
attest: str | dict[str, str] | ArgType = ...,
allow: str | ArgType = ...,
build_arg: dict[str, str] | ArgType = ...,
build_context: dict[str, str] | ArgType = ...,
builder: str | None = ...,
cache_from: str | dict[str, str] | ArgType = ...,
cache_to: str | dict[str, str] | ArgType = ...,
cgroup_parent: PathType | None = ...,
iidfile: PathType | None = ...,
label: dict[str, str] | ArgType = ...,
load: bool = True,
metadata_file: PathType | None = ...,
network: str | None = ...,
no_cache: t.Literal[True, False] = ...,
no_cache_filter: str | dict[str, str] | ArgType = ...,
output: str | dict[str, str] | ArgType = ...,
platform: str | ArgType = ...,
progress: t.Literal["auto", "tty", "plain"] = "auto",
provenance: str | dict[str, str] | ArgType = ...,
pull: t.Literal[True, False] = ...,
push: t.Literal[True, False] = ...,
quiet: t.Literal[True, False] = ...,
sbom: str | dict[str, str] | ArgType = ...,
secret: str | dict[str, str] | ArgType = ...,
shm_size: int | None = ...,
ssh: str | ArgType = ...,
target: str | None = ...,
ulimit: str | dict[str, tuple[int, int]] | ArgType = ...,
**kwargs: t.Any,
) -> t.Any: ...
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["nerdctl"] = ...,
image_tag: tuple[str] | None = ...,
features: t.Sequence[str] | None = ...,
*,
file: PathType | None = ...,
tag: tuple[str] | None = ...,
context_path: PathType = ".",
build_arg: dict[str, str] | ArgType = ...,
buildkit_host: str | None = ...,
cache_from: str | dict[str, str] | ArgType = ...,
cache_to: str | dict[str, str] | ArgType = ...,
iidfile: PathType | None = ...,
ipfs: t.Literal[True, False] = ...,
label: dict[str, str] | ArgType = ...,
no_cache: t.Literal[True, False] = ...,
output: str | dict[str, str] | ArgType = ...,
platform: str | ArgType = ...,
progress: t.Literal["auto", "tty", "plain"] | ArgType = ...,
quiet: t.Literal[True, False] = ...,
rm: t.Literal[True, False] = ...,
secret: str | dict[str, str] | ArgType = ...,
ssh: str | ArgType = ...,
target: str | None = ...,
# global flags
address: str | None = ...,
host: str | None = ...,
cgroup_manager: str | None = ...,
cni_netconfpath: PathType | None = ...,
cni_path: PathType | None = ...,
data_root: PathType | None = ...,
debug: t.Literal[True, False] = ...,
debug_full: t.Literal[True, False] = ...,
hosts_dir: str | ArgType = ...,
insecure_registry: t.Literal[True, False] = ...,
namespace: str | None = ...,
snapshotter: str | None = ...,
storage_driver: str | None = ...,
**kwargs: t.Any,
) -> t.Any: ...
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["podman"] = ...,
image_tag: tuple[str] | None = ...,
features: t.Sequence[str] | None = ...,
*,
file: PathType | None = ...,
context_path: PathType | None = ...,
add_host: dict[str, str] | ArgType = ...,
all_platforms: t.Literal[True, False] = ...,
annotation: dict[str, str] | ArgType = ...,
label: dict[str, str] | ArgType = ...,
arch: str | None = ...,
authfile: PathType | None = ...,
build_arg: dict[str, str] | ArgType = ...,
build_context: dict[str, str] | ArgType = ...,
cache_from: str | None = ...,
cache_to: str | None = ...,
cache_ttl: str | None = ...,
cap_add: str | ArgType = ...,
cap_drop: str | ArgType = ...,
cert_dir: PathType | None = ...,
cgroup_parent: PathType | None = ...,
cgroupns: str | None = ...,
cpp_flag: ArgType = ...,
cpu_period: int | None = ...,
cpu_quota: int | None = ...,
cpu_shares: int | None = ...,
cpuset_cpus: str | None = ...,
cpuset_mems: str | None = ...,
creds: str | dict[str, str] | ArgType = ...,
decryption_key: str | dict[str, str] | ArgType = ...,
device: str | ArgType = ...,
disable_compression: t.Literal[True, False] = ...,
dns: str | None = ...,
dns_option: str | ArgType = ...,
dns_search: str | ArgType = ...,
env: str | dict[str, str] | ArgType = ...,
force_rm: t.Literal[True, False] = ...,
format: str | t.Literal["docker", "oci"] | None = ...,
hooks_dir: str | ArgType = ...,
http_proxy: t.Literal[True, False] = ...,
identity_label: t.Literal[True, False] = ...,
ignorefile: PathType | None = ...,
iidfile: PathType | None = ...,
ipc: str | PathType | None = ...,
isolation: str | None = ...,
jobs: int | None = ...,
layers: t.Literal[True, False] = ...,
logfile: PathType | None = ...,
manifest: str | None = ...,
memory: str | None = ...,
memory_swap: str | None = ...,
network: str | None = ...,
no_cache: t.Literal[True, False] = ...,
no_hosts: t.Literal[True, False] = ...,
omit_history: t.Literal[True, False] = ...,
os: str | None = ...,
os_feature: str | None = ...,
os_version: str | None = ...,
pid: PathType | None = ...,
platform: str | ArgType = ...,
output: str | dict[str, str] | ArgType = ...,
pull: t.Literal[
True, False, "always", "true", "missing", "never", "false", "newer"
] = False,
quiet: t.Literal[True, False] = ...,
rm: t.Literal[True, False] = ...,
retry: int | None = ...,
retry_delay: int | None = ...,
runtime: PathType | None = ...,
runtime_flag: str | dict[str, str] | ArgType = ...,
secret: str | dict[str, str] | ArgType = ...,
security_opt: str | ArgType = ...,
shm_size: str | None = ...,
sign_by: str | None = ...,
skip_unused_stages: t.Literal[True, False] = ...,
squash: t.Literal[True, False] = ...,
squash_all: t.Literal[True, False] = ...,
ssh: str | ArgType = ...,
stdin: t.Literal[True, False] = ...,
target: str | None = ...,
timestamp: int | None = ...,
tls_verify: t.Literal[True, False] = ...,
ulimit: str | dict[str, tuple[int, int]] | ArgType = ...,
unsetenv: str | ArgType = ...,
userns: str | None = ...,
userns_gid_map: str | tuple[str, str, str] | None = ...,
userns_gid_map_group: str | None = ...,
userns_uid_map: str | tuple[str, str, str] | None = ...,
userns_uid_map_user: str | None = ...,
uts: str | None = ...,
variant: str | None = ...,
volume: str | tuple[str, str, str] | None = ...,
**kwargs: t.Any,
) -> t.Any: ...
@t.overload
def build(
bento_tag: Tag | str,
backend: t.Literal["buildah"] = ...,
image_tag: tuple[str] | None = ...,
features: t.Sequence[str] | None = ...,
*,
context_path: PathType = ...,
file: PathType | None = ...,
tag: tuple[str] | None = ...,
add_host: dict[str, str] | ArgType = ...,
annotation: dict[str, str] | ArgType = ...,
label: dict[str, str] | ArgType = ...,
arch: str | None = ...,
authfile: PathType | None = ...,
build_arg: dict[str, str] | ArgType = ...,
cache_from: str | None = ...,
cap_add: str | ArgType = ...,
cap_drop: str | ArgType = ...,
cert_dir: PathType | None = ...,
cgroup_parent: PathType | None = ...,
cni_config_dir: PathType | None = ...,
cni_plugin_path: PathType | None = ...,
compress: t.Literal[True, False] = ...,
cpu_period: int | None = ...,
cpu_quota: int | None = ...,
cpu_shares: int | None = ...,
cpuset_cpus: str | None = ...,
cpuset_mems: str | None = ...,
creds: str | dict[str, str] | ArgType = ...,
decryption_key: str | dict[str, str] | ArgType = ...,
device: str | ArgType = ...,
disable_compression: t.Literal[True, False] = ...,
dns: str | None = ...,
dns_option: str | ArgType = ...,
dns_search: str | ArgType = ...,
force_rm: t.Literal[True, False] = ...,
format: str | t.Literal["docker", "oci"] | None = ...,
http_proxy: t.Literal[True, False] = ...,
ignorefile: PathType | None = ...,
iidfile: PathType | None = ...,
ipc: str | PathType | None = ...,
isolation: str | None = ...,
jobs: int | None = ...,
layers: t.Literal[True, False] = ...,
logfile: PathType | None = ...,
manifest: str | None = ...,
memory: str | None = ...,
memory_swap: str | None = ...,
network: str | None = ...,
no_cache: t.Literal[True, False] = ...,
os: str | None = ...,
pid: PathType | None = ...,
platform: str | ArgType = ...,
pull: t.Literal[True, False] = ...,
pull_always: t.Literal[True, False] = ...,
pull_never: t.Literal[True, False] = ...,
quiet: t.Literal[True, False] = ...,
rm: t.Literal[True, False] = ...,
runtime: PathType | None = ...,
runtime_flag: str | dict[str, str] | ArgType = ...,
secret: str | dict[str, str] | ArgType = ...,
security_opt: str | ArgType = ...,
shm_size: str | None = ...,
sign_by: str | None = ...,
squash: t.Literal[True, False] = ...,
ssh: str | ArgType = ...,
stdin: t.Literal[True, False] = ...,
target: str | None = ...,
timestamp: int | None = ...,
tls_verify: t.Literal[True, False] = ...,
ulimit: str | dict[str, tuple[int, int]] | ArgType = ...,
userns: str | None = ...,
userns_gid_map_group: str | None = ...,
userns_uid_map_user: str | None = ...,
uts: str | None = ...,
variant: str | None = ...,
volume: str | tuple[str, str, str] | None = ...,
**kwargs: t.Any,
) -> t.Any: ...
[docs]
@inject
def build(
bento_tag: Tag | str,
backend: DefaultBuilder = "docker",
image_tag: tuple[str] | None = None,
features: t.Sequence[str] | None = None,
_bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
**kwargs: t.Any,
) -> t.Any:
"""
Build any given BentoML into a OCI-compliant image.
.. code-block:: python
import bentoml
bento = bentoml.get("pytorch_vgg:latest")
bentoml.container.build(bento, backend='podman', features=["grpc", "tracing"])
Args:
bento_tag: Bento tag in format of ``NAME:VERSION``
backend: The backend to use for building the image. Current supported builder backends
include ``docker``, ``podman``, ``buildah``, ``nerdctl``, ``buildctl``, and ``buildx``.
.. note::
``buildx`` is a syntatic sugar for ``docker buildx build``. See https://docs.docker.com/build/.
The reason for this is that ``buildx`` used to be the default behaviour of ``bentoml containerize``.
image_tag: Optional additional image tag to apply to the built image.
features: Optional features to include in the container file. See :ref:`concepts/bento:Python packages`
for additional BentoML features.
**kwargs: Additional keyword arguments to pass to the builder backend. Refer to the above overload
for each of the supported arguments per backend.
"""
from ._internal.container import determine_container_tag
# Run healthcheck
if not health(backend):
raise BentoMLException("Backend %s is not healthy." % backend)
if "tag" not in kwargs:
kwargs["tag"] = determine_container_tag(bento_tag, image_tag=image_tag)
bento = _bento_store.get(bento_tag)
logger.info("Building OCI-compliant image for %s with %s\n", bento.tag, backend)
return _internal_build(bento_tag, backend, features=features, **kwargs)
[docs]
@inject
def get_containerfile(
bento_tag: Tag | str,
output_path: str | None = None,
enable_buildkit: bool = True,
features: t.Sequence[str] | None = None,
_bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
):
"""
Returns the generated container file for a given Bento.
Note that the container file (Dockerfile) inside the Bento is minimal, whereas
this utility functions returns the container file that :ref:`bentoml containerize <reference/cli:containerize>` will
be using.
.. note::
If ``output_path`` is not specified, then the contents of the container file
will be printed out to ``sys.stderr``. If provided, then the final container file
will be written to that given path.
Args:
bento_tag: Given tag for the bento.
output_path: Optional output path to write the final container file to.
Note that if ``output_path`` is a directory, then the targeted file
will be ``output_path + os.sep + "<bento_tag>.dockerfile"``.
enable_buildkit: Whether the container file contains BuildKit syntax.
features: Optional features to include in the container file. See :ref:`concepts/bento:Python packages`
for additional BentoML features.
"""
bento = _bento_store.get(bento_tag)
with _internal_construct_containerfile(
bento,
enable_buildkit=enable_buildkit,
features=features,
add_header=True,
) as (_, final_containerfile):
if output_path is not None:
if os.path.isdir(output_path):
output_path = output_path = os.path.join(
output_path, f"{bento.tag.path().replace(os.sep,'_')}.dockerfile"
)
logger.info(
"Writting Containerfile for '%s' to '%s'.", bento.tag, output_path
)
shutil.copyfile(final_containerfile, output_path)
return
# otherwise we will just write this to stderr.
with open(final_containerfile, "r") as f:
sys.stderr.write(f.read())
__all__ = ["build", "health", "register_backend", "get_backend", "get_containerfile"]