Django

backup_local_db(context, file='')

Back up local db.

Source code in saritasa_invocations/django.py

@invoke.task
def backup_local_db(
    context: invoke.Context,
    file: str = "",
) -> None:
    """Back up local db."""
    db.backup_local_db(
        context,
        file=file,
        **load_django_db_settings(context),
    )

backup_remote_db(context, file='')

Make dump of remote db and download it.

Source code in saritasa_invocations/django.py

@invoke.task
def backup_remote_db(
    context: invoke.Context,
    file: str = "",
) -> str:
    """Make dump of remote db and download it."""
    settings = load_django_remote_env_db_settings(context)
    db_k8s.create_dump(
        context,
        file=file,
        **settings,
    )
    return db_k8s.get_dump(
        context,
        file=file,
    )

check_new_migrations(context)

Check if there is new migrations or not.

Source code in saritasa_invocations/django.py

@invoke.task
def check_new_migrations(context: invoke.Context) -> None:
    """Check if there is new migrations or not."""
    printing.print_success("Django: Checking migrations")
    manage(
        context,
        command="makemigrations --check --dry-run",
    )

createsuperuser(context, email='', username='', password='')

Create superuser.

Source code in saritasa_invocations/django.py

@invoke.task
def createsuperuser(
    context: invoke.Context,
    email: str = "",
    username: str = "",
    password: str = "",
) -> None:
    """Create superuser."""
    config = _config.Config.from_context(context)
    email_source = "cmd"
    username_source = "cmd"
    if not email:
        with contextlib.suppress(invoke.Failure):
            output = context.run(
                "git config user.email",
                echo=False,
                hide="out",
            )
            if output and (email := output.stdout.replace(" ", "").strip()):
                email_source = "git"
        if not email:
            email = config.django.default_superuser_email
            email_source = "config"
    if not username:
        with contextlib.suppress(invoke.Failure):
            output = context.run(
                "git config user.name",
                echo=False,
                hide="out",
            )
            if output and (username := output.stdout.replace(" ", "").strip()):
                username_source = "git"
        if not username:
            username = config.django.default_superuser_username
            username_source = "config"
    if not password:
        password = config.django.default_superuser_password

    printing.print_success(
        "Django: Creating superuser with the following ->\n"
        f"{email=} from {email_source}\n"
        f"{username=} from {username_source}",
    )

    responder_email = invoke.FailingResponder(
        pattern=rf"{config.django.verbose_email_name}.*: ",
        response=f"{email}\n",
        sentinel="That Email address is already taken.",
    )
    responder_user_name = invoke.Responder(
        pattern=rf"{config.django.verbose_username_name}.*: ",
        response=f"{username}\n",
    )
    password_pattern = config.django.verbose_password_name
    responder_password = invoke.Responder(
        pattern=rf"({password_pattern}: )|({password_pattern} \(again\): )",
        response=f"{password}\n",
    )
    try:
        manage(
            context,
            command="createsuperuser",
            watchers=(
                responder_email,
                responder_user_name,
                responder_password,
            ),
        )
    except invoke.Failure:
        printing.print_warn(
            "Superuser with that email already exists. Skipped.",
        )

dbshell(context)

Open database shell with credentials from either local or dev env.

Source code in saritasa_invocations/django.py

@invoke.task
def dbshell(context: invoke.Context) -> None:
    """Open database shell with credentials from either local or dev env."""
    printing.print_success("Entering DB shell")
    manage(context, command="dbshell")

load_db_dump(context, file='')

Reset db and load db dump.

Source code in saritasa_invocations/django.py

@invoke.task
def load_db_dump(context: invoke.Context, file: str = "") -> None:
    """Reset db and load db dump."""
    resetdb(context, apply_migrations=False)
    db.load_db_dump(
        context,
        file=file,
        **load_django_db_settings(context),
    )

load_django_db_settings(context)

Get database-related config from django settings.

Source code in saritasa_invocations/django.py

def load_django_db_settings(context: invoke.Context) -> dict[str, str]:
    """Get database-related config from django settings."""
    settings = load_django_settings(context)
    db_settings = settings.DATABASES["default"]
    return {
        "dbname": db_settings["NAME"],
        "host": db_settings["HOST"],
        "port": db_settings["PORT"],
        "username": str(db_settings["USER"]),
        "password": db_settings["PASSWORD"],
    }

load_django_remote_env_db_settings(context)

Load remote django settings from .env file.

Support two sources:

• single env variable from config.django.remote_db_url_config_name
• multiple env vars from config.django.remote_db_config_mapping

Requires python-decouple

https://github.com/HBNetwork/python-decouple

Source code in saritasa_invocations/django.py

