Refactor directory structure in preparation for building pypi-package

patroni.py

@@ -1,123 +1,5 @@
 #!/usr/bin/env python
-import logging
+from patroni import main
-import os
-import sys
-import time
-import yaml
-
-from helpers.api import RestApiServer
-from helpers.etcd import Etcd
-from helpers.ha import Ha
-from helpers.postgresql import Postgresql
-from helpers.utils import setup_signal_handlers, sleep, reap_children
-from helpers.zookeeper import ZooKeeper
-
-logger = logging.getLogger(__name__)
-
-
-class Patroni:
-
-    def __init__(self, config):
-        self.nap_time = config['loop_wait']
-        self.postgresql = Postgresql(config['postgresql'])
-        self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
-        host, port = config['restapi']['listen'].split(':')
-        self.api = RestApiServer(self, config['restapi'])
-        self.next_run = time.time()
-        self.shutdown_member_ttl = 300
-
-    @staticmethod
-    def get_dcs(name, config):
-        if 'etcd' in config:
-            return Etcd(name, config['etcd'])
-        if 'zookeeper' in config:
-            return ZooKeeper(name, config['zookeeper'])
-        raise Exception('Can not find sutable configuration of distributed configuration store')
-
-    def touch_member(self, ttl=None):
-        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
-        if self.ha.cluster:
-            for m in self.ha.cluster.members:
-                # Do not update member TTL when it is far from being expired
-                if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
-                    return True
-        return self.ha.dcs.touch_member(connection_string, ttl)
-
-    def initialize(self):
-        # wait for etcd to be available
-        while not self.touch_member():
-            logger.info('waiting on DCS')
-            sleep(5)
-
-        # is data directory empty?
-        if self.postgresql.data_directory_empty():
-            # racing to initialize
-            if self.ha.dcs.race('/initialize'):
-                self.postgresql.initialize()
-                self.ha.dcs.take_leader()
-                self.postgresql.start()
-                self.postgresql.create_replication_user()
-                self.postgresql.create_connection_users()
-            else:
-                while True:
-                    leader = self.ha.dcs.current_leader()
-                    if leader and self.postgresql.sync_from_leader(leader):
-                        self.postgresql.write_recovery_conf(leader)
-                        self.postgresql.start()
-                        break
-                    sleep(5)
-        elif self.postgresql.is_running():
-            self.postgresql.load_replication_slots()
-
-    def schedule_next_run(self):
-        self.next_run += self.nap_time
-        current_time = time.time()
-        nap_time = self.next_run - current_time
-        if nap_time <= 0:
-            self.next_run = current_time
-        else:
-            self.ha.dcs.sleep(nap_time)
-
-    def run(self):
-        self.api.start()
-        self.next_run = time.time()
-
-        while True:
-            self.touch_member()
-            logger.info(self.ha.run_cycle())
-            try:
-                if self.ha.state_handler.is_leader():
-                    self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
-                else:
-                    self.ha.state_handler.drop_replication_slots()
-            except:
-                logger.exception('Exception when changing replication slots')
-            reap_children()
-            self.schedule_next_run()
-
-
-def main():
-    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
-    logging.getLogger('requests').setLevel(logging.WARNING)
-    setup_signal_handlers()
-
-    if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
-        print('Usage: {} config.yml'.format(sys.argv[0]))
-        return
-
-    with open(sys.argv[1], 'r') as f:
-        config = yaml.load(f)
-
-    patroni = Patroni(config)
-    try:
-        patroni.initialize()
-        patroni.run()
-    except KeyboardInterrupt:
-        pass
-    finally:
-        patroni.touch_member(patroni.shutdown_member_ttl)  # schedule member removal
-        patroni.postgresql.stop()
-        patroni.ha.dcs.delete_leader()
 
 
 if __name__ == '__main__':

patroni/__init__.py

