<commit_before>import logging
import os
import psycopg2
import re
import sys
import time

is_py3 = sys.hexversion >= 0x03000000

if is_py3:
    from urllib.parse import urlparse
else:
    from urlparse import urlparse


logger = logging.getLogger(__name__)


def parseurl(url):
    r = urlparse(url)
    return {
        'hostname': r.hostname,
        'port': r.port or 5432,
        'username': r.username,
        'password': <PASSWORD>,
    }


class Postgresql:

    def __init__(self, config):
        self.name = config['name']
        self.host, self.port = config['listen'].split(':')
        self.libpq_parameters = {
            'host': self.host,
            'port': self.port,
            'fallback_application_name': 'Governor',
            'connect_timeout': 5,
            'options': '-c statement_timeout=2000'
        }
        self.data_dir = config['data_dir']
        self.replication = config['replication']
        self.superuser = config['superuser']
        self.admin = config['admin']
        self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
        self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir

        self.config = config

        self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
            connect_address=self.config['connect_address'], **self.replication)

        self.conn = None
        self.cursor_holder = None
        self.members = []  # list of already existing replication slots

    def cursor(self):
        if not self.cursor_holder:
            self.conn = psycopg2.connect(**self.libpq_parameters)
            self.conn.autocommit = True
            self.cursor_holder = self.conn.cursor()

        return self.cursor_holder

    def disconnect(self):
        try:
            self.conn.close()
        except:
            logger.exception('Error disconnecting')

    def query(self, sql, *params):
        max_attempts = 0
        while True:
            try:
                self.cursor().execute(sql, params)
                break
            except psycopg2.OperationalError as e:
                if self.conn:
                    self.disconnect()
                self.cursor_holder = None
                if max_attempts > 4:
                    raise e
                max_attempts += 1
                time.sleep(5)
        return self.cursor()

    def data_directory_empty(self):
        return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []

    def initialize(self):
        if os.system(self._pg_ctl + ' initdb') == 0:
            self.write_pg_hba()

            return True

        return False

    def sync_from_leader(self, leader):
        r = parseurl(leader.address)

        pgpass = '<PASSWORD>'
        with open(pgpass, 'w') as f:
            os.fchmod(f.fileno(), 0o600)
            f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r))

        try:
            os.environ['PGPASSFILE'] = pgpass
            return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
                data_dir=self.data_dir, **r)) == 0<commit_msg>Initialize cluster with encoding=UTF8 by default.<commit_after>import logging
import os
import psycopg2
import re
import sys
import time

is_py3 = sys.hexversion >= 0x03000000

if is_py3:
    from urllib.parse import urlparse
else:
    from urlparse import urlparse


logger = logging.getLogger(__name__)


def parseurl(url):
    r = urlparse(url)
    return {
        'hostname': r.hostname,
        'port': r.port or 5432,
        'username': r.username,
        'password': <PASSWORD>,
    }


class Postgresql:

    def __init__(self, config):
        self.name = config['name']
        self.host, self.port = config['listen'].split(':')
        self.libpq_parameters = {
            'host': self.host,
            'port': self.port,
            'fallback_application_name': 'Governor',
            'connect_timeout': 5,
            'options': '-c statement_timeout=2000'
        }
        self.data_dir = config['data_dir']
        self.replication = config['replication']
        self.superuser = config['superuser']
        self.admin = config['admin']
        self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
        self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir

        self.config = config

        self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
            connect_address=self.config['connect_address'], **self.replication)

        self.conn = None
        self.cursor_holder = None
        self.members = []  # list of already existing replication slots

    def cursor(self):
        if not self.cursor_holder:
            self.conn = psycopg2.connect(**self.libpq_parameters)
            self.conn.autocommit = True
            self.cursor_holder = self.conn.cursor()

        return self.cursor_holder

    def disconnect(self):
        try:
            self.conn.close()
        except:
            logger.exception('Error disconnecting')

    def query(self, sql, *params):
        max_attempts = 0
        while True:
            try:
                self.cursor().execute(sql, params)
                break
            except psycopg2.OperationalError as e:
                if self.conn:
                    self.disconnect()
                self.cursor_holder = None
                if max_attempts > 4:
                    raise e
                max_attempts += 1
                time.sleep(5)
        return self.cursor()

    def data_directory_empty(self):
        return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []

    def initialize(self):
        if os.system(self._pg_ctl + ' initdb -o --encoding=UTF8') == 0:
            self.write_pg_hba()

            return True

        return False

    def sync_from_leader(self, leader):
        r = parseurl(leader.address)

        pgpass = '<PASSWORD>'
        with open(pgpass, 'w') as f:
            os.fchmod(f.fileno(), 0o600)
            f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r))

        try:
            os.environ['PGPASSFILE'] = pgpass
            return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
                data_dir=self.data_dir, **r)) == 0
{'fraction_non_alphanumeric': 0.11896144767899292, 'fraction_numerical': 0.01070023603461841, 'mean_word_length': 1.9981132075471697, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3585805', 'n_tokens_mistral': 1966, 'n_tokens_neox': 1852, 'n_words': 400}