Python 子进程模块 | Postgres pg_dump 带密码答案

【问题标题】：Python Subprocess Module | Postgres pg_dump with passwordPython 子进程模块 | Postgres pg_dump 带密码
【发布时间】：2019-06-26 16:45:13
【问题描述】：

我有一个数据库，我想用我的 python 代码备份。

我试图以this discussion 中使用子进程模块和pg_dump 的代码为基础。我现在的问题是我必须手动输入密码才能获取备份文件。我在某处读到了.pgpass，但我确实想看看是否可以在子流程模块中进行。

我的代码如下：

from subprocess import Popen, PIPE
from pathlib import Path, PureWindowsPath

def backup():
    version = 11
    postgresDir = Path("C:/Program Files/PostgreSQL/{}/bin/".format(version))
    directory = PureWindowsPath(postgresDir)
    filename = 'myBackUp2'  # output filename here
    saveDir = Path("D:/Desktop/{}.tar".format(filename))  # output directory here
    file = PureWindowsPath(saveDir)

    host = 'localhost'
    user = 'postgres'
    port = '5434'
    dbname = 'database_name'  # database name here
    proc = Popen(['pg_dump', '-h', host, '-U', user, '-W', '-p', port,
                   '-F', 't', '-f', str(file), '-d', dbname],
                    cwd=directory, shell=True, stdin=PIPE)
    proc.wait()

backup()

上面的代码有效并且备份被创建是我输入密码。我尝试用下面的代码替换 proc.wait() 以消除手动输入密码的需要：

return proc.communicate('{}\n'.format(database_password))

但我会收到此错误：

TypeError: 需要一个类似字节的对象，而不是 'str'

这可以在子流程中进行吗？如果有，怎么做？

【问题讨论】：

在Popen()调用中添加text=True参数。
或者您可以以字节而不是字符串的形式传递密码。

标签： python python-3.x postgresql subprocess pg-dump

【解决方案1】：

另一种选择是使用dbname 参数

'pg_dump --dbname=postgresql://{}:{}@{}:{}/{}'.format(user, password, host, port, database_name)

【讨论】：

【解决方案2】：

有两个类：

创建 dsn 字符串所需的第一类。然后尝试连接带 dsn 参数。如果无法连接，请转到二等舱。
创建数据库和恢复所有表所需的第二类从文件。您需要重新制作这些字符串：

为了正确打开你的数据库转储文件

__folder_name = Path(__file__).parent.parent
__folder_name_data = os.path.join(__folder_name, 'data')
__file_to_open = os.path.join(__folder_name_data, 'bd.backup')

import os
import textwrap
from pathlib import Path
from subprocess import Popen, PIPE


class DataBaseAPI:
    __slots__ = ('__dsn', 'cur')

    def __init__(self):
        self.__dsn = self.__dsn_string()
        self.cur = self.__connection()

    @staticmethod
    def __dsn_string() -> dict:
        print(f'INPUT name of DataBase')
        name = input()
        print(f'INPUT password of DataBase')
        password = input()
        print(f'INPUT user_name of DataBase or press ENTER if user_name="postgres"')
        user_name = input()
        if len(user_name) == 0:
            user_name = 'postgres'
        print(f'INPUT host_name of DataBase or press ENTER if host_name="localhost"')
        host_name = input()
        if len(host_name) == 0:
            host_name = 'localhost'
        return {'dbname': name, 'user': user_name, 'password': password, 'host': host_name}

    def __connection(self):
        try:
            conn = psycopg2.connect(dbname=self.__dsn['dbname'], user=self.__dsn['user'],
                                    host=self.__dsn['host'], password=self.__dsn['password'], port=5432)
        except psycopg2.OperationalError:
            print(textwrap.fill(f'There is no existing DataBase. Creating new DataBase', 80,
                                   subsequent_indent='                   '))
            DataBaseCreator(self.__dsn)
            conn = psycopg2.connect(dbname=self.__dsn['dbname'], user=self.__dsn['user'],
                                    host=self.__dsn['host'], password=self.__dsn['password'], port=5432)
        finally:
            conn.autocommit = True
            cur = conn.cursor()
            print(f'DataBase connection complete')
            return cur



class DataBaseCreator:
    def __init__(self, dsn):
        self.__dsn = dsn
        self.__check_conf_file()
        self.__create_data_base()
        self.__restore_data_base()

    def __check_conf_file(self):
        __app_data = os.environ.copy()["APPDATA"]
        __postgres_path = Path(f'{__app_data}\postgresql')
        __pgpass_file = Path(f'{__postgres_path}\pgpass.conf')
        parameters = f'{self.__dsn["host"]}:{5432}:{self.__dsn["dbname"]}:' \
                     f'{self.__dsn["user"]}:{int(self.__dsn["password"])}\n'
        if not os.path.isdir(__postgres_path):
            os.makedirs(__postgres_path)
        if os.path.isfile(__pgpass_file):
            log.debug(f'File "pgpass.conf" already exists')
            with open(__pgpass_file, 'r+') as f:
                content = f.readlines()
                if parameters not in content:
                    # сервер: порт:база_данных: имя_пользователя:пароль
                    f.write(parameters)
                else:
                    log.info(f' {parameters} already in "pgpass.conf" file')
        else:
            log.debug(f'File "pgpass.conf" not exists')
            with open(__pgpass_file, 'x') as f:
                # сервер: порт:база_данных: имя_пользователя:пароль
                f.write(parameters)

    def __create_data_base(self):
        try:
            __conn = psycopg2.connect(dbname='postgres', user=self.__dsn['user'],
                                      host=self.__dsn['host'], password=self.__dsn['password'], port=5432)
        except Exception as _:
            log.exception(f'{_}')
        else:
            __conn.autocommit = True
            __cur = __conn.cursor()
            __query = f'CREATE DATABASE "{self.__dsn["dbname"]}"'
            __cur.execute(__query)
            log.info(f'{__query}')

    def __restore_data_base(self):
        __col = [x for x in self.__dsn.values()]
        __folder_name = Path(__file__).parent.parent
        __folder_name_data = os.path.join(__folder_name, 'data')
        __file_to_open = os.path.join(__folder_name_data, 'bd.backup')
        __cmd = f'pg_restore --host={__col[3]} --dbname={__col[0]} --username={__col[1]} ' \
                f'--verbose=True --no-password ' \
                f'{__file_to_open}'
        try:
            __proc = Popen(__cmd, stdout=PIPE, stderr=PIPE)
        except FileNotFoundError:
            log.info(f'FileNotFoundError: [WinError 2] Не удается найти указанный файл')
            log.info(textwrap.fill(f'You need to SET Windows $PATH for use "pg_restore" in cmd', 80,
                                   subsequent_indent='                   '))
        else:
            __stderr = __proc.communicate()[1].decode('utf-8', errors="ignore").strip()
            log.debug(textwrap.fill(f'{__stderr}', 80))

【讨论】：

【解决方案3】：

最简单的方法是使用PGPASSWORD environment variable。

【讨论】：

【解决方案4】：

使用a password file.

在 Microsoft Windows 上，该文件名为 %APPDATA%\postgresql\pgpass.conf（其中 %APPDATA% 指的是用户配置文件中的应用程序数据子目录）。

还有-w 或--no-password command line option（而不是-W）

-w

--no-password

永远不要发出密码提示。如果服务器需要密码验证并且密码不能通过其他方式（例如 .pgpass 文件）获得，则连接尝试将失败。此选项在没有用户输入密码的批处理作业和脚本中很有用。

【讨论】：