selfprivacy-rest-api/selfprivacy_api/jobs/nix_collect_garbage.py

from time import sleep
import re
import subprocess

from selfprivacy_api.jobs import Job, JobStatus, Jobs
from selfprivacy_api.utils.huey import huey


@huey.task()
def nix_collect_garbage(job: Job):

    Jobs.update(
        job=job,
        status=JobStatus.RUNNING,
        progress=0,
        status_text="Сalculate the number of dead packages...",
    )

    output = subprocess.check_output(
        ["nix-store --gc --print-dead", "--gc", "--print-dead"]
    )

    dead_packages = len(re.findall("/nix/store/", output.decode("utf-8")))
    package_equal_to_percent = 100 / dead_packages

    Jobs.update(
        job=job,
        status=JobStatus.RUNNING,
        progress=0,
        status_text=f"Found {dead_packages} packages to remove!",
    )

    def _parse_line(line):
        pattern = re.compile(r"[+-]?\d+\.\d+ \w+ freed")
        match = re.search(
            pattern,
            line,
        )

        if match is None:
            Jobs.update(
                job=job,
                status=JobStatus.FINISHED,
                progress=100,
                status_text="Completed with an error",
                result="We are sorry, result was not found :(",
            )

        else:
            Jobs.update(
                job=job,
                status=JobStatus.FINISHED,
                progress=100,
                status_text="Сleaning completed.",
                result=f"{match.group(0)} have been cleared",
            )

    def _stream_process(process):
        go = process.poll() is None
        percent = 0

        for line in process.stdout:
            if "deleting '/nix/store/" in line:
                percent += package_equal_to_percent

                Jobs.update(
                    job=job,
                    status=JobStatus.RUNNING,
                    progress=int(percent),
                    status_text="Сleaning...",
                )

            elif "store paths deleted," in line:
                _parse_line(line)

        return go

    process = subprocess.Popen(
        ["nix-collect-garbage", "-d"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    )
    _stream_process(process)