Bases: BaseService
A service class for Django operations.
This class provides methods for various Django management commands and operations.
The constructor for the DjangoService class.
Parameters:
Source code in dev_tool/services/django/service.py
| def __init__(self, strategy: ExecutionStrategyProtocol) -> None:
"""
The constructor for the DjangoService class.
:param strategy: The execution strategy for running commands.
"""
super().__init__()
self._strategy = strategy
|
A property that returns the ExecutionStrategy.
Returns:
A method that creates a Django server runner.
Parameters:
-
ip_address
(str)
–
The IP address to bind to.
-
port
(int)
–
The port number to bind to.
-
environment
(dict[str, str] | None, default:
None
)
–
Optional environment variables to pass to the server process.
Returns:
Source code in dev_tool/services/django/service.py
| def create_runner(self, ip_address: str, port: int, environment: dict[str, str] | None = None) -> DjangoServerRunner:
"""
A method that creates a Django server runner.
:param ip_address: The IP address to bind to.
:param port: The port number to bind to.
:param environment: Optional environment variables to pass to the server process.
:return: A server runner instance appropriate for the execution strategy.
"""
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()
runner = DjangoServerRunner(project_name=project_name, ip_address=ip_address, port=port)
if environment:
runner.environment = environment
return runner
|
A method that creates default Django superuser accounts.
This method requires Django environment variables to be properly set.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_create_superuser(self) -> None:
"""
A method that creates default Django superuser accounts.
This method requires Django environment variables to be properly set.
"""
superusers = [
{
'username': 'stratus',
'email': '[email protected]',
'password': 'stratus',
'first_name': 'Stratus',
'last_name': 'Adv'
},
]
for superuser in superusers:
username = superuser['username']
email = superuser['email']
password = superuser['password']
first_name = superuser.get('first_name', '')
last_name = superuser.get('last_name', '')
script = self._load_python_template('is_user.py', username=username)
try:
result = self.strategy.run_shell_script(script, check=False)
except Exception as exception:
message = f'Failed to check if superuser "{username}" exists'
log.exception(message)
raise DjangoSuperuserError(message) from exception
if result.returncode == 0:
message = f'Superuser "{username}" already exists. Skipping...'
emit_warning(message)
log.debug(message)
continue
message = f'Superuser "{username}" not found. Creating...'
emit_normal(message)
log.debug(message)
env_vars = {
'DJANGO_SUPERUSER_USERNAME': username,
'DJANGO_SUPERUSER_EMAIL': email,
'DJANGO_SUPERUSER_PASSWORD': password
}
stderr = ''
try:
if ExecutionStrategyProvider.is_containerized():
if isinstance(self.strategy, ContainerizedExecutionStrategy):
result = self.strategy.run_createsuperuser(env_vars)
stderr = result.stdout.decode() if result.stdout else ''
else:
env = os.environ.copy()
env.update(env_vars)
command = [VENV_PYTHON, BASE / 'manage.py', 'createsuperuser', '--noinput']
result = subprocess.run(
command,
capture_output=True,
check=True,
text=True,
env=env
)
stderr = result.stderr
except Exception as exception:
message = f'Unexpected error creating superuser "{username}"'
log.exception(message)
raise DjangoSuperuserError(message) from exception
if result.returncode != 0:
message = f'Failed to create superuser "{username}": {stderr}'
log.error(message)
raise DjangoSuperuserError(message)
message = f'Superuser "{username}" created successfully.'
emit_normal(message)
log.debug(message)
script = self._load_python_template(
'update_user.py',
username=username,
first_name=first_name,
last_name=last_name
)
try:
self.strategy.run_shell_script(script, check=True)
message = f"Superuser '{username}' updated with additional information."
emit_normal(message)
log.debug(message)
except subprocess.CalledProcessError as exception:
message = f'Failed to update superuser "{username}" information'
log.exception(message)
raise DjangoSuperuserError(message) from exception
except Exception as exception:
message = f'Unexpected error updating superuser "{username}"'
log.exception(message)
raise DjangoSuperuserError(message) from exception
|
A method that creates a normal Django user account.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_create_test_user(self) -> None:
"""A method that creates a normal Django user account."""
users = [
{
'username': 'test',
'first_name': 'Bobert',
'last_name': 'Bobertson',
'email': '[email protected]',
'password': 'test',
},
]
for user_data in users:
username = user_data['username']
first_name = user_data['first_name']
last_name = user_data['last_name']
script = self._load_python_template(
'create_user.py',
username=username,
first_name=first_name,
last_name=last_name,
email=user_data['email'],
password=user_data['password']
)
try:
self.strategy.run_shell_script(script, check=True)
message = f'User "{username}" created successfully.'
emit_normal(message)
log.debug(message)
except subprocess.CalledProcessError as exception:
message = f'Failed to create user "{username}"'
log.exception(message)
raise DjangoCommandError(message) from exception
|
A method that creates superusers and test users.
Source code in dev_tool/services/django/service.py
| def run_django_create_user(self) -> None:
"""A method that creates superusers and test users."""
self.run_django_create_superuser()
self.run_django_create_test_user()
|
A method that flushes the Django database.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_flush(self) -> None:
"""A method that flushes the Django database."""
message = 'Flushing the database...'
emit_normal(message)
log.debug(message)
try:
self.strategy.run_django_command(['flush', '--no-input'], check=True)
message = 'Database flushed successfully.'
emit_normal(message)
log.debug(message)
except subprocess.CalledProcessError as exception:
message = 'Failed to flush the database'
log.exception(message)
raise DjangoCommandError(message) from exception
|
A method that runs an interactive Django command prompt.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_interactive_shell(self) -> None:
"""A method that runs an interactive Django command prompt."""
user_input = input('Enter your Django command: ').strip()
args = user_input.split()
if not args:
message = 'No command entered. Please try again.'
emit_warning(message)
log.debug(message)
return
self.run_django_management_command(args)
|
A method that creates new Django migrations.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_make_migrations(self) -> None:
"""A method that creates new Django migrations."""
message = 'Creating migrations...'
emit_normal(message)
log.debug(message)
try:
self.strategy.run_django_command(['makemigrations'], check=True)
message = 'Migrations created successfully.'
emit_normal(message)
log.debug(message)
except subprocess.CalledProcessError as exception:
message = 'Failed to create migrations'
log.exception(message)
raise DjangoMigrationError(message) from exception
|
A method that runs a Django management command.
Parameters:
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_management_command(self, args: list[str]) -> None:
"""
A method that runs a Django management command.
:param args: The command arguments.
"""
try:
self.strategy.run_django_command(args, check=True)
except subprocess.CalledProcessError as exception:
message = f'Django command failed: {" ".join(args)}'
log.exception(message)
raise DjangoCommandError(message) from exception
|
A method that applies Django migrations.
Parameters:
-
apps
(str)
–
The specific apps to migrate, or empty string for all apps.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_migrate(self, apps: str) -> None:
"""
A method that applies Django migrations.
:param apps: The specific apps to migrate, or empty string for all apps.
"""
command = ['migrate']
if apps:
message = f'Running migrations for {apps}'
emit_normal(message)
log.debug(message)
command.append(apps)
else:
message = 'Running migrations for default database.'
emit_normal(message)
log.debug(message)
try:
self.strategy.run_django_command(command, check=True)
except subprocess.CalledProcessError as exception:
apps = apps or 'all apps'
message = f'Failed to run Django migrations for {apps}'
log.exception(message)
raise DjangoMigrationError(message) from exception
|
A method that runs database seeding for Django.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_seeding(self) -> None:
"""A method that runs database seeding for Django."""
seed_script = BASE / 'seed.py'
if seed_script.exists():
message = 'Running seed script...'
emit_normal(message)
log.debug(message)
try:
self.strategy.run_seed(seed_script)
except subprocess.CalledProcessError as exception:
message = 'Django seeding script failed'
log.exception(message)
raise DjangoSeedError(message) from exception
else:
message = 'No seed script found.'
emit_warning(message)
log.debug(message)
|
A method that starts a Django development server.
Parameters:
-
ip_address
(str)
–
The IP address to bind the server to.
-
port
(int)
–
The port number to bind the server to.
Source code in dev_tool/services/django/service.py
| @is_django_environment_variables
def run_django_server(self, ip_address: str, port: int) -> None:
"""
A method that starts a Django development server.
:param ip_address: The IP address to bind the server to.
:param port: The port number to bind the server to.
"""
try:
self.strategy.run_server(ip_address, port)
except Exception as exception:
message = 'Failed to start Django server'
log.exception(message)
raise DjangoServerError(message) from exception
|