execution

dev_tool.services.execution

__all__ = ['ContainerizedExecutionStrategy', 'ExecutionStrategy', 'ExecutionStrategyProvider', 'LocalExecutionStrategy'] 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)

LocalExecutionStrategy

Bases: ExecutionStrategy

An execution strategy for local development with only the database in Docker.

This is the legacy mode where Django runs locally and only PostgreSQL runs in a container.

The constructor for the LocalExecutionStrategy class.

Parameters:

• docker_service (DockerServiceProtocol) –

  The Docker service for container management.

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

def __init__(self, docker_service: DockerServiceProtocol) -> None:
    """
    The constructor for the LocalExecutionStrategy class.

    :param docker_service: The Docker service for container management.
    """

    self._docker_service = docker_service

cleanup

A method that performs any necessary cleanup.

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

def cleanup(self) -> None:
    """A method that performs any necessary cleanup."""

ensure_database

A method that ensures the database container is available.

Parameters:

• recreate (bool, default: False ) –

  Whether to recreate the database container.

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

def ensure_database(self, recreate: bool = False) -> None:
    """
    A method that ensures the database container is available.

    :param recreate: Whether to recreate the database container.
    """

    self._docker_service.ensure_local_container(recreate=recreate)

run_bun_command

A method that executes a Bun command locally.

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/local.py

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

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

    command = ['bun', *args]

    check = kwargs.pop('check', False)
    return subprocess.run(command, **kwargs, check=check)

run_django_command

A method that executes a Django management command locally.

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/local.py

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

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

    command = [VENV_PYTHON, BASE / 'manage.py', *args]

    check = kwargs.pop('check', False)
    return subprocess.run(command, **kwargs, check=check)

run_pip_command

A method that executes a uv pip command locally.

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/local.py

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

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

    command = ['uv', 'pip', *args]

    check = kwargs.pop('check', False)
    return subprocess.run(command, **kwargs, check=check)

run_python_command

A method that executes a Python command locally.

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/local.py

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

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

    command = [VENV_PYTHON, *args]

    check = kwargs.pop('check', False)
    return subprocess.run(command, **kwargs, check=check)

run_seed

A method that runs the database seeding script locally.

Parameters:

• seed_script (Path) –

  The path to the seed script.

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

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

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

    command = [VENV_PYTHON, seed_script]
    subprocess.run(command, check=True)

run_server

A method that starts the Django development server locally.

Parameters:

• ip_address (str) –

  The IP address to bind to.

• port (int) –

  The port number to bind to.

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

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

    :param ip_address: The IP address to bind to.
    :param port: The port number to bind to.
    """

    from dev_tool.context import CONTEXT  # noqa: PLC0415
    from dev_tool.services.django.runner import DjangoServerRunner  # noqa: PLC0415

    project_name = CONTEXT.configuration.get_project_name()
    server = DjangoServerRunner(project_name=project_name, ip_address=ip_address, port=port)
    server.run_and_wait()

run_shell_script

A method that executes a Django shell script locally.

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/local.py

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

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

    command = [VENV_PYTHON, BASE / 'manage.py', 'shell', '-c', script]

    check = kwargs.pop('check', False)
    return subprocess.run(command, **kwargs, check=check)

ExecutionStrategyProvider

A provider class for managing the singleton ExecutionStrategy instance.

This class centralizes strategy creation to avoid duplicate logic scattered across multiple services.

configure classmethod

A method that configures the provider with required dependencies.

Parameters:

• docker_service (DockerServiceProtocol) –

  The Docker service for container management.

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

@classmethod
def configure(cls, docker_service: DockerServiceProtocol) -> None:
    """
    A method that configures the provider with required dependencies.

    :param docker_service: The Docker service for container management.
    """

    cls._docker_service = docker_service

get classmethod

A method that returns the singleton ExecutionStrategy instance.

Returns:

• ExecutionStrategy –

  The ExecutionStrategy instance.

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

@classmethod
def get(cls) -> ExecutionStrategy:
    """
    A method that returns the singleton ExecutionStrategy instance.

    :return: The ExecutionStrategy instance.
    """

    if cls._instance is None:
        from dev_tool.context import CONTEXT  # noqa: PLC0415
        from dev_tool.services.execution.containerized import ContainerizedExecutionStrategy  # noqa: PLC0415
        from dev_tool.services.execution.local import LocalExecutionStrategy  # noqa: PLC0415

        containerized = CONTEXT.configuration.pyproject.get_dev_tool_config().get('containerized', False)

        if containerized:
            cls._instance = ContainerizedExecutionStrategy(
                configuration=CONTEXT.configuration,
                project_name=CONTEXT.configuration.get_project_name()
            )
        else:
            if cls._docker_service is None:
                message = 'ExecutionStrategyProvider not configured. Call configure() first.'
                raise RuntimeError(message)

            cls._instance = LocalExecutionStrategy(
                docker_service=cls._docker_service
            )

    return cls._instance

is_containerized classmethod

A method that checks if the current strategy is containerized.

Returns:

• bool –

  True if containerized, False otherwise.

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

@classmethod
def is_containerized(cls) -> bool:
    """
    A method that checks if the current strategy is containerized.

    :return: True if containerized, False otherwise.
    """

    from dev_tool.context import CONTEXT  # noqa: PLC0415

    return CONTEXT.configuration.pyproject.get_dev_tool_config().get('containerized', False)

reset classmethod

A method that resets the singleton instance.

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

@classmethod
def reset(cls) -> None:
    """A method that resets the singleton instance."""

    cls._instance = None

ExecutionStrategy

Bases: ABC

An abstract base class defining the interface for execution strategies.

This class allows swapping between containerized and local execution without scattering conditionals throughout the codebase.

cleanup abstractmethod

A method that performs any necessary cleanup.

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

@abstractmethod
def cleanup(self) -> None:
    """A method that performs any necessary cleanup."""

ensure_database abstractmethod

A method that ensures the database is available.

Parameters:

• recreate (bool, default: False ) –

  Whether to recreate the database container.

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

@abstractmethod
def ensure_database(self, recreate: bool = False) -> None:
    """
    A method that ensures the database is available.

    :param recreate: Whether to recreate the database container.
    """

run_bun_command abstractmethod

A method that executes a Bun command.

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/strategy.py

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

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

run_django_command abstractmethod

A method that executes a Django management command.

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/strategy.py

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

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

run_pip_command abstractmethod

A method that executes a pip command.

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/strategy.py

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

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

run_python_command abstractmethod

A method that executes a Python command.

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/strategy.py

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

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

run_seed abstractmethod

A method that runs the database seeding script.

Parameters:

• seed_script (Path) –

  The path to the seed script.

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

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

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

run_server abstractmethod

A method that starts the Django development server.

Parameters:

• ip_address (str) –

  The IP address to bind to.

• port (int) –

  The port number to bind to.

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

@abstractmethod
def run_server(self, ip_address: str, port: int) -> None:
    """
    A method that starts the Django development server.

    :param ip_address: The IP address to bind to.
    :param port: The port number to bind to.
    """

run_shell_script abstractmethod

A method that executes a Django shell script.

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/strategy.py

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

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