ds_pycontain code documentation

Submodules

ds_pycontain.docker_containers module

class DockerContainer(image: DockerImage, **kwargs: Any)

Bases: object

An isolated environment for running commands, based on docker container.

Examples:

If you need to run container for a single job:

>>> container = DockerContainer(DockerImage.from_tag("alpine"))
>>> status_code, logs = container.spawn_run("echo hello world")

To run a container in background and execute commands:

>>> with DockerContainer(DockerImage.from_tag("alpine")) as container:
>>>     status_code, logs = container.run("echo hello world")

Parameters:

image (DockerImage) –
kwargs (Any) –

__enter__() → DockerContainer

Enters container context. It means that container is started and you can execute commands inside it.

Return type:: DockerContainer

__exit__(exc_type: Optional[Type[BaseException]], exc: Optional[BaseException], traceback: Optional[TracebackType]) → bool

Cleanup container on exit. :param exc_type: exception type :param exc: exception instance (unused) :param traceback: traceback object (unused) :return: True if exception was handled, False otherwise

Parameters:

exc_type (Optional[Type[BaseException]]) –
exc (Optional[BaseException]) –
traceback (Optional[TracebackType]) –

Return type:

bool

property docker_container: Container: Returns docker container object. :return: docker container object

property name: str: Name of the container if it exists, empty string otherwise. :return: container name as string.

run(command: Union[str, List[str]], **kwargs: Any) → Tuple[int, Union[bytes, Tuple[bytes, bytes], Generator[bytes, None, None]]]

Run a script in the isolated environment which is docker container. You can send any args which docker-py exec_run accepts: https://docker-py.readthedocs.io/en/stable/containers.html#docker.models.containers.Container.exec_run Return is a tuple of exit code and output which is controlled by arguments: stream, socket and demux.

Parameters:

command (Union[str, List[str]]) – command to execute in the container
kwargs (Any) – additional arguments to pass to docker client containers.run()

Returns:

tuple of exit code and output (stream, socket or demux)

Return type:

Tuple[int, Union[bytes, Tuple[bytes, bytes], Generator[bytes, None, None]]]

spawn_run(command: Union[str, List[str]], **kwargs: Any) → Tuple[int, bytes]

Run a script in the isolated environment which is docker container with the same lifetime as this function call.

You can also pass all arguments that docker client containers.run() accepts. It blocks till command is finished.

Parameters:

command (Union[str, List[str]]) – command to execute in the container
kwargs (Any) – additional arguments to pass to docker client containers.run()

Returns:

tuple of exit code and logs

Return type:

Tuple[int, bytes]

unsafe_exit() → None

Cleanup container on exit. Please prefer to use with statement.

Return type:: None

unsafe_start() → None

Starts container without entering it. Please prefer to use with DockerContainer statement.

Return type:: None

class DockerImage(name: str)

Bases: object

Represents a locally available docker image as a tag. You can either use existing docker image or build a new one from Dockerfile.

>>> image = DockerImage.from_tag("alpine")
>>> image = DockerImage.from_tag("python", tag="3.9-slim")
>>> image = DockerImage.from_dockerfile("example/Dockerfile")
>>> image = DockerImage.from_dockerfile("path/to/dir_with_Dockerfile/", name="cow")

Parameters:: name (str) –

__repr__() → str

String representation of the object. :return: string representation of the object (container name).

Return type:: str

classmethod exists(name: str) → bool

Checks if the docker image exists. :param name: docker image name with tag, e.g. “alpine:latest” :return: True if docker image exists, False otherwise

Parameters:: name (str) –
Return type:: bool

classmethod from_dockerfile(dockerfile_path: ~typing.Union[~pathlib.Path, str], name: ~typing.Union[str, ~typing.Callable[[], str]] = <function generate_random_container_tag>, **kwargs: ~typing.Any) → DockerImage

Build a new image from Dockerfile given its file path.

Parameters:

dockerfile_path (Union[Path, str]) – path to Dockerfile
name (Union[str, Callable[[], str]]) – name of the image to build or name generator function
kwargs (Any) –

Return type:

DockerImage

defaults to generate_random_container_tag() :param kwargs: additional arguments to pass to docker client images.build() :return: DockerImage object representing built image on the system. :raises ValueError: if dockerfile_path is not a valid path to Dockerfile.

classmethod from_dockerfile_content(dockerfile_str: str, name: ~typing.Union[str, ~typing.Callable[[], str]] = <function generate_random_container_tag>, **kwargs: ~typing.Any) → DockerImage

Build a new image from Dockerfile given a string with Dockerfile content.

Parameters:

