Update Minha Receita in a Dokku server

update-minha-receita.py

"""Usage: ./update-minha-receita.py YYYY-MM-DD"""

from datetime import datetime, timedelta
from functools import cached_property
from pathlib import Path
from shutil import rmtree
from subprocess import run
from sys import argv
from urllib.parse import urlparse


class Updater:
    def __init__(self, date_named_data_dir):
        self.dir = Path(date_named_data_dir)
        if not self.dir.exists():
            raise ValueError(f"Directory {self.dir} does not exist.")

        input = datetime.strptime(self.dir.name, "%Y-%m-%d")
        self.new_db = f"minhareceita{input.strftime('%Y%m')}"

        previous_month = input.replace(day=1) - timedelta(days=1)
        self.old_db = f"minhareceita{previous_month.strftime('%Y%m')}"

    @cached_property
    def new_db_uri(self):
        cmd = ("dokku", "postgres:create", self.new_db)
        result = run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise RuntimeError(
                f"Error creating database with {' '.join(cmd)}: {result.stderr}"
            )

        for line in result.stdout.splitlines():
            line = line.strip()
            if not line.lower().startswith("dsn:"):
                continue

            return line[4:].strip()

        raise RuntimeError(f"Could not find database URL in output: {result.stdout}")

    @cached_property
    def new_db_exposed_uri(self):
        if not self.new_db_uri:
            raise RuntimeError("Could not find database URL")

        cmd = ("dokku", "postgres:expose", self.new_db)
        result = run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise RuntimeError(
                f"Error exposing database with {' '.join(cmd)}: {result.stderr}"
            )

        *_, port = result.stdout.split("->")
        parsed = urlparse(self.new_db_uri)
        netloc = parsed.netloc.removesuffix(str(parsed.port)) + port.strip()
        netloc = netloc.replace(parsed.hostname, "localhost")
        return parsed._replace(netloc=netloc).geturl()

    def __call__(self):
        etl = (
            "minha-receita",
            "transform",
            "-c",
            "-d",
            self.dir,
            "-u",
            self.new_db_exposed_uri,
        )
        unexpose = ("dokku", "postgres:unexpose", self.new_db)
        set_new_db = (
            "dokku",
            "config:set",
            "minhareceita",
            f"DATABASE_URL={self.new_db_uri}",
        )
        unlink = ("dokku", "postgres:unlink", self.old_db, "minhareceita")
        destroy = ("dokku", "postgres:destroy", self.old_db)
        cmds = (etl, unexpose, set_new_db, unlink, destroy)
        for cmd in cmds:
            if run(cmd).returncode != 0:
                raise RuntimeError(f"Error running {' '.join(cmd)}")
        rmtree(self.dir)


if __name__ == "__main__":
    Updater(argv[1])()