File docker_sdk_client.py has 787 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import base64
import json
import logging
import os
import queue

SdkDockerClient has 39 functions (exceeds 20 allowed). Consider refactoring.

Open

class SdkDockerClient(ContainerClient):
    """
    Class for managing Docker (or Podman) using the Python Docker SDK.

    The client also supports targeting Podman engines, as Podman is almost a drop-in replacement

Function run_container has 27 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def run_container(

Function create_container has 26 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def create_container(

Function create_container has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

    def create_container(
        self,
        image_name: str,
        *,
        name: Optional[str] = None,

• Read up
  Read up

Function start_container has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

    def start_container(
        self,
        container_name_or_id: str,
        stdin=None,
        interactive: bool = False,

• Read up
  Read up

Function exec_in_container has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def exec_in_container(
        self,
        container_name_or_id: str,
        command: Union[List[str], str],
        interactive=False,

• Read up
  Read up

Function push_image has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def push_image(self, docker_image: str) -> None:
        LOG.debug("Pushing Docker image: %s", docker_image)
        try:
            result = self.client().images.push(docker_image)
            # some SDK clients (e.g., 5.0.0) seem to return an error string, instead of raising

• Read up
  Read up

Function exec_in_container has 8 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def exec_in_container(

Function inspect_image has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def inspect_image(
        self,
        image_name: str,
        pull: bool = True,
        strip_wellknown_repo_prefixes: bool = True,

• Read up
  Read up

Function start_container has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def start_container(

Function remove_image has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def remove_image(self, image: str, force: bool = True):
        LOG.debug("Removing image %s %s", image, "(forced)" if force else "")
        try:
            self.client().images.remove(image=image, force=force)
        except ImageNotFound:

• Read up
  Read up

Function _create_client has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _create_client():
        from localstack.config import DOCKER_SDK_DEFAULT_RETRIES, DOCKER_SDK_DEFAULT_TIMEOUT_SECONDS

        for attempt in range(0, DOCKER_SDK_DEFAULT_RETRIES + 1):
            try:

• Read up
  Read up

Function list_containers has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def list_containers(self, filter: Union[List[str], str, None] = None, all=True) -> List[dict]:
        if filter:
            filter = [filter] if isinstance(filter, str) else filter
            filter = dict([f.split("=", 1) for f in filter])
        LOG.debug("Listing containers with filters: %s", filter)

• Read up
  Read up

Function get_container_logs has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def get_container_logs(self, container_name_or_id: str, safe: bool = False) -> str:
        try:
            container = self.client().containers.get(container_name_or_id)
            return to_str(container.logs())
        except NotFound:

• Read up
  Read up