dockerfile_str (str) – string with Dockerfile content.
name (Union[str, Callable[[], str]]) – name of the image to build or name generator function
kwargs (Any) –

Return type:

DockerImage

defaults to generate_random_container_tag() :param kwargs: additional arguments to pass to docker client images.build() :return: DockerImage object representing built image on the system.

classmethod from_tag(repository: str, tag: str = 'latest', auth_config: Optional[Dict[str, str]] = None) → DockerImage

Use image with a given repository and tag. It is going to pull it if it is not present on the system.

Examples:

>>> repository = "alpine" # (will get "latest" tag)
>>> repository = "python", tag = "3.9-slim"

Parameters:

repository (str) – docker image repository, e.g. “alpine”.
tag (str) – docker image tag, e.g. “latest”.
auth_config (Optional[Dict[str, str]]) – authentication configuration for private repositories.

Returns:

DockerImage object representing pulled image on the system.

Return type:

DockerImage

classmethod remove(name: str) → None

WARNING: Removes image from the system, be cautious with this function. It is irreversible operation!. :param name: docker image name with tag, e.g. “alpine:latest”

Parameters:: name (str) –
Return type:: None

generate_random_container_tag() → str

Generates a random tag for a docker container. Format: ds_pycontain_runner:YYYY-MM-DD-HH-MM-SS-<8 random chars>

Returns:: random tag for a docker container.
Return type:: str

get_docker_client(**kwargs: Any) → DockerClient

cached version to retrieve docker client. By default it will use environment variables to connect to docker daemon.

Parameters:: kwargs (Any) – additional arguments to pass to docker client
Returns:: docker client object
Return type:: DockerClient

ds_pycontain.python_dockerized_repl module

class PythonContainerREPL(port: int = 7123, image: Optional[DockerImage] = None, base_image: str = 'python:3.11-alpine3.18', docker_startup_timeout: float = 300.0, repl_startup_timeout: float = 3.0, **kwargs: Dict[str, Any])

Bases: object

This class is a wrapper around the docker container that runs the python REPL server. It is used to execute python code in the container and return the results.

It assumes specific docker image is used which runs langchain python runner server and it communicates by HTTP requests.

Parameters:

port (int) –
image (Optional[DockerImage]) –
base_image (str) –
docker_startup_timeout (float) –
repl_startup_timeout (float) –
kwargs (Dict[str, Any]) –

__del__() → None

Closes container and removes it.

Return type:: None

eval(code: str) → str

Evaluate code and return result as string.

Parameters:: code (str) – code to evaluate.
Returns:: result as string.
Return type:: str

exec(code: str) → str

Execute code and return stdout.

Parameters:: code (str) – code to execute.
Returns:: result as string.
Return type:: str

run(command: str, timeout: Optional[int] = None) → str

Run command and returns anything printed. Timeout, if provided, is not currently supported and will be ignored.

Parameters:

command (str) – command to run.
timeout (Optional[int]) – timeout in seconds.

Returns:

result from REPL as output.

Return type:

str

Module contents

ds_pycontain is a Python package for managing Docker containers and images.

class DockerContainer(image: DockerImage, **kwargs: Any)

Bases: object

An isolated environment for running commands, based on docker container.

Examples:

If you need to run container for a single job:

>>> container = DockerContainer(DockerImage.from_tag("alpine"))
>>> status_code, logs = container.spawn_run("echo hello world")

To run a container in background and execute commands:

>>> with DockerContainer(DockerImage.from_tag("alpine")) as container:
>>>     status_code, logs = container.run("echo hello world")

Parameters:

image (DockerImage) –
kwargs (Any) –

__enter__() → DockerContainer

Enters container context. It means that container is started and you can execute commands inside it.

Return type:: DockerContainer

__exit__(exc_type: Optional[Type[BaseException]], exc: Optional[BaseException], traceback: Optional[TracebackType]) → bool

Cleanup container on exit. :param exc_type: exception type :param exc: exception instance (unused) :param traceback: traceback object (unused) :return: True if exception was handled, False otherwise

Parameters:

exc_type (Optional[Type[BaseException]]) –
exc (Optional[BaseException]) –
traceback (Optional[TracebackType]) –

Return type:

bool

property docker_container: Container: Returns docker container object. :return: docker container object

property name: str: Name of the container if it exists, empty string otherwise. :return: container name as string.

run(command: Union[str, List[str]], **kwargs: Any) → Tuple[int, Union[bytes, Tuple[bytes, bytes], Generator[bytes, None, None]]]

Run a script in the isolated environment which is docker container. You can send any args which docker-py exec_run accepts: https://docker-py.readthedocs.io/en/stable/containers.html#docker.models.containers.Container.exec_run Return is a tuple of exit code and output which is controlled by arguments: stream, socket and demux.

Parameters:

command (Union[str, List[str]]) – command to execute in the container
kwargs (Any) – additional arguments to pass to docker client containers.run()

Returns:

tuple of exit code and output (stream, socket or demux)

Return type:

Tuple[int, Union[bytes, Tuple[bytes, bytes], Generator[bytes, None, None]]]

spawn_run(command: Union[str, List[str]], **kwargs: Any) → Tuple[int, bytes]

Run a script in the isolated environment which is docker container with the same lifetime as this function call.

You can also pass all arguments that docker client containers.run() accepts. It blocks till command is finished.

Parameters:

command (Union[str, List[str]]) – command to execute in the container
kwargs (Any) – additional arguments to pass to docker client containers.run()

Returns:

tuple of exit code and logs

Return type:

Tuple[int, bytes]

unsafe_exit() → None

Cleanup container on exit. Please prefer to use with statement.

Return type:: None

unsafe_start() → None

Starts container without entering it. Please prefer to use with DockerContainer statement.

Return type:: None

class DockerImage(name: str)

Bases: object

Represents a locally available docker image as a tag. You can either use existing docker image or build a new one from Dockerfile.

>>> image = DockerImage.from_tag("alpine")
>>> image = DockerImage.from_tag("python", tag="3.9-slim")
>>> image = DockerImage.from_dockerfile("example/Dockerfile")
>>> image = DockerImage.from_dockerfile("path/to/dir_with_Dockerfile/", name="cow")

Parameters:: name (str) –

__repr__() → str

String representation of the object. :return: string representation of the object (container name).

Return type:: str

classmethod exists(name: str) → bool

Checks if the docker image exists. :param name: docker image name with tag, e.g. “alpine:latest” :return: True if docker image exists, False otherwise

Parameters:: name (str) –
Return type:: bool

classmethod from_dockerfile(dockerfile_path: ~typing.Union[~pathlib.Path, str], name: ~typing.Union[str, ~typing.Callable[[], str]] = <function generate_random_container_tag>, **kwargs: ~typing.Any) → DockerImage

Build a new image from Dockerfile given its file path.

Parameters:

dockerfile_path (Union[Path, str]) – path to Dockerfile
name (Union[str, Callable[[], str]]) – name of the image to build or name generator function
kwargs (Any) –

Return type:

DockerImage

defaults to generate_random_container_tag() :param kwargs: additional arguments to pass to docker client images.build() :return: DockerImage object representing built image on the system. :raises ValueError: if dockerfile_path is not a valid path to Dockerfile.

classmethod from_dockerfile_content(dockerfile_str: str, name: ~typing.Union[str, ~typing.Callable[[], str]] = <function generate_random_container_tag>, **kwargs: ~typing.Any) → DockerImage

Build a new image from Dockerfile given a string with Dockerfile content.

Parameters:

dockerfile_str (str) – string with Dockerfile content.
name (Union[str, Callable[[], str]]) – name of the image to build or name generator function
kwargs (Any) –

Return type:

DockerImage

defaults to generate_random_container_tag() :param kwargs: additional arguments to pass to docker client images.build() :return: DockerImage object representing built image on the system.

classmethod from_tag(repository: str, tag: str = 'latest', auth_config: Optional[Dict[str, str]] = None) → DockerImage

Use image with a given repository and tag. It is going to pull it if it is not present on the system.

Examples:

>>> repository = "alpine" # (will get "latest" tag)
>>> repository = "python", tag = "3.9-slim"

Parameters:

repository (str) – docker image repository, e.g. “alpine”.
tag (str) – docker image tag, e.g. “latest”.
auth_config (Optional[Dict[str, str]]) – authentication configuration for private repositories.

Returns:

DockerImage object representing pulled image on the system.

Return type:

DockerImage

classmethod remove(name: str) → None

WARNING: Removes image from the system, be cautious with this function. It is irreversible operation!. :param name: docker image name with tag, e.g. “alpine:latest”

Parameters:: name (str) –
Return type:: None

generate_random_container_tag() → str

Generates a random tag for a docker container. Format: ds_pycontain_runner:YYYY-MM-DD-HH-MM-SS-<8 random chars>

Returns:: random tag for a docker container.
Return type:: str

get_docker_client(**kwargs: Any) → DockerClient

cached version to retrieve docker client. By default it will use environment variables to connect to docker daemon.

Parameters:: kwargs (Any) – additional arguments to pass to docker client
Returns:: docker client object
Return type:: DockerClient