PostgreSQL pg_dump backups via Python

The following scripts backup your PostgreSQL databases via Python using pg_dump. The scripts will then compress the backup via zip. These backup scripts are linux specific.

You’ll have to manage secrets either using a .pgpass file or environment variables. This is the first pass before using any type of password security.

import subprocess
import psycopg2
from datetime import datetime
import os

# PostgreSQL connection parameters
db_params = {
    "dbname": "postgres",  # change to your database name if different
    "user": "your_username",
    "password": "your_password",
    "host": "localhost"
}

# Directory to store the backups
backup_dir = "/mnt/backups/"
os.makedirs(backup_dir, exist_ok=True)

def get_databases(connection_parameters):
    """Retrieve a list of database names."""
    conn = None
    databases = []
    try:
        conn = psycopg2.connect(**connection_parameters)
        cur = conn.cursor()
        cur.execute("SELECT datname FROM pg_database WHERE datistemplate = false;")
        databases = [db[0] for db in cur.fetchall()]
        cur.close()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        if conn is not closed:
            conn.close()
    return databases

def backup_database(db_name, backup_directory):
    """Backup a single database and zip the backup file."""
    current_date = datetime.now().strftime('%Y%m%d')
    backup_file = os.path.join(backup_directory, f"{db_name}.sql")
    zip_file = f"{current_date}-{db_name}.zip"
    
    # Backup database to SQL file
    subprocess.run(f"pg_dump -U {db_params['user']} -w {db_name} > {backup_file}", shell=True, check=True)
    
    # Compress the SQL file
    subprocess.run(f"zip -j {backup_dir}{zip_file} {backup_file}", shell=True, check=True)
    
    # Remove the original SQL file
    os.remove(backup_file)

if __name__ == "__main__":
    databases = get_databases(db_params)
    for db_name in databases:
        print(f"Backing up database: {db_name}")
        backup_database(db_name, backup_dir)
    print("Backup process completed.")

Alternative version that I’ve used before

import os
import datetime
import subprocess

# Database connection details

HOST = "localhost"
PORT = 5432
USER = "postgres"
PASSWORD = "your_password"  # Replace with your actual password

# Backup directory

BACKUP_DIR = "/mnt/backups/"

# Get current date in YYYYMMDD format

current_date = datetime.datetime.now().strftime("%Y%m%d")


def main():
    # Connect to PostgreSQL server

    try:
        conn = psycopg2.connect(host=HOST, port=PORT, user=USER, password=PASSWORD)
    except psycopg2.Error as e:
        print("Error connecting to database:", e)
        return
    # Get all databases except system ones

    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT datname FROM pg_database WHERE datistemplate = false AND datname NOT IN ('postgres', 'template')"
        )
        databases = [row[0] for row in cur.fetchall()]
    except psycopg2.Error as e:
        print("Error fetching databases:", e)
        return
    # Backup each database

    for db in databases:
        backup_database(db)
    # Close connection

    conn.close()


def backup_database(db_name):
    # Dump database

    dump_file = os.path.join(BACKUP_DIR, f"{db_name}.sql")
    command = [
        "pg_dump",
        "-h",
        HOST,
        "-p",
        str(PORT),
        "-U",
        USER,
        "-d",
        db_name,
        ">",
        dump_file,
    ]
    subprocess.run(command, check=True)

    # Zip the dump

    zip_file = os.path.join(BACKUP_DIR, f"{current_date}-{db_name}.zip")
    subprocess.run(["zip", "-r", zip_file, dump_file], check=True)

    # Remove the uncompressed dump

    os.remove(dump_file)


if __name__ == "__main__":
    main()