86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""Daily PostgreSQL backup with 7-day retention.
|
|
|
|
Cron: 0 3 * * * cd /opt/voidea && python tools/backup_db.py
|
|
"""
|
|
|
|
import datetime
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
BACKUP_DIR = os.environ.get("BACKUP_DIR", "/opt/voidea/backups")
|
|
RETENTION_DAYS = int(os.environ.get("BACKUP_RETENTION_DAYS", "7"))
|
|
DB_URL = os.environ.get("DATABASE_URL", "")
|
|
|
|
|
|
def parse_db_url(url: str) -> dict:
|
|
"""Parse DATABASE_URL into pg_dump arguments."""
|
|
if not url:
|
|
print("ERROR: DATABASE_URL not set", file=sys.stderr)
|
|
sys.exit(1)
|
|
parts = url.replace("postgresql://", "").replace("postgres://", "").split("@")
|
|
user_pass = parts[0].split(":")
|
|
host_db = parts[1].split("/")
|
|
host_port = host_db[0].split(":")
|
|
return {
|
|
"user": user_pass[0],
|
|
"password": user_pass[1] if len(user_pass) > 1 else "",
|
|
"host": host_port[0],
|
|
"port": host_port[1] if len(host_port) > 1 else "5432",
|
|
"dbname": host_db[1].split("?")[0],
|
|
}
|
|
|
|
|
|
def main():
|
|
os.makedirs(BACKUP_DIR, exist_ok=True)
|
|
config = parse_db_url(DB_URL)
|
|
|
|
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = f"voidea_backup_{timestamp}.sql.gz"
|
|
filepath = os.path.join(BACKUP_DIR, filename)
|
|
|
|
env = os.environ.copy()
|
|
env["PGPASSWORD"] = config["password"]
|
|
|
|
cmd = [
|
|
"pg_dump",
|
|
"-h", config["host"],
|
|
"-p", config["port"],
|
|
"-U", config["user"],
|
|
"-d", config["dbname"],
|
|
"--clean",
|
|
"--if-exists",
|
|
"--no-owner",
|
|
]
|
|
|
|
print(f"Backing up to {filepath} ...")
|
|
with open(filepath, "wb") as f:
|
|
proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
gzip = subprocess.Popen(["gzip"], stdin=proc.stdout, stdout=f, stderr=subprocess.PIPE)
|
|
_, stderr = gzip.communicate()
|
|
|
|
if proc.returncode != 0 or gzip.returncode != 0:
|
|
print(f"ERROR: Backup failed: {stderr.decode()}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
file_size = os.path.getsize(filepath)
|
|
print(f"Backup complete: {filepath} ({file_size / 1024 / 1024:.1f} MB)")
|
|
|
|
# Cleanup old backups
|
|
cutoff = datetime.datetime.now() - datetime.timedelta(days=RETENTION_DAYS)
|
|
removed = 0
|
|
for fname in os.listdir(BACKUP_DIR):
|
|
fpath = os.path.join(BACKUP_DIR, fname)
|
|
if not os.path.isfile(fpath):
|
|
continue
|
|
mtime = datetime.datetime.fromtimestamp(os.path.getmtime(fpath))
|
|
if mtime < cutoff:
|
|
os.remove(fpath)
|
|
removed += 1
|
|
|
|
print(f"Cleaned up {removed} old backup(s) (retention: {RETENTION_DAYS} days)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|