Использование Python скриптов для обслуживания баз PostgreSQL

from datetime import datetime
import logging
import socket
import urllib.request

import pg_handler
import file_handler

logging.basicConfig(
    filename='pg_backup.log',
    level=logging.INFO,
    format='%(asctime)s:%(levelname)s:%(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)


def health_check(hc_key: str):
    """Отправляет запрос к https://healthchecks.io/ на указанную точку мониторинга

    Args:
        hc_key (str): Ключ точки мониторинга
    """
    try:
        urllib.request.urlopen(f'https://hc-ping.com/{hc_key}', timeout=10)
    except socket.error as e:
        logging.warning(f'Ping failed: {e}')


def main():

    config = file_handler.load_config('pg_backup_config.json')
    if config is None:
        return

    # Заглушка для корректной работы finally
    filename = 'tmp.backup'

    try:
        dbnames_src = pg_handler.get_database_list(
            config.host_src, config.username_src, config.password_src)

        for dbname in config.databases:
            if dbname not in dbnames_src:
                logging.warning(f'Not database {dbname} in {config.host_src}')
                continue

            # Вычисляю имя файла от даты и папок назначения
            now = datetime.now()
            if now.day == 1:
                subfolder = 'month'
                save_period = '4M'
            elif now.isoweekday() == 1:
                subfolder = 'week'
                save_period = '1M'
            else:
                subfolder = 'day'
                save_period = '10d'

            filename = f'{dbname}_{now.strftime("%Y-%m-%d_%H-%M-%S")}.backup'
            local_folder_full = '/'.join([config.local_folder,
                                         dbname, subfolder])
            dist_folder_full = '/'.join([config.dist_folder,
                                        dbname, subfolder])

            file_handler.create_dir(local_folder_full)
            file_handler.create_dir(dist_folder_full)
            file_handler.remove_file('/'.join([local_folder_full, filename]))

            pg_handler.backup_database(config.host_src, config.username_src,
                                       config.password_src, dbname, '/'.join([local_folder_full, filename]))

            file_handler.copy_files(
                '/'.join([local_folder_full, filename]), dist_folder_full)
            file_handler.delete_old_files(dist_folder_full, save_period)
            file_handler.delete_old_files(local_folder_full, '2d')

        health_check(config.hc_key)

    except Exception as e:
        logging.warning(e)
        return
    finally:
        file_handler.remove_file(filename)


if __name__ == '__main__':
    main()

{
  "local_folder": "/backup/postgres",
  "dist_folder": "BackupFTP:/backup_1c/postgres",
  "host_src": "10.10.1.181",
  "username_src": "postgres",
  "password_src": "",
  "hc_key": "eb095278-f28d-448d-87fb-7b75c171a6aa",
  "databases": [
    "rt_test_1"
  ]
}

import argparse
import logging

import pg_handler
import file_handler

logging.basicConfig(
    filename='pg_copy_prod_to_test.log',
    level=logging.INFO,
    format='%(asctime)s:%(levelname)s:%(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)


def main(src_base, dst_base):

    config = file_handler.load_config('pg_copy_prod_to_test_config.json')
    if config is None:
        return

    filename = 'tmp.backup'

    try:
        dbnames_src = pg_handler.get_database_list(
            config.host_src, config.username_src, config.password_src)

        if src_base not in dbnames_src:
            logging.warning(f'Not database {src_base} in {config.host_src}')
            return

        file_handler.remove_file(filename)

        pg_handler.backup_database(config.host_src, config.username_src,
                                   config.password_src, src_base, filename)
        dbnames_dst = pg_handler.get_database_list(
            config.host_dst, config.username_dst, config.password_dst)

        if dst_base in dbnames_dst:
            pg_handler.drop_database(config.host_dst, config.username_dst,
                                     config.password_dst, dst_base)

        pg_handler.create_database(config.host_dst, config.username_dst,
                                   config.password_dst, dst_base)
        pg_handler.restore_database(config.host_dst, config.username_dst,
                                    config.password_dst, dst_base, filename)
        file_handler.remove_file(filename)

    except Exception as e:
        logging.warning(e)
        return
    finally:
        file_handler.remove_file(filename)


if __name__ == '__main__':

    parser = argparse.ArgumentParser(description='pg_copy')
    parser.add_argument('--src-base', dest='src_base', required=True, type=str)
    parser.add_argument('--dst-base', dest='dst_base', required=True, type=str)

    args = parser.parse_args()
    main(args.src_base, args.dst_base)

{
  "host_src": "pg1c-02",
  "host_dst": "10.10.1.181",
  "username_src": "postgres",
  "password_src": "",
  "username_dst": "postgres",
  "password_dst": ""
}

import logging
import pg_handler
import file_handler

logging.basicConfig(
    filename='pg_copy_base.log',
    level=logging.DEBUG,
    format='%(asctime)s:%(levelname)s:%(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)


def main():
    config = file_handler.load_config('pg_copy_base_config.json')
    if config is None:
        return

    filename = 'tmp.backup'

    try:
        dbnames_src = pg_handler.get_database_list(
            config.host_src, config.username_src, config.password_src)

        for dbname in config.move_databases:

            if dbname not in dbnames_src:
                logging.warning(f'Not database {dbname} in {config.host_src}')
                continue

            filename = f'{dbname}.backup'
            file_handler.remove_file(filename)

            pg_handler.backup_database(config.host_src, config.username_src,
                                       config.password_src, dbname, filename)
            dbnames_dst = pg_handler.get_database_list(
                config.host_dst, config.username_dst, config.password_dst)
            if dbname in dbnames_dst:
                pg_handler.drop_database(config.host_dst, config.username_dst,
                                         config.password_dst, dbname)

            pg_handler.create_database(config.host_dst, config.username_dst,
                                       config.password_dst, dbname)
            pg_handler.restore_database(config.host_dst, config.username_dst,
                                        config.password_dst, dbname, filename)
            file_handler.remove_file(filename)
            return
    except Exception as e:
        logging.warning(e)
        return
    finally:
        file_handler.remove_file(filename)


if __name__ == '__main__':
    main()

{
  "host_src": "pg1c-02",
  "host_dst": "pg-prod-14",
  "username_src": "postgres",
  "password_src": "",
  "username_dst": "postgres",
  "password_dst": "",
  "move_databases": [
    "rt"
  ]
}

import subprocess
import logging
import os

DEFAULT_PG_PORT = '5432'


def get_database_list(host: str, username: str, password: str, port: str = DEFAULT_PG_PORT) -> list:
    """Получить список базданных сервера

    Args:
        host (str): Адрес сервера
        username (str): Имя пользователя
        password (str): Пароль
        port (str): Порт сервера

    Raises:
        Exception: Ошибки при подключении к PG

    Returns:
        list: Список баз данных
    """
    sql_command_select_databases = 'SELECT datname FROM pg_database;'
    result = subprocess.run(
        [
            'psql',
            '-h', host,
            '-p', port,
            '-U', username,
            '--tuples-only',
            '-c', sql_command_select_databases
        ],
        encoding='UTF-8',
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )

    if result.returncode != 0:
        raise Exception(f'Get DB list: {host} {result.stderr}')
    else:
        dbnames = result.stdout.split('\n')
        dbnames = list(map(lambda dbname: dbname.strip(), dbnames))
        dbnames = list(dbname for dbname in dbnames if dbname != '')

        logging.info(f'{host} dbnames: {",".join(dbnames)}')
        return dbnames


def backup_database(host: str, username: str, password: str, dbname: str, filename: str, port: str = DEFAULT_PG_PORT):
    """Создать резервную копию базы данных в custom формате с включением больших объектов

    Args:
        host (str): Адрес сервера
        username (str): Имя пользователя
        password (str): Пароль
        dbname (str): Имя базы данных
        filename (str): Имя файла приемника
        port (str): Порт сервера

    Raises:
        Exception: Ошибки при подключении к PG
    """
    logging.info(f'backup {dbname} from {host}')
    result = subprocess.run([
        'pg_dump', dbname,
        '-h', host,
        '-p', port,
        '-U', username,
        '--format=c', '--blobs',
        '--file', filename
    ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )

    if result.returncode != 0:
        raise Exception(f'Backup DB + {result.stderr}')
    else:
        file_stats = os.stat(filename)
        logging.info(
            f'Backup DB: end {dbname} size {round(file_stats.st_size / (1024 * 1024))}MB')


def drop_database(host: str, username: str, password: str, dbname: str, port: str = DEFAULT_PG_PORT):
    """Удаляет базу данных с предвартельным отключением соединений

    Args:
        host (str): Адрес сервера
        username (str): Имя пользователя
        password (str): Пароль
        dbname (str): Имя базы данных
        port (str): Порт сервера

    Raises:
        Exception: Ошибки при подключении к PG
    """
    logging.info(f'Drop DB: {dbname} to {host}')
    sql_command_drop_database = f" \
        UPDATE pg_database SET datallowconn = 'false' WHERE datname = '{dbname}'; \
        SELECT pg_terminate_backend(pg_stat_activity.pid) \
        FROM pg_stat_activity \
        WHERE pg_stat_activity.datname = '{dbname}' AND pid <> pg_backend_pid();"
    result = subprocess.run(
        ['psql',
         '-h', host,
         '-p', port,
         '-U', username,
         '-c', sql_command_drop_database,
         ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )

    if result.returncode != 0:
        raise Exception(f'Drop DB: {result.stderr}')

    result = subprocess.run(
        ['psql',
         '-h', host,
         '-p', port,
         '-U', username,
         '-c', f"DROP DATABASE {dbname};",
         ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )
    if result.returncode != 0:
        raise Exception(f'Drop DB: {result.stderr}')


def create_database(host: str, username: str, password: str, dbname: str, port: str = DEFAULT_PG_PORT):
    """Создает базу данных

    Args:
        host (str): Адрес сервера
        username (str): Имя пользователя
        password (str): Пароль
        dbname (str): Имя базы данных
        port (str): Порт сервера

    Raises:
        Exception: Ошибки при подключении к PG
    """
    logging.info(f'Create DB: {dbname} to {host}')
    sql_command_create_database = f'CREATE DATABASE {dbname}'
    result = subprocess.run(
        ['psql',
         '-h', host,
         '-p', port,
         '-U', username,
         '-d', 'postgres',
         '-c', sql_command_create_database,
         ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )

    if result.returncode != 0:
        raise Exception(f'Create DB: {result.stderr}')


def restore_database(host: str, username: str, password: str, dbname: str, filename: str, port: str = DEFAULT_PG_PORT):
    """Восстанавливает БД из файла источника

    Args:
        host (str): Адрес сервера
        username (str): Имя пользователя
        password (str): Пароль
        dbname (str): Имя базы данных
        filename (str): Имя файла источника
        port (str): Порт сервера

    Raises:
        Exception: Ошибки при подключении к PG
    """
    logging.info(f'Restore DB: {dbname} to {host}')

    result = subprocess.run([
        'pg_restore', filename,
        '-h', host,
        '-p', port,
        '-U', username,
        '--dbname', dbname,
    ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        env={'PGPASSWORD': password},
    )

    if result.returncode != 0:
        raise Exception(f'Restore DB: {result.stderr}')
    else:
        logging.info(f'Restore DB: {dbname} end')

import subprocess
import logging
import os
import json
from types import SimpleNamespace


def load_config(filename: str) -> object:
    """Получает настройки из файла JSON

    Args:
        filename (str): Имя файла конфигурации в формате JSON

    Returns:
        object: объект настроек
    """
    try:
        with open(filename, 'r') as f:
            return json.load(f, object_hook=lambda d: SimpleNamespace(**d))
    except FileNotFoundError as e:
        logging.warning(f'Load config file: {e.strerror}')
        return
    except json.decoder.JSONDecodeError as e:
        logging.warning(f'Parsing config file: {e.msg}')
        return


def remove_file(filename: str):
    """Удаляет файл если он присутствует

    Args:
        filename (str): Имя файла
    """
    if os.path.exists(filename):
        logging.info(f'Remove file: {filename}')
        os.remove(filename)


def copy_files(folder_src: str, folder_dst: str):
    """Копирует файл(ы) с использованием rclone

    Args:
        folder_src (str): Папка (файл) источник
        folder_dst (str): Папка (файл) приемник

    Raises:
        Exception: Ошибки работы rclone
    """
    logging.info(f'Copy files: {folder_src} to {folder_dst}')
    # rclone copy /backup_1c/postgres BackupFTP:/backup_1c/postgres/$dbname/$subf
    result = subprocess.run([
        'rclone', 'copy',
        folder_src,
        folder_dst,
    ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception(f'Copy files: {result.stderr}')


def create_dir(folder_dst: str):
    """Создает папку с помощью rclone

    Args:
        folder_dst (str): Имя папки

    Raises:
        Exception: Ошибки работы rclone
    """
    logging.info(f'Create dir: {folder_dst}')
    # rclone mkdir BackupFTP:/backup_1c/postgres/$dbname/$subf
    result = subprocess.run([
        'rclone', 'mkdir',
        folder_dst,
    ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception(f'Create dir: {result.stderr}')


def delete_old_files(folder: str, save_period: str):
    """_summary_

    Args:
        folder (str): Имя папки
        save_period (str): Возраст файлов старше которого они удаляются https://rclone.org/filtering/#min-age-don-t-transfer-any-file-younger-than-this

    Raises:
        Exception: Ошибки работы rclone
    """
    logging.info(f'Delete old files: {folder} to {save_period}')
    # rclone delete --min-age $savePeriod BackupFTP:/backup_1c/postgres/$dbname/$subf
    result = subprocess.run([
        'rclone', 'delete',
        '--min-age', save_period,
        folder
    ],
        encoding='UTF-8',
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception(f'Delete old files: {result.stderr}')