containerized

dev_tool.services.execution.containerized

log = logging.getLogger(__name__) module-attribute

ContainerizedExecutionStrategy

Bases: ExecutionStrategy, BaseService

An execution strategy for fully containerized development.

Both Django and PostgreSQL run in Docker containers via docker-compose.

The constructor for the ContainerizedExecutionStrategy class.

Parameters:

• configuration (ConfigManager) –

  The configuration manager instance.

• project_name (str) –

  The name of the project.

Source code in dev_tool/services/execution/containerized.py

def __init__(self, configuration: ConfigManager, project_name: str) -> None:
    """
    The constructor for the ContainerizedExecutionStrategy class.

    :param configuration: The configuration manager instance.
    :param project_name: The name of the project.
    """

    BaseService.__init__(self)

    self._client: DockerClient = from_env()
    self._configuration = configuration
    self._project_name = project_name
    self._compose_dir: Path | None = None

build_shared_app_image

A method that builds the shared app image.

Parameters:

• force (bool, default: False ) –

  Whether to force a rebuild even if image exists.

Source code in dev_tool/services/execution/containerized.py

def build_shared_app_image(self, force: bool = False) -> None:
    """
    A method that builds the shared app image.

    :param force: Whether to force a rebuild even if image exists.
    """

    if force:
        with contextlib.suppress(docker.errors.ImageNotFound):
            self._client.images.remove(DockerImageDefault.APP, force=True)

    self._build_image(DockerImageDefault.APP, DockerFileDefault.APP)

cleanup

A method that cleans up temporary build context directories.

Source code in dev_tool/services/execution/containerized.py

def cleanup(self) -> None:
    """A method that cleans up temporary build context directories."""

    if self._compose_dir and self._compose_dir.exists():
        shutil.rmtree(self._compose_dir, ignore_errors=True)
        self._compose_dir = None

ensure_database

A method that ensures Docker services are running.

Parameters:

• recreate (bool, default: False ) –

  Whether to recreate the containers.

Source code in dev_tool/services/execution/containerized.py

def ensure_database(self, recreate: bool = False) -> None:
    """
    A method that ensures Docker services are running.

    :param recreate: Whether to recreate the containers.
    """

    self._stop_foreign_container()

    db_port = int(os.environ.get('DATABASE_PORT', DevelopmentDefault.DATABASE_PORT))

    self._stop_container_using_port(DockerDefault.APP_PORT)
    self._stop_container_using_port(DockerDefault.SSH_PORT)
    self._stop_container_using_port(db_port)

    if recreate:
        self._compose_down()

        message = 'Rebuilding Docker services...'
        self.notification.normal_text(message)
        log.debug(message)

        with contextlib.suppress(docker.errors.ImageNotFound):
            self._client.images.remove(DockerImageDefault.DB, force=True)

        with contextlib.suppress(docker.errors.ImageNotFound):
            self._client.images.remove(DockerImageDefault.APP, force=True)

    self._build_image(DockerImageDefault.DB, DockerFileDefault.DB)
    self._build_image(DockerImageDefault.APP, DockerFileDefault.APP)

    self._compose_up()

    if recreate and self._wait_for_container(f'{self._project_name}{DockerContainerDefault.APP_SUFFIX}'):
        self._install_dependency()

run_bun_command

A method that executes a Bun command in the container.

Parameters:

• args (list[str]) –

  The command arguments.

• kwargs (Any, default: {} ) –

  Additional subprocess arguments.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_bun_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Bun command in the container.

    :param args: The command arguments.
    :param kwargs: Additional subprocess arguments.
    :return: The completed process result.
    """

    return self._exec(['bun', *args], **kwargs)

run_createsuperuser

A method that creates a Django superuser with environment variables.

Parameters:

• env_vars (dict[str, str]) –

  Environment variables for the superuser creation.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_createsuperuser(self, env_vars: dict[str, str]) -> subprocess.CompletedProcess:
    """
    A method that creates a Django superuser with environment variables.

    :param env_vars: Environment variables for the superuser creation.
    :return: The completed process result.
    """

    return self._exec(
        ['python', 'manage.py', 'createsuperuser', '--noinput'],
        environment=env_vars
    )

run_django_command

A method that executes a Django management command in the container.

Parameters:

• args (list[str]) –

  The command arguments.

• kwargs (Any, default: {} ) –

  Additional subprocess arguments.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_django_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Django management command in the container.

    :param args: The command arguments.
    :param kwargs: Additional subprocess arguments.
    :return: The completed process result.
    """

    return self._exec(['python', 'manage.py', *args], **kwargs)

run_pip_command

A method that executes a uv pip command in the container.

Parameters:

• args (list[str]) –

  The command arguments.

• kwargs (Any, default: {} ) –

  Additional subprocess arguments.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_pip_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a uv pip command in the container.

    :param args: The command arguments.
    :param kwargs: Additional subprocess arguments.
    :return: The completed process result.
    """

    return self._exec(['uv', 'pip', *args], **kwargs)

run_python_command

A method that executes a Python command in the container.

Parameters:

• args (list[str]) –

  The command arguments.

• kwargs (Any, default: {} ) –

  Additional subprocess arguments.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_python_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Python command in the container.

    :param args: The command arguments.
    :param kwargs: Additional subprocess arguments.
    :return: The completed process result.
    """

    return self._exec(['python', *args], **kwargs)

run_seed

A method that runs the database seeding script in the container.

Parameters:

• seed_script (Path) –

  The path to the seed script.

Source code in dev_tool/services/execution/containerized.py

def run_seed(self, seed_script: Path) -> None:
    """
    A method that runs the database seeding script in the container.

    :param seed_script: The path to the seed script.
    """

    relative_path = seed_script.relative_to(BASE)
    container_path = f'{ContainerPathDefault.APP_ROOT}/{relative_path.as_posix()}'

    self._exec(['python', container_path])

run_server

A method that starts the Django development server in the container.

Parameters:

• ip_address (str) –

  The IP address to bind to (ignored, uses 0.0.0.0 inside container).

• port (int) –

  The port number to bind to.

Source code in dev_tool/services/execution/containerized.py

def run_server(self, ip_address: str, port: int) -> None:  # noqa: ARG002
    """
    A method that starts the Django development server in the container.

    :param ip_address: The IP address to bind to (ignored, uses 0.0.0.0 inside container).
    :param port: The port number to bind to.
    """

    name = f'{self._project_name}{DockerContainerDefault.APP_SUFFIX}'

    try:
        self._client.containers.get(name)
    except docker.errors.NotFound:
        return

    stop_event = threading.Event()
    process = None

    def signal_handler(_signal: int, _frame: object) -> None:
        stop_event.set()

        if process:
            if sys.platform == OperatingSystem.WINDOWS:
                process.send_signal(signal.CTRL_BREAK_EVENT)
            else:
                process.send_signal(signal.SIGINT)

    def stream_output() -> None:
        if process is None or process.stdout is None:
            return

        while not stop_event.is_set():
            if process.poll() is not None:
                break

            try:
                line = process.stdout.readline()

                if not line:
                    break

                if 'Starting development server' in line:
                    line = f'Starting development server at http://127.0.0.1:{port}/\n'

                if 'Quit the server with CONTROL-C.' in line:
                    line = 'Quit the server with CTRL-C or CTRL-Z.\n'

                print(line, end='')  # noqa: T201
            except Exception:
                break

    creationflags = 0

    if sys.platform == OperatingSystem.WINDOWS:
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP

    command = [
        'docker', 'exec', '-it', name,
        'python', 'manage.py', 'runserver', f'0.0.0.0:{port}'
    ]

    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        creationflags=creationflags
    )

    output_thread = threading.Thread(target=stream_output, daemon=True)
    output_thread.start()

    handler = signal.getsignal(signal.SIGINT)

    try:
        signal.signal(signal.SIGINT, signal_handler)
        signal_registered = True
    except ValueError:
        signal_registered = False

    try:
        while not stop_event.is_set():
            if process.poll() is not None:
                break

            stop_event.wait(timeout=0.1)
    except KeyboardInterrupt:
        stop_event.set()
    finally:
        stop_event.set()

        if process and process.poll() is None:
            process.terminate()

            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()

        if process and process.stdout:
            process.stdout.close()

        if signal_registered:
            signal.signal(signal.SIGINT, handler)

run_shell_script

A method that executes a Django shell script in the container.

Parameters:

• script (str) –

  The Python script to execute in Django shell.

• kwargs (Any, default: {} ) –

  Additional subprocess arguments.

Returns:

• CompletedProcess –

  The completed process result.

Source code in dev_tool/services/execution/containerized.py

def run_shell_script(self, script: str, **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Django shell script in the container.

    :param script: The Python script to execute in Django shell.
    :param kwargs: Additional subprocess arguments.
    :return: The completed process result.
    """

    return self._exec(['python', 'manage.py', 'shell', '-c', script], **kwargs)