@@ -0,0 +1,119 @@
+import logging
+import os
+import sys
+import time
+import yaml
+
+from patroni.api import RestApiServer
+from patroni.etcd import Etcd
+from patroni.ha import Ha
+from patroni.postgresql import Postgresql
+from patroni.utils import setup_signal_handlers, sleep, reap_children
+from patroni.zookeeper import ZooKeeper
+
+logger = logging.getLogger(__name__)
+
+
+class Patroni:
+
+    def __init__(self, config):
+        self.nap_time = config['loop_wait']
+        self.postgresql = Postgresql(config['postgresql'])
+        self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
+        host, port = config['restapi']['listen'].split(':')
+        self.api = RestApiServer(self, config['restapi'])
+        self.next_run = time.time()
+        self.shutdown_member_ttl = 300
+
+    @staticmethod
+    def get_dcs(name, config):
+        if 'etcd' in config:
+            return Etcd(name, config['etcd'])
+        if 'zookeeper' in config:
+            return ZooKeeper(name, config['zookeeper'])
+        raise Exception('Can not find sutable configuration of distributed configuration store')
+
+    def touch_member(self, ttl=None):
+        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
+        if self.ha.cluster:
+            for m in self.ha.cluster.members:
+                # Do not update member TTL when it is far from being expired
+                if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
+                    return True
+        return self.ha.dcs.touch_member(connection_string, ttl)
+
+    def initialize(self):
+        # wait for etcd to be available
+        while not self.touch_member():
+            logger.info('waiting on DCS')
+            sleep(5)
+
+        # is data directory empty?
+        if self.postgresql.data_directory_empty():
+            # racing to initialize
+            if self.ha.dcs.race('/initialize'):
+                self.postgresql.initialize()
+                self.ha.dcs.take_leader()
+                self.postgresql.start()
+                self.postgresql.create_replication_user()
+                self.postgresql.create_connection_users()
+            else:
+                while True:
+                    leader = self.ha.dcs.current_leader()
+                    if leader and self.postgresql.sync_from_leader(leader):
+                        self.postgresql.write_recovery_conf(leader)
+                        self.postgresql.start()
+                        break
+                    sleep(5)
+        elif self.postgresql.is_running():
+            self.postgresql.load_replication_slots()
+
+    def schedule_next_run(self):
+        self.next_run += self.nap_time
+        current_time = time.time()
+        nap_time = self.next_run - current_time
+        if nap_time <= 0:
+            self.next_run = current_time
+        else:
+            self.ha.dcs.sleep(nap_time)
+
+    def run(self):
+        self.api.start()
+        self.next_run = time.time()
+
+        while True:
+            self.touch_member()
+            logger.info(self.ha.run_cycle())
+            try:
+                if self.ha.state_handler.is_leader():
+                    self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
+                else:
+                    self.ha.state_handler.drop_replication_slots()
+            except:
+                logger.exception('Exception when changing replication slots')
+            reap_children()
+            self.schedule_next_run()
+
+
+def main():
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    logging.getLogger('requests').setLevel(logging.WARNING)
+    setup_signal_handlers()
+
+    if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
+        print('Usage: {} config.yml'.format(sys.argv[0]))
+        return
+
+    with open(sys.argv[1], 'r') as f:
+        config = yaml.load(f)
+
+    patroni = Patroni(config)
+    try:
+        patroni.initialize()
+        patroni.run()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        patroni.touch_member(patroni.shutdown_member_ttl)  # schedule member removal
+        patroni.postgresql.stop()
+        patroni.ha.dcs.delete_leader()

patroni/__main__.py

@@ -0,0 +1,5 @@
+from patroni import main
+
+
+if __name__ == '__main__':
+    main()

patroni/dcs.py

@@ -1,7 +1,7 @@
 import abc
 
 from collections import namedtuple
-from helpers.utils import calculate_ttl, sleep
+from patroni.utils import calculate_ttl, sleep
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
 
 

patroni/etcd.py

@@ -8,8 +8,8 @@ import socket
 
 from dns.exception import DNSException
 from dns import resolver
-from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
+from patroni.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
-from helpers.utils import sleep
+from patroni.utils import sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)

patroni/ha.py

@@ -1,6 +1,6 @@
 import logging
 
-from helpers.dcs import DCSError
+from patroni.dcs import DCSError
 from psycopg2 import InterfaceError, OperationalError
 
 logger = logging.getLogger(__name__)

patroni/postgresql.py

@@ -5,7 +5,7 @@ import shlex
 import shutil
 import subprocess
 
-from helpers.utils import sleep
+from patroni.utils import sleep
 from six.moves.urllib_parse import urlparse
 
 logger = logging.getLogger(__name__)

patroni/version.py

@@ -0,0 +1 @@
+__version__ = '0.1'

patroni/zookeeper.py

@@ -3,10 +3,10 @@ import random
 import requests
 import time
 
-from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
-from helpers.utils import sleep
 from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
+from patroni.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
+from patroni.utils import sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)

setup.py

@@ -19,13 +19,20 @@ if sys.version_info < (2, 7, 0):
 __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
 
 
+def read_version(package):
+    data = {}
+    with open(os.path.join(package, 'version.py'), 'r') as fd:
+        exec(fd.read(), data)
+    return data['__version__']
+
+
 NAME = 'patroni'
-MAIN_PACKAGE = 'patroni.py'
+MAIN_PACKAGE = NAME
-HELPERS = 'helpers'
 SCRIPTS = 'scripts'
+VERSION = read_version(MAIN_PACKAGE)
 VERSION = '0.1'
 DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
-LICENSE = 'The MIT License'
+LICENSE = 'MIT License'
 
 COVERAGE_XML = True
 COVERAGE_HTML = False
@@ -62,8 +69,7 @@ class PyTest(TestCommand):
     def finalize_options(self):
         TestCommand.finalize_options(self)
         if self.cov_xml or self.cov_html:
-            self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov', SCRIPTS, '--cov-report',
+            self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
-                        'term-missing']
             if self.cov_xml:
                 self.cov.extend(['--cov-report', 'xml'])
             if self.cov_html:
