Skip to content

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 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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 = create_docker_client()
    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 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
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.
    """

    app_image = self._app_image_name()

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

    self._build_image(app_image, DockerFileDefault.APP, {'PYTHON_VERSION': self._python_version()})

cleanup

A method that cleans up temporary build context directories.

Source code in dev_tool/services/execution/containerized.py 
520
521
522
523
524
525
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 
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
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(self._configuration.get_app_port())
    self._stop_container_using_port(self._configuration.get_ssh_port())
    self._stop_container_using_port(db_port)

    if recreate:
        self._compose_down(volumes=True)

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

    postgres_version = str(self._configuration.get_docker_config().get('postgres-version', DockerDefault.POSTGRES_VERSION))

    self._build_image(DockerImageDefault.DB, DockerFileDefault.DB, {'POSTGRES_VERSION': postgres_version})
    self._build_image(self._app_image_name(), DockerFileDefault.APP, {'PYTHON_VERSION': self._python_version()})

    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:

Source code in dev_tool/services/execution/containerized.py 
559
560
561
562
563
564
565
566
567
568
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:

Source code in dev_tool/services/execution/containerized.py 
570
571
572
573
574
575
576
577
578
579
580
581
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:

Source code in dev_tool/services/execution/containerized.py 
583
584
585
586
587
588
589
590
591
592
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:

Source code in dev_tool/services/execution/containerized.py 
594
595
596
597
598
599
600
601
602
603
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:

Source code in dev_tool/services/execution/containerized.py 
605
606
607
608
609
610
611
612
613
614
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_sass_command

A method that executes a Sass command in the container.

Parameters:

  • args (list[str]) –

    The command arguments.

  • kwargs (Any, default: {} ) –

    Additional subprocess arguments.

Returns:

Source code in dev_tool/services/execution/containerized.py 
616
617
618
619
620
621
622
623
624
625
def run_sass_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Sass command in the container.

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

    return self._exec(['sass', *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 
627
628
629
630
631
632
633
634
635
636
637
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_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:

Source code in dev_tool/services/execution/containerized.py 
639
640
641
642
643
644
645
646
647
648
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:

Source code in dev_tool/services/execution/local.py 
29
30
31
32
33
34
35
36
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 
38
39
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 
41
42
43
44
45
46
47
48
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:

Source code in dev_tool/services/execution/local.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
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:

Source code in dev_tool/services/execution/local.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
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:

Source code in dev_tool/services/execution/local.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
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:

Source code in dev_tool/services/execution/local.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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_sass_command

A method that executes a Sass command locally.

Parameters:

  • args (list[str]) –

    The command arguments.

  • kwargs (Any, default: {} ) –

    Additional subprocess arguments.

Returns:

Source code in dev_tool/services/execution/local.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
def run_sass_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Sass command locally.

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

    command = ['sass', *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 
120
121
122
123
124
125
126
127
128
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_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:

Source code in dev_tool/services/execution/local.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
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:

Source code in dev_tool/services/execution/provider.py 
21
22
23
24
25
26
27
28
29
@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:

Source code in dev_tool/services/execution/provider.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@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 
62
63
64
65
66
67
68
69
70
71
72
@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 
74
75
76
77
78
@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 
21
22
23
@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 
25
26
27
28
29
30
31
@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:

Source code in dev_tool/services/execution/strategy.py 
33
34
35
36
37
38
39
40
41
@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:

Source code in dev_tool/services/execution/strategy.py 
43
44
45
46
47
48
49
50
51
@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:

Source code in dev_tool/services/execution/strategy.py 
53
54
55
56
57
58
59
60
61
@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:

Source code in dev_tool/services/execution/strategy.py 
63
64
65
66
67
68
69
70
71
@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_sass_command abstractmethod

A method that executes a Sass command.

Parameters:

  • args (list[str]) –

    The command arguments.

  • kwargs (Any, default: {} ) –

    Additional subprocess arguments.

Returns:

Source code in dev_tool/services/execution/strategy.py 
73
74
75
76
77
78
79
80
81
@abstractmethod
def run_sass_command(self, args: list[str], **kwargs: Any) -> subprocess.CompletedProcess:
    """
    A method that executes a Sass 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 
83
84
85
86
87
88
89
@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_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:

Source code in dev_tool/services/execution/strategy.py 
91
92
93
94
95
96
97
98
99
@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.
    """