Backup Script in PG

The following script is for backing up a postgres database

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/backup/"
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.")

Backup Via .env file

The following script uses a .env instead of a .pgpass file

import os
import psycopg2
from datetime import datetime
import subprocess
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Database connection parameters
DB_HOST = os.getenv('DB_HOST')
DB_PORT = os.getenv('DB_PORT')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')

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

def get_databases():
    """Retrieve a list of database names."""
    conn = None
    databases = []
    try:
        conn = psycopg2.connect(
            dbname='postgres',  # Adjust if you use a different db for listing databases
            user=DB_USER,
            password=DB_PASSWORD,
            host=DB_HOST,
            port=DB_PORT
        )
        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 None:
            conn.close()
    return databases

def backup_database(db_name):
    """Backup a single database and zip the backup file."""
    current_date = datetime.now().strftime('%Y%m%d')
    backup_file = os.path.join(backup_dir, f"{db_name}.sql")
    zip_file = f"{current_date}-{db_name}.zip"
    
    # Dump the database
    dump_command = f"pg_dump -h {DB_HOST} -p {DB_PORT} -U {DB_USER} -w {db_name} > {backup_file}"
    os.environ['PGPASSWORD'] = DB_PASSWORD
    subprocess.run(dump_command, shell=True, check=True)
    
    # Compress the SQL file
    zip_command = f"zip -j {backup_dir}/{zip_file} {backup_file}"
    subprocess.run(zip_command, shell=True, check=True)
    
    # Remove the original SQL file
    os.remove(backup_file)

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