ccproxy.docker.protocol

ccproxy.docker.protocol

Protocol definition for Docker operations.

DockerAdapterProtocol

Bases: Protocol

Protocol for Docker operations.

is_available

is_available()

Check if Docker is available on the system.

Returns:

Type	Description
Awaitable[bool]	True if Docker is available, False otherwise

Source code in ccproxy/docker/protocol.py

def is_available(self) -> Awaitable[bool]:
    """Check if Docker is available on the system.

    Returns:
        True if Docker is available, False otherwise
    """
    ...

run

run(
    image,
    volumes,
    environment,
    command=None,
    middleware=None,
    user_context=None,
    entrypoint=None,
    ports=None,
)

Run a Docker container with specified configuration.

Alias for run_container method.

Parameters:

Name	Type	Description	Default
image	str	Docker image name/tag to run	required
volumes	list[DockerVolume]	List of volume mounts (host_path, container_path)	required
environment	DockerEnv	Dictionary of environment variables	required
command	list[str] | None	Optional command to run in the container	None
middleware	OutputMiddleware[T] | None	Optional middleware for processing output	None
user_context	DockerUserContext | None	Optional user context for Docker --user flag	None
entrypoint	str | None	Optional custom entrypoint to override the image's default	None
ports	list[DockerPortSpec] | None	Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])	None

Returns:

Type	Description
Awaitable[ProcessResult[T]]	Tuple containing (return_code, stdout_lines, stderr_lines)

Raises:

Type	Description
DockerError	If the container fails to run

Source code in ccproxy/docker/protocol.py

def run(
    self,
    image: str,
    volumes: list[DockerVolume],
    environment: DockerEnv,
    command: list[str] | None = None,
    middleware: OutputMiddleware[T] | None = None,
    user_context: DockerUserContext | None = None,
    entrypoint: str | None = None,
    ports: list[DockerPortSpec] | None = None,
) -> Awaitable[ProcessResult[T]]:
    """Run a Docker container with specified configuration.

    Alias for run_container method.

    Args:
        image: Docker image name/tag to run
        volumes: List of volume mounts (host_path, container_path)
        environment: Dictionary of environment variables
        command: Optional command to run in the container
        middleware: Optional middleware for processing output
        user_context: Optional user context for Docker --user flag
        entrypoint: Optional custom entrypoint to override the image's default
        ports: Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])

    Returns:
        Tuple containing (return_code, stdout_lines, stderr_lines)

    Raises:
        DockerError: If the container fails to run
    """
    ...

run_container

run_container(
    image,
    volumes,
    environment,
    command=None,
    middleware=None,
    user_context=None,
    entrypoint=None,
    ports=None,
)

Run a Docker container with specified configuration.

Parameters:

Name	Type	Description	Default
image	str	Docker image name/tag to run	required
volumes	list[DockerVolume]	List of volume mounts (host_path, container_path)	required
environment	DockerEnv	Dictionary of environment variables	required
command	list[str] | None	Optional command to run in the container	None
middleware	OutputMiddleware[T] | None	Optional middleware for processing output	None
user_context	DockerUserContext | None	Optional user context for Docker --user flag	None
entrypoint	str | None	Optional custom entrypoint to override the image's default	None
ports	list[DockerPortSpec] | None	Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])	None

Returns:

Type	Description
Awaitable[ProcessResult[T]]	Tuple containing (return_code, stdout_lines, stderr_lines)

Raises:

Type	Description
DockerError	If the container fails to run

Source code in ccproxy/docker/protocol.py

def run_container(
    self,
    image: str,
    volumes: list[DockerVolume],
    environment: DockerEnv,
    command: list[str] | None = None,
    middleware: OutputMiddleware[T] | None = None,
    user_context: DockerUserContext | None = None,
    entrypoint: str | None = None,
    ports: list[DockerPortSpec] | None = None,
) -> Awaitable[ProcessResult[T]]:
    """Run a Docker container with specified configuration.

    Args:
        image: Docker image name/tag to run
        volumes: List of volume mounts (host_path, container_path)
        environment: Dictionary of environment variables
        command: Optional command to run in the container
        middleware: Optional middleware for processing output
        user_context: Optional user context for Docker --user flag
        entrypoint: Optional custom entrypoint to override the image's default
        ports: Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])

    Returns:
        Tuple containing (return_code, stdout_lines, stderr_lines)

    Raises:
        DockerError: If the container fails to run
    """
    ...

exec_container

exec_container(
    image,
    volumes,
    environment,
    command=None,
    user_context=None,
    entrypoint=None,
    ports=None,
)

Execute a Docker container by replacing the current process.

This method builds the Docker command and replaces the current process with the Docker command using os.execvp, effectively handing over control to Docker.

Parameters:

Name	Type	Description	Default
image	str	Docker image name/tag to run	required
volumes	list[DockerVolume]	List of volume mounts (host_path, container_path)	required
environment	DockerEnv	Dictionary of environment variables	required
command	list[str] | None	Optional command to run in the container	None
user_context	DockerUserContext | None	Optional user context for Docker --user flag	None
entrypoint	str | None	Optional custom entrypoint to override the image's default	None
ports	list[DockerPortSpec] | None	Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])	None

Raises:

Type	Description
DockerError	If the container fails to execute
OSError	If the command cannot be executed

Source code in ccproxy/docker/protocol.py

def exec_container(
    self,
    image: str,
    volumes: list[DockerVolume],
    environment: DockerEnv,
    command: list[str] | None = None,
    user_context: DockerUserContext | None = None,
    entrypoint: str | None = None,
    ports: list[DockerPortSpec] | None = None,
) -> None:
    """Execute a Docker container by replacing the current process.

    This method builds the Docker command and replaces the current process
    with the Docker command using os.execvp, effectively handing over control to Docker.

    Args:
        image: Docker image name/tag to run
        volumes: List of volume mounts (host_path, container_path)
        environment: Dictionary of environment variables
        command: Optional command to run in the container
        user_context: Optional user context for Docker --user flag
        entrypoint: Optional custom entrypoint to override the image's default
        ports: Optional port specifications (e.g., ["8080:80", "localhost:9000:9000"])

    Raises:
        DockerError: If the container fails to execute
        OSError: If the command cannot be executed
    """
    ...

build_image

build_image(
    dockerfile_dir,
    image_name,
    image_tag="latest",
    no_cache=False,
    middleware=None,
)

Build a Docker image from a Dockerfile.

Parameters:

Name	Type	Description	Default
dockerfile_dir	Path	Directory containing the Dockerfile	required
image_name	str	Name to tag the built image with	required
image_tag	str	Tag to use for the image	'latest'
no_cache	bool	Whether to use Docker's cache during build	False
middleware	OutputMiddleware[T] | None	Optional middleware for processing output	None

Returns:

Type	Description
Awaitable[ProcessResult[T]]	ProcessResult containing (return_code, stdout_lines, stderr_lines)

Raises:

Type	Description
DockerError	If the image fails to build

Source code in ccproxy/docker/protocol.py

def build_image(
    self,
    dockerfile_dir: Path,
    image_name: str,
    image_tag: str = "latest",
    no_cache: bool = False,
    middleware: OutputMiddleware[T] | None = None,
) -> Awaitable[ProcessResult[T]]:
    """Build a Docker image from a Dockerfile.

    Args:
        dockerfile_dir: Directory containing the Dockerfile
        image_name: Name to tag the built image with
        image_tag: Tag to use for the image
        no_cache: Whether to use Docker's cache during build
        middleware: Optional middleware for processing output

    Returns:
        ProcessResult containing (return_code, stdout_lines, stderr_lines)

    Raises:
        DockerError: If the image fails to build
    """
    ...

image_exists

image_exists(image_name, image_tag='latest')

Check if a Docker image exists locally.

Parameters:

Name	Type	Description	Default
image_name	str	Name of the image to check	required
image_tag	str	Tag of the image to check	'latest'

Returns:

Type	Description
Awaitable[bool]	True if the image exists locally, False otherwise

Source code in ccproxy/docker/protocol.py

def image_exists(
    self, image_name: str, image_tag: str = "latest"
) -> Awaitable[bool]:
    """Check if a Docker image exists locally.

    Args:
        image_name: Name of the image to check
        image_tag: Tag of the image to check

    Returns:
        True if the image exists locally, False otherwise
    """
    ...

pull_image

pull_image(image_name, image_tag='latest', middleware=None)

Pull a Docker image from registry.

Parameters:

Name	Type	Description	Default
image_name	str	Name of the image to pull	required
image_tag	str	Tag of the image to pull	'latest'
middleware	OutputMiddleware[T] | None	Optional middleware for processing output	None

Returns:

Type	Description
Awaitable[ProcessResult[T]]	ProcessResult containing (return_code, stdout_lines, stderr_lines)

Raises:

Type	Description
DockerError	If the image fails to pull

Source code in ccproxy/docker/protocol.py

def pull_image(
    self,
    image_name: str,
    image_tag: str = "latest",
    middleware: OutputMiddleware[T] | None = None,
) -> Awaitable[ProcessResult[T]]:
    """Pull a Docker image from registry.

    Args:
        image_name: Name of the image to pull
        image_tag: Tag of the image to pull
        middleware: Optional middleware for processing output

    Returns:
        ProcessResult containing (return_code, stdout_lines, stderr_lines)

    Raises:
        DockerError: If the image fails to pull
    """
    ...