Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def add_hba_entry(self, *, type='host', database, user, address=None,
auth_method, auth_options=None):
"""Add a record to pg_hba.conf."""
status = self.get_status()
if status == 'not-initialized':
raise ClusterError(
'cannot modify HBA records: cluster is not initialized')
if type not in {'local', 'host', 'hostssl', 'hostnossl'}:
raise ValueError('invalid HBA record type: {!r}'.format(type))
pg_hba = os.path.join(self._data_dir, 'pg_hba.conf')
record = '{} {} {}'.format(type, database, user)
if type != 'local':
if address is None:
raise ValueError(
'{!r} entry requires a valid address'.format(type))
else:
record += ' {}'.format(address)
def reload(self):
"""Reload server configuration."""
status = self.get_status()
if status != 'running':
raise ClusterError('cannot reload: cluster is not running')
process = subprocess.run(
[self._pg_ctl, 'reload', '-D', self._data_dir],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stderr = process.stderr
if process.returncode != 0:
raise ClusterError(
'pg_ctl stop exited with status {:d}: {}'.format(
process.returncode, stderr.decode()))
def _find_pg_binary(self, binary):
bpath = platform_exe(os.path.join(self._pg_bin_dir, binary))
if not os.path.isfile(bpath):
raise ClusterError(
'could not find {} executable: '.format(binary) +
'{!r} does not exist or is not a file'.format(bpath))
return bpath
def reset_hba(self):
raise ClusterError('cannot modify HBA records of unmanaged cluster')
def reset_wal(self, *, oid=None, xid=None):
status = self.get_status()
if status == 'not-initialized':
raise ClusterError(
'cannot modify WAL status: cluster is not initialized')
if status == 'running':
raise ClusterError(
'cannot modify WAL status: cluster is running')
opts = []
if oid is not None:
opts.extend(['-o', str(oid)])
if xid is not None:
opts.extend(['-x', str(xid)])
if not opts:
return
opts.append(self._data_dir)
try:
reset_wal = self._find_pg_binary('pg_resetwal')
except ClusterError:
reset_wal = self._find_pg_binary('pg_resetxlog')
def _init_env(self):
if not self._pg_bin_dir:
pg_config = self._find_pg_config(self._pg_config_path)
pg_config_data = self._run_pg_config(pg_config)
self._pg_bin_dir = pg_config_data.get('bindir')
if not self._pg_bin_dir:
raise ClusterError(
'pg_config output did not provide the BINDIR value')
self._pg_ctl = self._find_pg_binary('pg_ctl')
self._postgres = self._find_pg_binary('postgres')
self._pg_version = self._get_pg_version()
def add_hba_entry(self, *, type='host', database, user, address=None,
auth_method, auth_options=None):
raise ClusterError('cannot modify HBA records of unmanaged cluster')