def load_django_remote_env_db_settings(
    context: invoke.Context,
) -> dict[str, str]:
    """Load remote django settings from .env file.

    Support two sources:

    - single env variable from config.django.remote_db_url_config_name
    - multiple env vars from config.django.remote_db_config_mapping

    Requires python-decouple:
        https://github.com/HBNetwork/python-decouple

    """
    system.create_tmp_folder(context)
    env_path = ".tmp/.env.tmp"
    config = _config.Config.from_context(context)
    k8s.download_file(
        context,
        path_to_file_in_pod=config.django.path_to_remote_config_file,
        path_to_where_save_file=env_path,
    )

    # decouple could not be installed during project init
    # so we import decouple this way to avoid import errors
    # during project initialization

    import decouple

    env_config = decouple.Config(decouple.RepositoryEnv(env_path))
    pathlib.Path(env_path).unlink()
    if database_url := str(
        env_config(
            config.django.remote_db_url_config_name,
            env_config(
                config.django.remote_db_url_config_name.lower(),
                default="",
            ),
        ),
    ):
        parsed_url = urllib.parse.urlsplit(database_url)
        return {
            "dbname": parsed_url.path.lstrip("/"),
            "host": parsed_url.hostname or "",
            "port": str(parsed_url.port or ""),
            "username": str(parsed_url.username or ""),
            "password": str(parsed_url.password or ""),
        }
    return {
        arg: str(env_config(env_var, env_config(env_var.lower(), default="")))
        for arg, env_var in config.django.remote_db_config_mapping.items()
    }

load_django_settings(context)

Load django settings from settings file (DJANGO_SETTINGS_MODULE).

Source code in saritasa_invocations/django.py

def load_django_settings(context: invoke.Context):  # noqa: ANN201
    """Load django settings from settings file (DJANGO_SETTINGS_MODULE)."""
    config = _config.Config.from_context(context)
    os.environ["DJANGO_SETTINGS_MODULE"] = config.django.settings_path

    from django.conf import settings  # type: ignore

    return settings

load_remote_db(context, file='')

Make dump of remote db, download it and apply it.

Source code in saritasa_invocations/django.py

@invoke.task
def load_remote_db(
    context: invoke.Context,
    file: str = "",
) -> None:
    """Make dump of remote db, download it and apply it."""
    file = backup_remote_db(context, file=file)
    load_db_dump(context, file=file)

makemigrations(context)

Run makemigrations command and chown created migrations.

Source code in saritasa_invocations/django.py

@invoke.task
def makemigrations(context: invoke.Context) -> None:
    """Run makemigrations command and chown created migrations."""
    printing.print_success("Django: Make migrations")
    manage(context, command="makemigrations")
    if python.get_python_env() == python.PythonEnv.DOCKER:
        system.chown(context)

manage(context, command, docker_params='', watchers=(), **kwargs)

Run manage.py command.

This command also handle starting of required services and waiting DB to be ready.

---

context: Invoke context
command: Manage command
docker_params: Params for docker run
watchers: Automated responders to command
kwargs: additional arguments for python.run

Source code in saritasa_invocations/django.py

@invoke.task
def manage(
    context: invoke.Context,
    command: str,
    docker_params: str = "",
    watchers: collections.abc.Sequence[invoke.StreamWatcher] = (),
    **kwargs,
) -> invoke.runners.Result | None:
    """Run `manage.py` command.

    This command also handle starting of required services and waiting DB to
    be ready.

    Args:
    ----
        context: Invoke context
        command: Manage command
        docker_params: Params for docker run
        watchers: Automated responders to command
        kwargs: additional arguments for python.run

    """
    config = _config.Config.from_context(context)
    wait_for_database(context)
    return python.run(
        context,
        docker_params=docker_params,
        command=f"{config.django.manage_file_path} {command}",
        watchers=watchers,
        env={
            "DJANGO_SETTINGS_MODULE": config.django.settings_path,
        },
        **kwargs,
    )

migrate(context)

Run migrate command.

Source code in saritasa_invocations/django.py

@invoke.task
def migrate(context: invoke.Context) -> None:
    """Run `migrate` command."""
    printing.print_success("Django: Apply migrations")
    config = _config.Config.from_context(context)
    manage(context, command=config.django.migrate_command)

recompile_messages(context)

Generate and recompile translation messages.

https://docs.djangoproject.com/en/4.2/ref/django-admin/#makemessages

Source code in saritasa_invocations/django.py

@invoke.task
def recompile_messages(context: invoke.Context) -> None:
    """Generate and recompile translation messages.

    https://docs.djangoproject.com/en/4.2/ref/django-admin/#makemessages

    """
    printing.print_success("Recompiling translation messages")
    config = _config.Config.from_context(context)
    manage(
        context,
        command=f"makemessages {config.django.makemessages_params}",
    )
    manage(
        context,
        command=f"compilemessages {config.django.compilemessages_params}",
    )

resetdb(context, apply_migrations=True)

Reset database to initial state (including test DB).

Requires django-extensions

https://django-extensions.readthedocs.io/en/latest/installation_instructions.html

