Python 子进程模块 | Postgres pg_dump 带密码
Python Subprocess Module | Postgres pg_dump with password
我有一个数据库，我想使用 python 代码进行备份。
我试图将我的代码基于 中使用子进程模块和 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: a bytes-like object is required, not 'str'
这可以在子流程中完成吗？如果是，怎么做？
On Microsoft Windows the file is named %APPDATA%\postgresql\pgpass.conf (where %APPDATA% refers to the Application Data subdirectory in the user's profile).
和-w或--no-passwordcommand line option（而不是-W）
-w
--no-password
Never issue a password prompt. If the server requires password authentication and a password is not available by other means such as a .pgpass file, the connection attempt will fail. This option can be useful in batch jobs and scripts where no user is present to enter a password.
最简单的是使用PGPASSWORD environment variable。
有两个class：
- 创建 dsn 字符串需要第一个 class。然后尝试连接
带dsn参数。如果无法连接，请转到第二个 class。
- 创建数据库和恢复所有表需要第二个 class
从文件。您需要重新制作此字符串：
正确打开您的数据库dump_file
__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))
另一种选择是使用 dbname 参数
'pg_dump --dbname=postgresql://{}:{}@{}:{}/{}'.format(user, password, host, port, database_name)
我有一个数据库，我想使用 python 代码进行备份。
我试图将我的代码基于 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: a bytes-like object is required, not 'str'
这可以在子流程中完成吗？如果是，怎么做？
On Microsoft Windows the file is named
%APPDATA%\postgresql\pgpass.conf(where%APPDATA%refers to the Application Data subdirectory in the user's profile).
和-w或--no-passwordcommand line option（而不是-W）
-w
--no-passwordNever issue a password prompt. If the server requires password authentication and a password is not available by other means such as a .pgpass file, the connection attempt will fail. This option can be useful in batch jobs and scripts where no user is present to enter a password.
最简单的是使用PGPASSWORD environment variable。
有两个class：
- 创建 dsn 字符串需要第一个 class。然后尝试连接 带dsn参数。如果无法连接，请转到第二个 class。
- 创建数据库和恢复所有表需要第二个 class 从文件。您需要重新制作此字符串：
正确打开您的数据库dump_file
__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))
另一种选择是使用 dbname 参数
'pg_dump --dbname=postgresql://{}:{}@{}:{}/{}'.format(user, password, host, port, database_name)