@@ -82,7 +88,7 @@ class PyTest(TestCommand):
             params['plugins'] = ['cov']
         if self.junitxml:
             params['args'] += self.junitxml
-        params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s']
+        params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv']
         errno = pytest.main(**params)
         sys.exit(errno)
 

tests/test_api.py

@@ -1,7 +1,7 @@
 import psycopg2
 import unittest
 
-from helpers.api import RestApiHandler, RestApiServer
+from patroni.api import RestApiHandler, RestApiServer
 from six import BytesIO as IO
 from test_postgresql import psycopg2_connect
 

tests/test_aws.py

@@ -2,7 +2,7 @@ import unittest
 import requests
 import boto.ec2
 from collections import namedtuple
-from scripts.aws import AWSConnection
+from patroni.scripts.aws import AWSConnection
 from requests.exceptions import RequestException
 
 

tests/test_etcd.py

@@ -8,9 +8,9 @@ import time
 import unittest
 
 from dns.exception import DNSException
-from helpers.dcs import Cluster, DCSError, Member
-from helpers.etcd import Client, Etcd
 from mock import Mock, patch
+from patroni.dcs import Cluster, DCSError, Member
+from patroni.etcd import Client, Etcd
 
 
 class MockResponse:

tests/test_ha.py

@@ -1,9 +1,9 @@
 import unittest
 
-from helpers.dcs import Cluster, DCSError
-from helpers.etcd import Client, Etcd
-from helpers.ha import Ha
 from mock import Mock, patch
+from patroni.dcs import Cluster, DCSError
+from patroni.etcd import Client, Etcd
+from patroni.ha import Ha
 from test_etcd import etcd_read, etcd_write
 
 

tests/test_patroni.py

@@ -1,5 +1,5 @@
 import datetime
-import helpers.zookeeper
+import patroni.zookeeper
 import psycopg2
 import subprocess
 import sys
@@ -7,12 +7,12 @@ import time
 import unittest
 import yaml
 
-from helpers.api import RestApiServer
-from helpers.dcs import Cluster, Member
-from helpers.etcd import Etcd
-from helpers.zookeeper import ZooKeeper
 from mock import Mock, patch
+from patroni.api import RestApiServer
+from patroni.dcs import Cluster, Member
+from patroni.etcd import Etcd
 from patroni import Patroni, main
+from patroni.zookeeper import ZooKeeper
 from six.moves import BaseHTTPServer
 from test_etcd import Client, etcd_read, etcd_write
 from test_ha import true, false
@@ -70,7 +70,7 @@ class TestPatroni(unittest.TestCase):
         Postgresql.write_recovery_conf = self.write_recovery_conf
 
     def test_get_dcs(self):
-        helpers.zookeeper.KazooClient = MockKazooClient
+        patroni.zookeeper.KazooClient = MockKazooClient
         self.assertIsInstance(self.p.get_dcs('', {'zookeeper': {'scope': '', 'hosts': ''}}), ZooKeeper)
         self.assertRaises(Exception, self.p.get_dcs, '', {})
 

tests/test_postgresql.py

@@ -4,8 +4,8 @@ import shutil
 import subprocess
 import unittest
 
-from helpers.dcs import Cluster, Member
+from patroni.dcs import Cluster, Member
-from helpers.postgresql import Postgresql
+from patroni.postgresql import Postgresql
 
 
 def nop(*args, **kwargs):

tests/test_restore.py

@@ -1,7 +1,7 @@
 import unittest
 from mock import MagicMock, patch
 import os
-from scripts.restore import Restore, WALERestore
+from patroni.scripts.restore import Restore, WALERestore
 
 
 def fake_cursor_fetchone(*args, **kwargs):

tests/test_utils.py

@@ -2,7 +2,7 @@ import os
 import time
 import unittest
 
-from helpers.utils import reap_children, sigchld_handler, sigterm_handler, sleep
+from patroni.utils import reap_children, sigchld_handler, sigterm_handler, sleep
 
 
 def nop(*args, **kwargs):

tests/test_zookeeper.py

@@ -1,8 +1,8 @@
-import helpers.zookeeper
+import patroni.zookeeper
 import requests
 import unittest
 
-from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
+from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
 from kazoo.client import KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.protocol.states import ZnodeStat
@@ -105,7 +105,7 @@ class TestExhibitorEnsembleProvider(unittest.TestCase):
 
     def set_up(self):
         requests.get = requests_get
-        helpers.zookeeper.sleep = exhibitor_sleep
+        patroni.zookeeper.sleep = exhibitor_sleep
 
     def test_init(self):
         self.assertRaises(Exception, ExhibitorEnsembleProvider, ['localhost'], 8181)
@@ -119,7 +119,7 @@ class TestZooKeeper(unittest.TestCase):
 
     def set_up(self):
         requests.get = requests_get
-        helpers.zookeeper.KazooClient = MockKazooClient
+        patroni.zookeeper.KazooClient = MockKazooClient
         self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'})
 
     def test_session_listener(self):