Skip to content

Instantly share code, notes, and snippets.

@cuducos
Last active August 25, 2023 15:39
Show Gist options
  • Save cuducos/19e0e6a994c89628ccf470c1b5f62c1d to your computer and use it in GitHub Desktop.
Save cuducos/19e0e6a994c89628ccf470c1b5f62c1d to your computer and use it in GitHub Desktop.
Update Minha Receita in a Dokku server
"""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])()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment