Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 44 additions & 3 deletions tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import os
import pathlib
import re
import shlex
import shutil
import subprocess
import sys
Expand All @@ -17,7 +18,7 @@

import invoke
from invoke import Collection, task
from invoke.exceptions import UnexpectedExit
from invoke.exceptions import Exit, UnexpectedExit


def safe_value(fnc):
Expand Down Expand Up @@ -497,6 +498,43 @@ def manage(c, cmd, pty: bool = False, env=None, verbose: bool = False, **kwargs)
)


def manage_interactive(cmd: str, env=None, verbose: bool = False):
"""Run a Django management command with inherited stdio.

This bypasses Invoke PTY mediation and mirrors direct shell usage, which is
required for some interactive commands in Docker environments.

Args:
cmd: Django management command and arguments.
env: Optional environment variables to add for command execution.
verbose: If True, print the resolved command before execution.

Raises:
Exit: If the subprocess returns a non-zero exit code.
"""
args = ['python3', 'manage.py', *shlex.split(cmd)]

# Keep behavior aligned with `manage`: default to quiet output.
if '-v' not in cmd and '--verbosity' not in cmd:
args.extend(['-v', '1' if verbose else '0'])

if verbose:
info(f'Running interactive command: {" ".join(args)}')

cmd_env = dict(os.environ)
if env:
cmd_env.update(env)

# Avoid Invoke's PTY stdin mediation for interactive commands; run with
# inherited stdio to match direct `manage.py` behavior in Docker TTYs.
result = subprocess.run(args, cwd=manage_py_dir(), env=cmd_env, check=False)

if result.returncode != 0:
error(f"ERROR: InvenTree command failed: '{' '.join(args)}'")
warning('- Refer to the error messages in the log above for more information')
raise Exit(code=result.returncode)


def installed_apps(c) -> list[str]:
"""Returns a list of all installed apps, including plugins."""
result = manage(c, 'list_apps', pty=False, hide=True)
Expand Down Expand Up @@ -762,7 +800,7 @@ def shell(c):
@task
def superuser(c):
"""Create a superuser/admin account for the database."""
manage(c, 'createsuperuser', pty=True)
manage_interactive('createsuperuser')


@task
Expand Down Expand Up @@ -1456,7 +1494,10 @@ def delete_data(c, force: bool = False, migrate: bool = False, verbose: bool = F
if migrate:
manage(c, 'migrate --run-syncdb', verbose=verbose)

manage(c, f'flush{" --noinput" if force else ""}', verbose=verbose)
if force:
manage(c, 'flush --noinput', verbose=verbose)
else:
manage_interactive('flush', verbose=verbose)

success('Existing data deleted')

Expand Down
Loading