Source code in saritasa_invocations/django.py

@invoke.task
def resetdb(
    context: invoke.Context,
    apply_migrations: bool = True,
) -> None:
    """Reset database to initial state (including test DB).

    Requires django-extensions:
        https://django-extensions.readthedocs.io/en/latest/installation_instructions.html

    """
    printing.print_success("Reset database to its initial state")
    manage(context, command="drop_test_database --noinput")
    manage(context, command="reset_db -c --noinput")
    if not apply_migrations:
        return
    makemigrations(context)
    migrate(context)
    createsuperuser(context)
    set_default_site(context)

run(context)

Run development web-server.

Source code in saritasa_invocations/django.py

@invoke.task
def run(context: invoke.Context) -> None:
    """Run development web-server."""
    printing.print_success("Running app")
    config = _config.Config.from_context(context)
    manage(
        context,
        docker_params=config.django.runserver_docker_params,
        command="{command} {host}:{port} {params}".format(  # noqa: UP032
            command=config.django.runserver_command,
            host=config.django.runserver_host,
            port=config.django.runserver_port,
            params=config.django.runserver_params,
        ),
    )

set_default_site(context)

Set default site to localhost.

Set default site domain to localhost:8000 so get_absolute_url works correctly in local environment

Source code in saritasa_invocations/django.py

def set_default_site(context: invoke.Context) -> None:
    """Set default site to localhost.

    Set default site domain to `localhost:8000` so `get_absolute_url` works
    correctly in local environment

    """
    manage(
        context,
        command=(
            "set_default_site --name localhost:8000 --domain localhost:8000"
        ),
    )

shell(context, params='')

Shortcut for manage.py shell command.

Requires django-extensions

https://django-extensions.readthedocs.io/en/latest/installation_instructions.html

Additional params available here

https://django-extensions.readthedocs.io/en/latest/shell_plus.html

Source code in saritasa_invocations/django.py

@invoke.task
def shell(
    context: invoke.Context,
    params: str = "",
) -> None:
    """Shortcut for manage.py shell command.

    Requires django-extensions:
        https://django-extensions.readthedocs.io/en/latest/installation_instructions.html

    Additional params available here:
        https://django-extensions.readthedocs.io/en/latest/shell_plus.html

    """
    printing.print_success("Entering Django Shell")
    config = _config.Config.from_context(context)
    manage(
        context,
        command=f"{config.django.shell_command} {params}",
    )

show_urls(context, search='')

Show urls of project.

Use search param to filter urls by regex.

Not using grep to make cross-platform and also keep color output.

Source code in saritasa_invocations/django.py

@invoke.task
def show_urls(
    context: invoke.Context,
    search: str = "",
) -> None:
    """Show urls of project.

    Use search param to filter urls by regex.

    Not using grep to make cross-platform and also keep color output.

    """
    console = rich.console.Console()
    for url_info in manage(
        context,
        command="show_urls",
        hide=True,
    ).stdout.splitlines():
        if search and not re.search(search, url_info):
            continue
        url_parts = url_info.split()
        console.print(
            rich.text.Text.assemble(
                url_parts[0],
                " ",
                url_parts[1],
                " ",
                url_parts[2],
            ),
        )

startapp(context)

Create new django app using copier.

Requires uv

https://docs.astral.sh/uv/getting-started/installation/

Source code in saritasa_invocations/django.py

@invoke.task
def startapp(context: invoke.Context) -> None:
    """Create new django app using copier.

    Requires uv:
        https://docs.astral.sh/uv/getting-started/installation/

    """
    config = _config.Config.from_context(context)
    if not config.django.app_boilerplate_link:
        raise invoke.Exit(
            code=1,
            message="Please, provide link to your django app boilerplate!",
        )
    context.run(
        "uvx copier copy "
        f"'{config.django.app_boilerplate_link}' "
        f"'{config.django.apps_path}' "
        f"--trust --vcs-ref=HEAD",
    )

wait_for_database(context)

Ensure that database is up and ready to accept connections.

Function called just once during subsequent calls of management commands.

Requires django_probes

https://github.com/painless-software/django-probes#basic-usage

Source code in saritasa_invocations/django.py

@invoke.task
def wait_for_database(context: invoke.Context) -> None:
    """Ensure that database is up and ready to accept connections.

    Function called just once during subsequent calls of management commands.

    Requires django_probes:
        https://github.com/painless-software/django-probes#basic-usage

    """
    config = _config.Config.from_context(context)
    if hasattr(wait_for_database, "_called"):
        return
    docker.up(context)
    # Not using manage to avoid infinite loop
    python.run(
        context,
        command=(
            f"{config.django.manage_file_path} wait_for_database --stable 0"
        ),
        env={
            "DJANGO_SETTINGS_MODULE": config.django.settings_path,
        },
    )
    wait_for_database._called = True  # type: ignore