Convert roles to enums (#3303)

patroni/__main__.py

@@ -212,13 +212,15 @@ class Patroni(AbstractPatroniDaemon, Tags):
         the change and cache the new dynamic configuration values in ``patroni.dynamic.json`` file under Postgres data
         directory.
         """
+        from patroni.postgresql.misc import PostgresqlRole
+
         logger.info(self.ha.run_cycle())
 
         if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \
                 and self.config.set_dynamic_configuration(self.dcs.cluster.config):
             self.reload_config()
 
-        if self.postgresql.role != 'uninitialized':
+        if self.postgresql.role != PostgresqlRole.UNINITIALIZED:
             self.config.save_cache()
 
         self.schedule_next_run()

patroni/api.py

@@ -30,7 +30,7 @@ from . import global_config, psycopg
 from .__main__ import Patroni
 from .dcs import Cluster
 from .exceptions import PostgresConnectionException, PostgresException
-from .postgresql.misc import postgres_version_to_int, PostgresqlState
+from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState
 from .utils import cluster_as_json, deep_compare, enable_keepalive, parse_bool, \
     parse_int, patch_config, Retry, RetryFailedError, split_host_port, tzutc, uri
 
@@ -324,17 +324,19 @@ class RestApiHandler(BaseHTTPRequestHandler):
         is_lagging = leader_optime and leader_optime > replayed_location + max_replica_lag
 
         replica_status_code = 200 if not patroni.noloadbalance and not is_lagging and \
-            response.get('role') == 'replica' and response.get('state') == PostgresqlState.RUNNING else 503
+            response.get('role') == PostgresqlRole.REPLICA and response.get('state') == PostgresqlState.RUNNING else 503
 
         if not cluster and response.get('pause'):
-            leader_status_code = 200 if response.get('role') in ('primary', 'standby_leader') else 503
-            primary_status_code = 200 if response.get('role') == 'primary' else 503
-            standby_leader_status_code = 200 if response.get('role') == 'standby_leader' else 503
+            leader_status_code = 200 if response.get('role') in (PostgresqlRole.PRIMARY,
+                                                                 PostgresqlRole.STANDBY_LEADER) else 503
+            primary_status_code = 200 if response.get('role') == PostgresqlRole.PRIMARY else 503
+            standby_leader_status_code = 200 if response.get('role') == PostgresqlRole.STANDBY_LEADER else 503
         elif patroni.ha.is_leader():
             leader_status_code = 200
             if config.is_standby_cluster:
                 primary_status_code = replica_status_code = 503
-                standby_leader_status_code = 200 if response.get('role') in ('replica', 'standby_leader') else 503
+                standby_leader_status_code =\
+                    200 if response.get('role') in (PostgresqlRole.REPLICA, PostgresqlRole.STANDBY_LEADER) else 503
             else:
                 primary_status_code = 200
                 standby_leader_status_code = 503
@@ -436,7 +438,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         """
         patroni: Patroni = self.server.patroni
-        is_primary = patroni.postgresql.role == 'primary' and patroni.postgresql.is_running()
+        is_primary = patroni.postgresql.role == PostgresqlRole.PRIMARY and patroni.postgresql.is_running()
         # We can tolerate Patroni problems longer on the replica.
         # On the primary the liveness probe most likely will start failing only after the leader key expired.
         # It should not be a big problem because replicas will see that the primary is still alive via REST API call.
@@ -584,7 +586,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         metrics.append("# HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise.")
         metrics.append("# TYPE patroni_primary gauge")
-        metrics.append("patroni_primary{0} {1}".format(labels, int(postgres['role'] == 'primary')))
+        metrics.append("patroni_primary{0} {1}".format(labels, int(postgres['role'] == PostgresqlRole.PRIMARY)))
 
         metrics.append("# HELP patroni_xlog_location Current location of the Postgres"
                        " transaction log, 0 if this node is not the leader.")
@@ -593,11 +595,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.")
         metrics.append("# TYPE patroni_standby_leader gauge")
-        metrics.append("patroni_standby_leader{0} {1}".format(labels, int(postgres['role'] == 'standby_leader')))
+        metrics.append("patroni_standby_leader{0} {1}".format(labels,
+                                                              int(postgres['role'] == PostgresqlRole.STANDBY_LEADER)))
 
         metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.")
         metrics.append("# TYPE patroni_replica gauge")
-        metrics.append("patroni_replica{0} {1}".format(labels, int(postgres['role'] == 'replica')))
+        metrics.append("patroni_replica{0} {1}".format(labels, int(postgres['role'] == PostgresqlRole.REPLICA)))
 
         metrics.append("# HELP patroni_sync_standby Value is 1 if this node is a sync standby, 0 otherwise.")
         metrics.append("# TYPE patroni_sync_standby gauge")
@@ -910,7 +913,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     status_code = _
                     break
             elif k == 'role':
-                if request[k] not in ('primary', 'standby_leader', 'replica'):
+                if request[k] not in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER, PostgresqlRole.REPLICA):
                     status_code = 400
                     data = "PostgreSQL role should be either primary, standby_leader, or replica"
                     break
@@ -1266,16 +1269,17 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
             * ``state``: one of :class:`~patroni.postgresql.misc.PostgresqlState` or ``unknown``;
             * ``postmaster_start_time``: ``pg_postmaster_start_time()``;
-            * ``role``: ``replica`` or ``primary`` based on ``pg_is_in_recovery()`` output;
+            * ``role``: :class:`~patroni.postgresql.misc.PostgresqlRole.REPLICA` or
+                :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY` based on ``pg_is_in_recovery()`` output;
             * ``server_version``: Postgres version without periods, e.g. ``150002`` for Postgres ``15.2``;
             * ``latest_end_lsn``: latest_end_lsn value from ``pg_stat_get_wal_receiver()``, only on replica nodes;
             * ``xlog``: dictionary. Its structure depends on ``role``:
 
-                * If ``primary``:
+                * If :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY`:
 
                     * ``location``: ``pg_current_wal_flush_lsn()``
 
-                * If ``replica``:
+                * If :class:`~patroni.postgresql.misc.PostgresqlRole.REPLICA`:
 
                     * ``received_location``: ``pg_wal_lsn_diff(pg_last_wal_receive_lsn(), '0/0')``;
                     * ``replayed_location``: ``pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/0)``;
@@ -1327,7 +1331,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             result = {
                 'state': postgresql.state,
                 'postmaster_start_time': row[0],
-                'role': 'replica' if row[1] == 0 else 'primary',
+                'role': PostgresqlRole.REPLICA if row[1] == 0 else PostgresqlRole.PRIMARY,
                 'server_version': postgresql.server_version,
                 'xlog': ({
                     'received_location': row[10] or row[4] or row[3],
@@ -1338,10 +1342,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 })
             }
 
-            if result['role'] == 'replica' and config.is_standby_cluster:
+            if result['role'] == PostgresqlRole.REPLICA and config.is_standby_cluster:
                 result['role'] = postgresql.role
 
-            if result['role'] == 'replica' and config.is_synchronous_mode\
+            if result['role'] == PostgresqlRole.REPLICA and config.is_synchronous_mode\
                     and cluster and cluster.sync.matches(postgresql.name):
                 result['quorum_standby' if global_config.is_quorum_commit_mode else 'sync_standby'] = True
 

patroni/ctl.py

@@ -29,6 +29,7 @@ import time
 
 from collections import defaultdict
 from contextlib import contextmanager
+from enum import Enum
 from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union
 from urllib.parse import urlparse
 
@@ -64,7 +65,7 @@ from . import global_config
 from .config import Config
 from .dcs import AbstractDCS, Cluster, get_dcs as _get_dcs, Member
 from .exceptions import PatroniException
-from .postgresql.misc import postgres_version_to_int, PostgresqlState
+from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState
 from .postgresql.mpp import get_mpp
 from .request import PatroniRequest
 from .utils import cluster_as_json, patch_config, polling_loop
@@ -80,6 +81,20 @@ DCS_DEFAULTS: Dict[str, Dict[str, Any]] = {
     'etcd3': {'port': 2379, 'template': "etcd3:\n host: '{host}:{port}'"}}
 
 
+class CtlPostgresqlRole(str, Enum):
+
+    LEADER = 'leader'
+    PRIMARY = 'primary'
+    STANDBY_LEADER = 'standby-leader'
+    REPLICA = 'replica'
+    STANDBY = 'standby'
+    ANY = 'any'
+
+    def __repr__(self) -> str:
+        """Get a string representation of a :class:`CtlPostgresqlRole` member."""
+        return self.value
+
+
 class PatroniCtlException(click.ClickException):
     """Raised upon issues faced by ``patronictl`` utility."""
 
@@ -289,7 +304,7 @@ arg_cluster_name = click.argument('cluster_name', required=False,
 option_default_citus_group = click.option('--group', required=False, type=int, help='Citus group',
                                           default=lambda: _get_configuration().get('citus', {}).get('group'))
 option_citus_group = click.option('--group', required=False, type=int, help='Citus group')
-role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any'])
+role_choice = click.Choice([role.value for role in CtlPostgresqlRole])
 
 
 @click.group(cls=click.Group)
@@ -493,45 +508,42 @@ def watching(w: bool, watch: Optional[int], max_count: Optional[int] = None, cle
         yield 0
 
 
-def get_all_members(cluster: Cluster, group: Optional[int], role: str = 'leader') -> Iterator[Member]:
+def get_all_members(cluster: Cluster, group: Optional[int],
+                    role: CtlPostgresqlRole = CtlPostgresqlRole.LEADER) -> Iterator[Member]:
     """Get all cluster members that have the given *role*.
 
     :param cluster: the Patroni cluster.
     :param group: filter which Citus group we should get members from. If ``None`` get from all groups.
-    :param role: role to filter members. Can be one among:
-
-        * ``primary``: the primary PostgreSQL instance;
-        * ``replica`` or ``standby``: a standby PostgreSQL instance;
-        * ``leader``: the leader of a Patroni cluster. Can also be used to get the leader of a Patroni standby cluster;
-        * ``standby-leader``: the leader of a Patroni standby cluster;
-        * ``any``: matches any node independent of its role.
+    :param role: role to filter members. One of :class:`CtlPostgresqlRole` values.
 
     :yields: members that have the given *role*.
     """
     clusters = {0: cluster}
     if is_citus_cluster() and group is None:
         clusters.update(cluster.workers)
-    if role in ('leader', 'primary', 'standby-leader'):
+    if role in (CtlPostgresqlRole.LEADER, CtlPostgresqlRole.PRIMARY, CtlPostgresqlRole.STANDBY_LEADER):
         # In the DCS the members' role can be one among: ``primary``, ``master``, ``replica`` or ``standby_leader``.
         # ``primary`` and ``master`` are the same thing.
-        role = {'standby-leader': 'standby_leader'}.get(role, role)
         for cluster in clusters.values():
             if cluster.leader is not None and cluster.leader.name and\
-                    (role == 'leader'
-                     or cluster.leader.data.get('role') not in ('primary', 'master') and role == 'standby_leader'
-                     or cluster.leader.data.get('role') != 'standby_leader' and role == 'primary'):
+                (role == CtlPostgresqlRole.LEADER
+                 or cluster.leader.data.get('role') not in (PostgresqlRole.PRIMARY, PostgresqlRole.MASTER)
+                 and role == CtlPostgresqlRole.STANDBY_LEADER
+                 or cluster.leader.data.get('role') != PostgresqlRole.STANDBY_LEADER
+                 and role == CtlPostgresqlRole.PRIMARY):
                 yield cluster.leader.member
         return
 
     for cluster in clusters.values():
         leader_name = (cluster.leader.member.name if cluster.leader else None)
         for m in cluster.members:
-            if role == 'any' or role in ('replica', 'standby') and m.name != leader_name:
+            if role == CtlPostgresqlRole.ANY or\
+                    role in (CtlPostgresqlRole.REPLICA, CtlPostgresqlRole.STANDBY) and m.name != leader_name:
                 yield m
 
 
 def get_any_member(cluster: Cluster, group: Optional[int],
-                   role: Optional[str] = None, member: Optional[str] = None) -> Optional[Member]:
+                   role: Optional[CtlPostgresqlRole] = None, member: Optional[str] = None) -> Optional[Member]:
     """Get the first found cluster member that has the given *role*.
 
     :param cluster: the Patroni cluster.
@@ -547,9 +559,9 @@ def get_any_member(cluster: Cluster, group: Optional[int],
     if member is not None:
         if role is not None:
             raise PatroniCtlException('--role and --member are mutually exclusive options')
-        role = 'any'
+        role = CtlPostgresqlRole.ANY
     elif role is None:
-        role = 'leader'
+        role = CtlPostgresqlRole.LEADER
 
     for m in get_all_members(cluster, group, role):
         if member is None or m.name == member:
@@ -573,7 +585,8 @@ def get_all_members_leader_first(cluster: Cluster) -> Iterator[Member]:
 
 
 def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[str, Any],
-               role: Optional[str] = None, member_name: Optional[str] = None) -> Union['cursor', 'Cursor[Any]', None]:
+               role: Optional[CtlPostgresqlRole] = None,
+               member_name: Optional[str] = None) -> Union['cursor', 'Cursor[Any]', None]:
     """Get a cursor object to execute queries against a member that has the given *role* or *member_name*.
 
     .. note::
@@ -612,7 +625,7 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[
     # If we want ``any`` node we are fine to return the cursor. ``None`` is similar to ``any`` at this point, as it's
     # been dealt with through :func:`get_any_member`.
     # If we want the Patroni leader node, :func:`get_any_member` already checks that for us
-    if role in (None, 'any', 'leader'):
+    if role in (None, CtlPostgresqlRole.ANY, CtlPostgresqlRole.LEADER):
         return cursor
 
     # If we want something other than ``any`` or ``leader``, then we do not rely only on the DCS information about
@@ -621,7 +634,9 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[
     row = cursor.fetchone()
     in_recovery = not row or row[0]
 
-    if in_recovery and role in ('replica', 'standby', 'standby-leader') or not in_recovery and role == 'primary':
+    if in_recovery and\
+        role in (CtlPostgresqlRole.REPLICA, CtlPostgresqlRole.STANDBY, CtlPostgresqlRole.STANDBY_LEADER) or\
+            not in_recovery and role == CtlPostgresqlRole.PRIMARY:
         return cursor
 
     conn.close()
@@ -629,7 +644,7 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[
     return None
 
 
-def get_members(cluster: Cluster, cluster_name: str, member_names: List[str], role: str,
+def get_members(cluster: Cluster, cluster_name: str, member_names: List[str], role: CtlPostgresqlRole,
                 force: bool, action: str, ask_confirmation: bool = True, group: Optional[int] = None) -> List[Member]:
     """Get the list of members based on the given filters.
 
@@ -752,7 +767,7 @@ def confirm_members_action(members: List[Member], force: bool, action: str,
 @click.option('--member', '-m', help='Generate a dsn for this member', type=str)
 @arg_cluster_name
 @option_citus_group
-def dsn(cluster_name: str, group: Optional[int], role: Optional[str], member: Optional[str]) -> None:
+def dsn(cluster_name: str, group: Optional[int], role: Optional[CtlPostgresqlRole], member: Optional[str]) -> None:
     """Process ``dsn`` command of ``patronictl`` utility.
 
     Get DSN to connect to *member*.
@@ -798,7 +813,7 @@ def dsn(cluster_name: str, group: Optional[int], role: Optional[str], member: Op
 def query(
     cluster_name: str,
     group: Optional[int],
-    role: Optional[str],
+    role: Optional[CtlPostgresqlRole],
     member: Optional[str],
     w: bool,
     watch: Optional[int],
@@ -865,7 +880,7 @@ def query(
 
 
 def query_member(cluster: Cluster, group: Optional[int], cursor: Union['cursor', 'Cursor[Any]', None],
-                 member: Optional[str], role: Optional[str], command: str,
+                 member: Optional[str], role: Optional[CtlPostgresqlRole], command: str,
                  connect_parameters: Dict[str, Any]) -> Tuple[List[List[Any]], Optional[List[Any]]]:
     """Execute SQL *command* against a member.
 
@@ -903,7 +918,7 @@ def query_member(cluster: Cluster, group: Optional[int], cursor: Union['cursor',
             if member is not None:
                 message = f'No connection to member {member} is available'
             elif role is not None:
-                message = f'No connection to role {role} is available'
+                message = f'No connection to role {role!r} is available'
             else:
                 message = 'No connection is available'
             logging.debug(message)
@@ -1030,9 +1045,11 @@ def parse_scheduled(scheduled: Optional[str]) -> Optional[datetime.datetime]:
 @click.argument('cluster_name')
 @click.argument('member_names', nargs=-1)
 @option_citus_group
-@click.option('--role', '-r', help='Reload only members with this role', type=role_choice, default='any')
+@click.option('--role', '-r', help='Reload only members with this role',
+              type=role_choice, default=CtlPostgresqlRole.ANY)
 @option_force
-def reload(cluster_name: str, member_names: List[str], group: Optional[int], force: bool, role: str) -> None:
+def reload(cluster_name: str, member_names: List[str], group: Optional[int],
+           force: bool, role: CtlPostgresqlRole) -> None:
     """Process ``reload`` command of ``patronictl`` utility.
 
     Reload configuration of cluster members based on given filters.
@@ -1067,7 +1084,8 @@ def reload(cluster_name: str, member_names: List[str], group: Optional[int], for
 @click.argument('cluster_name')
 @click.argument('member_names', nargs=-1)
 @option_citus_group
-@click.option('--role', '-r', help='Restart only members with this role', type=role_choice, default='any')
+@click.option('--role', '-r', help='Restart only members with this role', type=role_choice,
+              default=CtlPostgresqlRole.ANY)
 @click.option('--any', 'p_any', help='Restart a single member only', is_flag=True)
 @click.option('--scheduled', help='Timestamp of a scheduled restart in unambiguous format (e.g. ISO 8601)',
               default=None)
@@ -1077,7 +1095,7 @@ def reload(cluster_name: str, member_names: List[str], group: Optional[int], for
 @click.option('--timeout', help='Return error and fail over if necessary when restarting takes longer than this.')
 @option_force
 def restart(cluster_name: str, group: Optional[int], member_names: List[str],
-            force: bool, role: str, p_any: bool, scheduled: Optional[str], version: Optional[str],
+            force: bool, role: CtlPostgresqlRole, p_any: bool, scheduled: Optional[str], version: Optional[str],
             pending: bool, timeout: Optional[str]) -> None:
     """Process ``restart`` command of ``patronictl`` utility.
 
@@ -1184,7 +1202,8 @@ def reinit(cluster_name: str, group: Optional[int], member_names: List[str], for
     :param wait: wait for the operation to complete.
     """
     cluster = get_dcs(cluster_name, group).get_cluster()
-    members = get_members(cluster, cluster_name, member_names, 'replica', force, 'reinitialize', group=group)
+    members = get_members(cluster, cluster_name, member_names, CtlPostgresqlRole.REPLICA,
+                          force, 'reinitialize', group=group)
 
     wait_on_members: List[Member] = []
     for member in members:
@@ -1709,10 +1728,10 @@ def timestamp(precision: int = 6) -> str:
 @option_citus_group
 @click.argument('member_names', nargs=-1)
 @click.argument('target', type=click.Choice(['restart', 'switchover']))
-@click.option('--role', '-r', help='Flush only members with this role', type=role_choice, default='any')
+@click.option('--role', '-r', help='Flush only members with this role', type=role_choice, default=CtlPostgresqlRole.ANY)
 @option_force
 def flush(cluster_name: str, group: Optional[int],
-          member_names: List[str], force: bool, role: str, target: str) -> None:
+          member_names: List[str], force: bool, role: CtlPostgresqlRole, target: str) -> None:
     """Process ``flush`` command of ``patronictl`` utility.
 
     Discard scheduled restart or switchover events.
@@ -2193,7 +2212,7 @@ def version(cluster_name: str, group: Optional[int], member_names: List[str]) ->
 
     click.echo("")
     cluster = get_dcs(cluster_name, group).get_cluster()
-    for m in get_all_members(cluster, group, 'any'):
+    for m in get_all_members(cluster, group, CtlPostgresqlRole.ANY):
         if m.api_url:
             if not member_names or m.name in member_names:
                 try:

patroni/dcs/__init__.py

@@ -25,6 +25,7 @@ from ..utils import deep_compare, parse_int, uri
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
     from ..postgresql import Postgresql
+    from ..postgresql.misc import PostgresqlRole
     from ..postgresql.mpp import AbstractMPP
 
 slot_name_re = re.compile('^[a-z0-9_]{1,63}$')
@@ -416,10 +417,13 @@ class Leader(NamedTuple):
 
             >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote
         """
+        from ..postgresql.misc import PostgresqlRole
+
         version = self.member.patroni_version
         # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false
         if version and version > (1, 5, 6):
-            return self.data.get('role') in ('master', 'primary') and 'checkpoint_after_promote' not in self.data
+            return self.data.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY) \
+                and 'checkpoint_after_promote' not in self.data
         return None
 
 
@@ -1010,7 +1014,8 @@ class Cluster(NamedTuple('Cluster',
         return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)}
 
     def get_replication_slots(self, postgresql: 'Postgresql', member: Tags, *,
-                              role: Optional[str] = None, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
+                              role: Optional['PostgresqlRole'] = None,
+                              show_error: bool = False) -> Dict[str, Dict[str, Any]]:
         """Lookup configured slot names in the DCS, report issues found and merge with permanent slots.
 
         Will log an error if:
@@ -1019,7 +1024,8 @@ class Cluster(NamedTuple('Cluster',
 
         :param postgresql: reference to :class:`Postgresql` object.
         :param member: reference to an object implementing :class:`Tags` interface.
-        :param role: role of the node, if not set will be taken from *postgresql*.
+        :param role: role of the node, if not set will be taken from *postgresql*
+            One of :class:`~patroni.postgresql.misc.PostgresqlRole` values.
         :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found.
 
         :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
@@ -1041,7 +1047,7 @@ class Cluster(NamedTuple('Cluster',
         return slots
 
     def _merge_permanent_slots(self, slots: Dict[str, Dict[str, Any]], permanent_slots: Dict[str, Any],
-                               name: str, role: str, can_advance_slots: bool) -> List[str]:
+                               name: str, role: 'PostgresqlRole', can_advance_slots: bool) -> List[str]:
         """Merge replication *slots* for members with *permanent_slots*.
 
         Perform validation of configured permanent slot name, skipping invalid names.
@@ -1051,13 +1057,15 @@ class Cluster(NamedTuple('Cluster',
 
         :param slots: Slot names with existing attributes if known.
         :param name: name of this node.
-        :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``.
+        :param role: role of the node. One of :class:`~patroni.postgresql.misc.PostgresqlRole` values.
         :param permanent_slots: dictionary containing slot name key and slot information values.
         :param can_advance_slots: ``True`` if ``pg_replication_slot_advance()`` function is available,
                                   ``False`` otherwise.
 
         :returns: List of disabled permanent, logical slot names, if postgresql version < 11.
         """
+        from ..postgresql.misc import PostgresqlRole
+
         name = slot_name_from_member_name(name)
         topology = {slot_name_from_member_name(m.name): m.replicatefrom and slot_name_from_member_name(m.replicatefrom)
                     for m in self.members}
@@ -1084,7 +1092,8 @@ class Cluster(NamedTuple('Cluster',
                         # case we should have the following: A(B: active, C: inactive) <- B (C: active) <- C
                         # We don't consider the same situation on node B, because if node C doesn't exists, we will not
                         # be able to know its `replicatefrom` tag value.
-                        expected_active = not topology.get(slot_name) and role in ('primary', 'standby_leader')
+                        expected_active = not topology.get(slot_name) and role in (PostgresqlRole.PRIMARY,
+                                                                                   PostgresqlRole.STANDBY_LEADER)
                         slots[slot_name] = {**value, 'expected_active': expected_active}
                     continue
 
@@ -1101,7 +1110,7 @@ class Cluster(NamedTuple('Cluster',
             logger.error("Bad value for slot '%s' in permanent_slots: %s", slot_name, permanent_slots[slot_name])
         return disabled_permanent_logical_slots
 
-    def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: str) -> Dict[str, Any]:
+    def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: 'PostgresqlRole') -> Dict[str, Any]:
         """Get configured permanent replication slots.
 
         .. note::
@@ -1117,20 +1126,23 @@ class Cluster(NamedTuple('Cluster',
 
         :param postgresql: reference to :class:`Postgresql` object.
         :param tags: reference to an object implementing :class:`Tags` interface.
-        :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``.
+        :param role: role of the node. One of :class:`~patroni.postgresql.misc.PostgresqlRole` values.
 
         :returns: dictionary of permanent slot names mapped to attributes.
         """
+        from ..postgresql.misc import PostgresqlRole
+
         if not global_config.use_slots or tags.nofailover:
             return {}
 
         if global_config.is_standby_cluster or self.get_slot_name_on_primary(postgresql.name, tags) is None:
-            return self.permanent_physical_slots if postgresql.can_advance_slots or role == 'standby_leader' else {}
+            return self.permanent_physical_slots\
+                if postgresql.can_advance_slots or role == PostgresqlRole.STANDBY_LEADER else {}
 
-        return self.__permanent_slots if postgresql.can_advance_slots or role == 'primary' \
+        return self.__permanent_slots if postgresql.can_advance_slots or role == PostgresqlRole.PRIMARY \
             else self.__permanent_logical_slots
 
-    def _get_members_slots(self, name: str, role: str, nofailover: bool,
+    def _get_members_slots(self, name: str, role: 'PostgresqlRole', nofailover: bool,
                            can_advance_slots: bool) -> Dict[str, Dict[str, Any]]:
         """Get physical replication slots configuration for a given member.
 
@@ -1162,6 +1174,8 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: dictionary of physical replication slots that should exist on a given node.
         """
+        from ..postgresql.misc import PostgresqlRole
+
         if not global_config.use_slots:
             return {}
 
@@ -1195,7 +1209,8 @@ class Cluster(NamedTuple('Cluster',
         # In case when retention of replication slots is possible the `expected_active` function
         # will be used to figure out whether the replication slot is expected to be active.
         # Otherwise it will be used to find replication slots that should exist on a current node.
-        expected_active = leader_filter if role in ('primary', 'standby_leader') else replica_filter
+        expected_active = leader_filter if role in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER) \
+            else replica_filter
 
         if can_advance_slots and global_config.member_slots_ttl > 0:
             # if the node does only cascading and can't become the leader, we
@@ -1239,7 +1254,9 @@ class Cluster(NamedTuple('Cluster',
                        the node that we are checking permanent logical replication slots for.
         :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``.
         """
-        role = 'replica'
+        from ..postgresql.misc import PostgresqlRole
+
+        role = PostgresqlRole.REPLICA
         members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(postgresql.name, role,
                                                                            member.nofailover,
                                                                            postgresql.can_advance_slots)
@@ -1260,10 +1277,13 @@ class Cluster(NamedTuple('Cluster',
         :param slots: slot names with LSN values.
         :returns: a :class:`dict` object that contains only slots that are known to be permanent.
         """
+        from ..postgresql.misc import PostgresqlRole
+
         if global_config.member_slots_ttl > 0:
             return slots
 
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}), 'replica')
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}),
+                                                                    PostgresqlRole.REPLICA)
         members_slots = {slot_name_from_member_name(m.name) for m in self.members}
 
         return {name: value for name, value in slots.items() if name in permanent_slots
@@ -1279,7 +1299,9 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``.
         """
-        slots = self.get_replication_slots(postgresql, member, role='replica').values()
+        from ..postgresql.misc import PostgresqlRole
+
+        slots = self.get_replication_slots(postgresql, member, role=PostgresqlRole.REPLICA).values()
         return any(v for v in slots if v.get("type") == "logical")
 
     def should_enforce_hot_standby_feedback(self, postgresql: 'Postgresql', member: Tags) -> bool:

patroni/dcs/consul.py

@@ -19,7 +19,7 @@ from consul import base, Check, ConsulException, NotFound
 from urllib3.exceptions import HTTPError
 
 from ..exceptions import DCSError
-from ..postgresql.misc import PostgresqlState
+from ..postgresql.misc import PostgresqlRole, PostgresqlState
 from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
 from . import AbstractDCS, catch_return_false_exception, Cluster, ClusterConfig, \
@@ -533,8 +533,8 @@ class Consul(AbstractDCS):
             check['TLSServerName'] = self._service_check_tls_server_name
         tags = self._service_tags[:]
         tags.append(role)
-        if role == 'primary':
-            tags.append('master')
+        if data['role'] == PostgresqlRole.PRIMARY:
+            tags.append(PostgresqlRole.MASTER)
         self._previous_loop_service_tags = self._service_tags
         self._previous_loop_token = self._client.token
 
@@ -552,7 +552,7 @@ class Consul(AbstractDCS):
             return self.deregister_service(params['service_id'])
 
         self._previous_loop_register_service = self._register_service
-        if role in ['primary', 'replica', 'standby-leader']:
+        if data['role'] in [PostgresqlRole.PRIMARY, PostgresqlRole.REPLICA, PostgresqlRole.STANDBY_LEADER]:
             if state != PostgresqlState.RUNNING:
                 return
             return self.register_service(service_name, **params)

patroni/dcs/kubernetes.py

@@ -23,7 +23,7 @@ from urllib3.exceptions import HTTPError
 
 from ..collections import EMPTY_DICT
 from ..exceptions import DCSError
-from ..postgresql.misc import PostgresqlState
+from ..postgresql.misc import PostgresqlRole, PostgresqlState
 from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, iter_response_objects, \
     keepalive_socket_options, Retry, RetryFailedError, tzutc, uri, USER_AGENT
@@ -1319,9 +1319,10 @@ class Kubernetes(AbstractDCS):
     def touch_member(self, data: Dict[str, Any]) -> bool:
         cluster = self.cluster
         if cluster and cluster.leader and cluster.leader.name == self._name:
-            role = self._standby_leader_label_value if data['role'] == 'standby_leader' else self._leader_label_value
+            role = self._standby_leader_label_value \
+                if data['role'] == PostgresqlRole.STANDBY_LEADER else self._leader_label_value
             tmp_role = 'primary'
-        elif data['state'] == PostgresqlState.RUNNING and data['role'] != 'primary':
+        elif data['state'] == PostgresqlState.RUNNING and data['role'] != PostgresqlRole.PRIMARY:
             role = {'replica': self._follower_label_value}.get(data['role'], data['role'])
             tmp_role = data['role']
         else:

patroni/ha.py

@@ -17,7 +17,7 @@ from .collections import CaseInsensitiveSet
 from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, SyncState
 from .exceptions import DCSError, PatroniFatalException, PostgresConnectionException
 from .postgresql.callback_executor import CallbackAction
-from .postgresql.misc import postgres_version_to_int, PostgresqlState
+from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState
 from .postgresql.postmaster import PostmasterProcess
 from .postgresql.rewind import Rewind
 from .quorum import QuorumStateResolver
@@ -54,7 +54,8 @@ class _MemberStatus(Tags, NamedTuple('_MemberStatus',
         # If one of those is not in a response we want to count the node as not healthy/reachable
         wal: Dict[str, Any] = json.get('wal') or json['xlog']
         # abuse difference in primary/replica response format
-        in_recovery = not (bool(wal.get('location')) or json.get('role') in ('master', 'primary'))
+        in_recovery = not (bool(wal.get('location'))
+                           or json.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY))
         lsn = int(in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0)))
         return cls(member, True, in_recovery, lsn, json)
 
@@ -474,7 +475,7 @@ class Ha(object):
                         # Unfortunately such optimization isn't possible on the standby_leader,
                         # therefore we will get the timeline from pg_control, either by calling
                         # pg_control_checkpoint() on 9.6+ or by parsing the output of pg_controldata.
-                        if self.state_handler.role == 'standby_leader':
+                        if self.state_handler.role == PostgresqlRole.STANDBY_LEADER:
                             timeline = pg_control_timeline or self.state_handler.pg_control_timeline()
                         else:
                             timeline = self.state_handler.replica_cached_timeline(self._leader_timeline) or 0
@@ -493,7 +494,7 @@ class Ha(object):
             ret = self.dcs.touch_member(data)
             if ret:
                 new_state = (data['state'], data['role'])
-                if self._last_state != new_state and new_state == (PostgresqlState.RUNNING, 'primary'):
+                if self._last_state != new_state and new_state == (PostgresqlState.RUNNING, PostgresqlRole.PRIMARY):
                     self.notify_mpp_coordinator('after_promote')
                 self._last_state = new_state
             return ret
@@ -560,7 +561,7 @@ class Ha(object):
         with self._async_response:  # pretend that post_bootstrap was already executed
             self._async_response.complete(result)
         if result:
-            self.state_handler.set_role('standby_leader')
+            self.state_handler.set_role(PostgresqlRole.STANDBY_LEADER)
 
         return result
 
@@ -635,7 +636,7 @@ class Ha(object):
                 and data.get('Database cluster state') in ('in production', 'in crash recovery',
                                                            'shutting down', 'shut down')\
                 and self.state_handler.state == PostgresqlState.CRASHED\
-                and self.state_handler.role == 'primary'\
+                and self.state_handler.role == PostgresqlRole.PRIMARY\
                 and not self.state_handler.config.recovery_conf_exists():
             # We know 100% that we were running as a primary a few moments ago, therefore could just start postgres
             msg = 'starting primary after failure'
@@ -654,7 +655,7 @@ class Ha(object):
 
         self.load_cluster_from_dcs()
 
-        role = 'replica'
+        role = PostgresqlRole.REPLICA
         if self.has_lock() and not self.is_standby_cluster():
             self._rewind.reset_state()  # we want to later trigger CHECKPOINT after promote
             msg = "starting as readonly because i had the session lock"
@@ -668,7 +669,7 @@ class Ha(object):
 
             if self.has_lock():  # in standby cluster
                 msg = "starting as a standby leader because i had the session lock"
-                role = 'standby_leader'
+                role = PostgresqlRole.STANDBY_LEADER
                 node_to_follow = self._get_node_to_follow(self.cluster)
             elif self.is_standby_cluster() and self.cluster.is_unlocked():
                 msg = "trying to follow a remote member because standby cluster is unhealthy"
@@ -732,10 +733,10 @@ class Ha(object):
             if not (self._rewind.is_needed and self._rewind.can_rewind_or_reinitialize_allowed)\
                     or self.cluster.is_unlocked():
                 if is_leader:
-                    self.state_handler.set_role('primary')
+                    self.state_handler.set_role(PostgresqlRole.PRIMARY)
                     return 'continue to run as primary without lock'
-                elif self.state_handler.role != 'standby_leader':
-                    self.state_handler.set_role('replica')
+                elif self.state_handler.role != PostgresqlRole.STANDBY_LEADER:
+                    self.state_handler.set_role(PostgresqlRole.REPLICA)
 
                 if not node_to_follow:
                     return 'no action. I am ({0})'.format(self.state_handler.name)
@@ -755,10 +756,12 @@ class Ha(object):
         if not self.is_paused():
             self.state_handler.handle_parameter_change()
 
-        role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
+        role = PostgresqlRole.STANDBY_LEADER \
+            if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else PostgresqlRole.REPLICA
         # It might happen that leader key in the standby cluster references non-exiting member.
         # In this case it is safe to continue running without changing recovery.conf
-        if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url):
+        if self.is_standby_cluster() and role == PostgresqlRole.REPLICA \
+                and not (node_to_follow and node_to_follow.conn_url):
             return 'continue following the old known standby leader'
         else:
             change_required, restart_required = self.state_handler.config.check_recovery_conf(node_to_follow)
@@ -769,7 +772,7 @@ class Ha(object):
                 else:
                     self.state_handler.follow(node_to_follow, role, do_reload=True)
                 self._rewind.trigger_check_diverged_lsn()
-            elif role == 'standby_leader' and self.state_handler.role != role:
+            elif role == PostgresqlRole.STANDBY_LEADER and self.state_handler.role != role:
                 self.state_handler.set_role(role)
                 self.state_handler.call_nowait(CallbackAction.ON_ROLE_CHANGE)
 
@@ -1095,12 +1098,12 @@ class Ha(object):
         if self.state_handler.is_primary():
             # Inform the state handler about its primary role.
             # It may be unaware of it if postgres is promoted manually.
-            self.state_handler.set_role('primary')
+            self.state_handler.set_role(PostgresqlRole.PRIMARY)
             self.process_sync_replication()
             self.update_cluster_history()
             self.state_handler.mpp_handler.sync_meta_data(self.cluster)
             return message
-        elif self.state_handler.role in ('primary', 'promoted'):
+        elif self.state_handler.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED):
             self.process_sync_replication()
             return message
         else:
@@ -1108,7 +1111,7 @@ class Ha(object):
                 # Somebody else updated sync state, it may be due to us losing the lock. To be safe,
                 # postpone promotion until next cycle. TODO: trigger immediate retry of run_cycle.
                 return 'Postponing promotion because synchronous replication state was updated by somebody else'
-            if self.state_handler.role not in ('primary', 'promoted'):
+            if self.state_handler.role not in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED):
                 # reset failsafe state when promote
                 self._failsafe.set_is_active(0)
 
@@ -1156,7 +1159,7 @@ class Ha(object):
 
         :returns: the reason why caller shouldn't continue as a primary or the current value of received/replayed LSN.
         """
-        if self.state_handler.state == PostgresqlState.RUNNING and self.state_handler.role == 'primary':
+        if self.state_handler.state == PostgresqlState.RUNNING and self.state_handler.role == PostgresqlRole.PRIMARY:
             return 'Running as a leader'
         self._failsafe.update(data)
         return self._last_wal_lsn
@@ -1570,7 +1573,7 @@ class Ha(object):
             # location, we can remove the leader key and allow them to start leader race.
             time.sleep(1)  # give replicas some more time to catch up
             if self.is_failover_possible(cluster_lsn=checkpoint_location):
-                self.state_handler.set_role('demoted')
+                self.state_handler.set_role(PostgresqlRole.DEMOTED)
                 with self._async_executor:
                     self.release_leader_key_voluntarily(prev_location)
                     status['released'] = True
@@ -1586,7 +1589,7 @@ class Ha(object):
                                 on_shutdown=on_shutdown if mode_control['release'] else None,
                                 before_shutdown=before_shutdown if mode == 'graceful' else None,
                                 stop_timeout=self.primary_stop_timeout())
-        self.state_handler.set_role('demoted')
+        self.state_handler.set_role(PostgresqlRole.DEMOTED)
         self.set_is_leader(False)
 
         if mode_control['release']:
@@ -1758,7 +1761,7 @@ class Ha(object):
                     # enforce anything, since the leader is not a primary
                     # So just remind the role.
                     msg = 'no action. I am ({0}), the standby leader with the lock'.format(self.state_handler.name) \
-                          if self.state_handler.role == 'standby_leader' else \
+                          if self.state_handler.role == PostgresqlRole.STANDBY_LEADER else \
                           'promoted self to a standby leader because i had the session lock'
                     return self.enforce_follow_remote_member(msg)
                 else:
@@ -1958,7 +1961,7 @@ class Ha(object):
                     self.state_handler.cancellable.cancel()
                     return 'lost leader before promote'
 
-            if self.state_handler.role == 'primary':
+            if self.state_handler.role == PostgresqlRole.PRIMARY:
                 logger.info('Demoting primary during %s', self._async_executor.scheduled_action)
                 if self._async_executor.scheduled_action in ('restart', 'starting primary after failure'):
                     # Restart needs a special interlocking cancel because postmaster may be just started in a
@@ -1984,8 +1987,8 @@ class Ha(object):
         if not self.state_handler.is_running():
             self.watchdog.disable()
             if self.has_lock():
-                if self.state_handler.role in ('primary', 'standby_leader'):
-                    self.state_handler.set_role('demoted')
+                if self.state_handler.role in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER):
+                    self.state_handler.set_role(PostgresqlRole.DEMOTED)
                     self.state_handler.call_nowait(CallbackAction.ON_ROLE_CHANGE)
                 self._delete_leader()
                 return 'removed leader key after trying and failing to start postgres'
@@ -2010,7 +2013,7 @@ class Ha(object):
             if not self.state_handler.is_primary():
                 return 'waiting for end of recovery after bootstrap'
 
-            self.state_handler.set_role('primary')
+            self.state_handler.set_role(PostgresqlRole.PRIMARY)
             ret = self._async_executor.try_run_async('post_bootstrap', self.state_handler.bootstrap.post_bootstrap,
                                                      args=(self.patroni.config['bootstrap'], self._async_response))
             return ret or 'running post_bootstrap'
@@ -2160,7 +2163,7 @@ class Ha(object):
                 data_directory_error = e
 
             if not data_directory_is_accessible or data_directory_is_empty:
-                self.state_handler.set_role('uninitialized')
+                self.state_handler.set_role(PostgresqlRole.UNINITIALIZED)
                 self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout'])
                 # In case datadir went away while we were primary
                 self.watchdog.disable()

patroni/postgresql/__init__.py

@@ -28,7 +28,7 @@ from .callback_executor import CallbackAction, CallbackExecutor
 from .cancellable import CancellableSubprocess
 from .config import ConfigHandler, mtime
 from .connection import ConnectionPool, get_connection_cursor
-from .misc import parse_history, parse_lsn, postgres_major_version_to_int, PostgresqlState
+from .misc import parse_history, parse_lsn, postgres_major_version_to_int, PostgresqlRole, PostgresqlState
 from .mpp import AbstractMPP
 from .postmaster import PostmasterProcess
 from .slots import SlotsHandler
@@ -95,7 +95,7 @@ class Postgresql(object):
         self.mpp_handler = mpp.get_handler_impl(self)
         self._bin_dir = config.get('bin_dir') or ''
         self._role_lock = Lock()
-        self.set_role('uninitialized')
+        self.set_role(PostgresqlRole.UNINITIALIZED)
         self.config = ConfigHandler(self, config)
         self.config.check_directories()
 
@@ -142,20 +142,20 @@ class Postgresql(object):
             # we know that PostgreSQL is accepting connections and can read some GUC's from pg_settings
             self.config.load_current_server_parameters()
 
-            self.set_role('primary' if self.is_primary() else 'replica')
+            self.set_role(PostgresqlRole.PRIMARY if self.is_primary() else PostgresqlRole.REPLICA)
 
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
 
-            if self.major_version < 120000 or self.role == 'primary':
+            if self.major_version < 120000 or self.role == PostgresqlRole.PRIMARY:
                 # If PostgreSQL is running as a primary or we run PostgreSQL that is older than 12 we can
                 # call reload_config() once again (the first call happened in the ConfigHandler constructor),
                 # so that it can figure out if config files should be updated and pg_ctl reload executed.
                 self.config.reload_config(config, sighup=bool(hba_saved or ident_saved))
             elif hba_saved or ident_saved:
                 self.reload()
-        elif not self.is_running() and self.role == 'primary':
-            self.set_role('demoted')
+        elif not self.is_running() and self.role == PostgresqlRole.PRIMARY:
+            self.set_role(PostgresqlRole.DEMOTED)
 
     @property
     def create_replica_methods(self) -> List[str]:
@@ -237,7 +237,7 @@ class Postgresql(object):
                          " pg_catalog.pg_stat_get_activity(w.pid)"
                          " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name)
                         if global_config.is_synchronous_mode
-                        and self.role in ('primary', 'promoted') else "'on', '', NULL")
+                        and self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED) else "'on', '', NULL")
 
         if self._major_version >= 90600:
             filter_failover = ' WHERE NOT failover' if self._major_version >= 170000 else ''
@@ -252,7 +252,7 @@ class Postgresql(object):
                            if self._major_version >= 130000 else "NULL")
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END, {0}, slot_name, "
                      "conninfo, status, {1} FROM pg_catalog.pg_stat_get_wal_receiver()").format(written_lsn, extra)
-            if self.role == 'standby_leader':
+            if self.role == PostgresqlRole.STANDBY_LEADER:
                 extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()"
             else:
                 extra = "0" + extra
@@ -366,13 +366,13 @@ class Postgresql(object):
             self._sysid = data.get('Database system identifier', '')
         return self._sysid
 
-    def get_postgres_role_from_data_directory(self) -> str:
+    def get_postgres_role_from_data_directory(self) -> PostgresqlRole:
         if self.data_directory_empty() or not self.controldata():
-            return 'uninitialized'
+            return PostgresqlRole.UNINITIALIZED
         elif self.config.recovery_conf_exists():
-            return 'replica'
+            return PostgresqlRole.REPLICA
         else:
-            return 'primary'
+            return PostgresqlRole.PRIMARY
 
     @property
     def server_version(self) -> int:
@@ -592,7 +592,7 @@ class Postgresql(object):
             return bool(self._cluster_info_state_get('timeline'))
         except PostgresConnectionException:
             logger.warning('Failed to determine PostgreSQL state from the connection, falling back to cached role')
-            return bool(self.is_running() and self.role == 'primary')
+            return bool(self.is_running() and self.role == PostgresqlRole.PRIMARY)
 
     def replay_paused(self) -> bool:
         return self._cluster_info_state_get('replay_paused') or False
@@ -694,7 +694,7 @@ class Postgresql(object):
 
         if self.callback and cb_type in self.callback:
             cmd = self.callback[cb_type]
-            role = 'primary' if self.role == 'promoted' else self.role
+            role = PostgresqlRole.PRIMARY if self.role == PostgresqlRole.PROMOTED else self.role
             try:
                 cmd = shlex.split(self.callback[cb_type]) + [cb_type, role, self.scope]
                 self._callback_executor.call(cmd)
@@ -702,11 +702,11 @@ class Postgresql(object):
                 logger.exception('callback %s %r %s %s failed', cmd, cb_type, role, self.scope)
 
     @property
-    def role(self) -> str:
+    def role(self) -> PostgresqlRole:
         with self._role_lock:
             return self._role
 
-    def set_role(self, value: str) -> None:
+    def set_role(self, value: PostgresqlRole) -> None:
         with self._role_lock:
             self._role = value
 
@@ -747,7 +747,7 @@ class Postgresql(object):
         return False
 
     def start(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None,
-              block_callbacks: bool = False, role: Optional[str] = None,
+              block_callbacks: bool = False, role: Optional[PostgresqlRole] = None,
               after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]:
         """Start PostgreSQL
 
@@ -1032,7 +1032,7 @@ class Postgresql(object):
         return self.state == PostgresqlState.RUNNING
 
     def restart(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None,
-                block_callbacks: bool = False, role: Optional[str] = None,
+                block_callbacks: bool = False, role: Optional[PostgresqlRole] = None,
                 before_shutdown: Optional[Callable[..., Any]] = None,
                 after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]:
         """Restarts PostgreSQL.
@@ -1140,14 +1140,15 @@ class Postgresql(object):
                 logger.exception('Failed to read and parse %s', (history_path,))
         return history
 
-    def follow(self, member: Union[Leader, Member, None], role: str = 'replica',
+    def follow(self, member: Union[Leader, Member, None], role: PostgresqlRole = PostgresqlRole.REPLICA,
                timeout: Optional[float] = None, do_reload: bool = False) -> Optional[bool]:
         """Reconfigure postgres to follow a new member or use different recovery parameters.
 
         Method may call `on_role_change` callback if role is changing.
 
         :param member: The member to follow
-        :param role: The desired role, normally 'replica', but could also be a 'standby_leader'
+        :param role: The desired role, one of :class:`~misc.PostgresqlRole` values, normally
+            :class:`~misc.PostgresqlRole.REPLICA`, but could also be a :class:`~misc.PostgresqlRole.STANDBY_LEADER`
         :param timeout: start timeout, how long should the `start()` method wait for postgres accepting connections
         :param do_reload: indicates that after updating postgresql.conf we just need to do a reload instead of restart
 
@@ -1165,8 +1166,9 @@ class Postgresql(object):
         # and we know for sure that postgres was already running before, we will only execute on_role_change
         # callback and prevent execution of on_restart/on_start callback.
         # If the role remains the same (replica or standby_leader), we will execute on_start or on_restart
-        change_role = self.cb_called and (self.role in ('primary', 'demoted')
-                                          or not {'standby_leader', 'replica'} - {self.role, role})
+        change_role = self.cb_called and \
+            (self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.DEMOTED)
+             or not {PostgresqlRole.STANDBY_LEADER, PostgresqlRole.REPLICA} - {self.role, role})
         if change_role:
             self.__cb_pending = CallbackAction.NOOP
 
@@ -1191,7 +1193,7 @@ class Postgresql(object):
         for _ in polling_loop(wait_seconds):
             data = self.controldata()
             if data.get('Database cluster state') == 'in production':
-                self.set_role('primary')
+                self.set_role(PostgresqlRole.PRIMARY)
                 return True
 
     def _pre_promote(self) -> bool:
@@ -1226,7 +1228,7 @@ class Postgresql(object):
 
     def promote(self, wait_seconds: int, task: CriticalTask,
                 before_promote: Optional[Callable[..., Any]] = None) -> Optional[bool]:
-        if self.role in ('promoted', 'primary'):
+        if self.role in (PostgresqlRole.PROMOTED, PostgresqlRole.PRIMARY):
             return True
 
         ret = self._pre_promote()
@@ -1250,7 +1252,7 @@ class Postgresql(object):
 
         ret = self.pg_ctl('promote', '-W')
         if ret:
-            self.set_role('promoted')
+            self.set_role(PostgresqlRole.PROMOTED)
             self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
             ret = self._wait_promote(wait_seconds)
         return ret
@@ -1358,7 +1360,7 @@ class Postgresql(object):
                 logger.exception("Could not rename data directory %s", self._data_dir)
 
     def remove_data_directory(self) -> None:
-        self.set_role('uninitialized')
+        self.set_role(PostgresqlRole.UNINITIALIZED)
         logger.info('Removing data directory: %s', self._data_dir)
         try:
             if os.path.islink(self._data_dir):

patroni/postgresql/config.py

@@ -20,7 +20,7 @@ from ..psycopg import parse_conninfo
 from ..utils import compare_values, get_postgres_version, is_subpath, \
     maybe_convert_from_base_unit, parse_bool, parse_int, split_host_port, uri, validate_directory
 from ..validator import EnumValidator, IntValidator
-from .misc import get_major_from_minor_version, postgres_version_to_int, PostgresqlState
+from .misc import get_major_from_minor_version, postgres_version_to_int, PostgresqlRole, PostgresqlState
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -1062,7 +1062,7 @@ class ConfigHandler(object):
             synchronous_standby_names = self._server_parameters.get('synchronous_standby_names')
             if synchronous_standby_names is None:
                 if global_config.is_synchronous_mode_strict\
-                        and self._postgresql.role in ('primary', 'promoted'):
+                        and self._postgresql.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED):
                     parameters['synchronous_standby_names'] = '*'
                 else:
                     parameters.pop('synchronous_standby_names', None)
@@ -1330,7 +1330,7 @@ class ConfigHandler(object):
         As a workaround we will start it with the values from controldata and set `pending_restart`
         to true as an indicator that current values of parameters are not matching expectations."""
 
-        if self._postgresql.role == 'primary':
+        if self._postgresql.role == PostgresqlRole.PRIMARY:
             return self._server_parameters
 
         options_mapping = {

patroni/postgresql/misc.py

@@ -34,6 +34,22 @@ class PostgresqlState(str, Enum):
         return self.value
 
 
+class PostgresqlRole(str, Enum):
+    """Possible values of :attr:`Postgresql.role`."""
+
+    PRIMARY = 'primary'
+    MASTER = 'master'
+    STANDBY_LEADER = 'standby_leader'
+    REPLICA = 'replica'
+    DEMOTED = 'demoted'
+    UNINITIALIZED = 'uninitialized'
+    PROMOTED = 'promoted'
+
+    def __repr__(self) -> str:
+        """Get a string representation of a :class:`PostgresqlRole` member."""
+        return self.value
+
+
 def postgres_version_to_int(pg_version: str) -> int:
     """Convert the server_version to integer
 

patroni/postgresql/mpp/citus.py

@@ -9,7 +9,7 @@ from urllib.parse import urlparse
 from ...dcs import Cluster
 from ...psycopg import connect, ProgrammingError, quote_ident
 from ...utils import parse_int
-from ..misc import PostgresqlState
+from ..misc import PostgresqlRole, PostgresqlState
 from . import AbstractMPP, AbstractMPPHandler
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -76,7 +76,7 @@ class PgDistNode:
 
         :returns: ``True`` if this object represents the ``primary``.
         """
-        return self.role in ('primary', 'demoted')
+        return self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.DEMOTED)
 
     def as_tuple(self, include_nodeid: bool = False) -> Tuple[str, int, str, Optional[int]]:
         """Helper method to compare two :class:`PgDistGroup` objects.
@@ -478,7 +478,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         for groupid, worker in cluster.workers.items():
             leader = worker.leader
             if leader and leader.conn_url\
-                    and leader.data.get('role') in ('master', 'primary')\
+                    and leader.data.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY)\
                     and leader.data.get('state') == PostgresqlState.RUNNING:
                 self.add_task('after_promote', groupid, worker, leader.name, leader.conn_url)
 

patroni/postgresql/rewind.py

@@ -14,7 +14,7 @@ from ..collections import EMPTY_DICT
 from ..dcs import Leader, RemoteMember
 from . import Postgresql
 from .connection import get_connection_cursor
-from .misc import format_lsn, fsync_dir, parse_history, parse_lsn
+from .misc import format_lsn, fsync_dir, parse_history, parse_lsn, PostgresqlRole
 
 logger = logging.getLogger(__name__)
 
@@ -223,7 +223,8 @@ class Rewind(object):
         if local_timeline is None or local_lsn is None:
             return
 
-        if isinstance(leader, Leader) and leader.member.data.get('role') not in ('master', 'primary'):
+        if isinstance(leader, Leader) and leader.member.data.get('role') not in (PostgresqlRole.MASTER,
+                                                                                 PostgresqlRole.PRIMARY):
             return
 
         # We want to use replication credentials when connecting to the "postgres" database in case if

patroni/postgresql/slots.py

@@ -18,7 +18,7 @@ from ..file_perm import pg_perm
 from ..psycopg import OperationalError
 from ..tags import Tags
 from .connection import get_connection_cursor
-from .misc import format_lsn, fsync_dir
+from .misc import format_lsn, fsync_dir, PostgresqlRole
 
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
@@ -693,7 +693,7 @@ class SlotsHandler:
         leader = cluster.leader
         if not leader:
             return
-        slots = cluster.get_replication_slots(self._postgresql, tags, role='replica')
+        slots = cluster.get_replication_slots(self._postgresql, tags, role=PostgresqlRole.REPLICA)
         copy_slots: Dict[str, Dict[str, Any]] = {}
         with self._get_leader_connection_cursor(leader) as cur:
             try:

tests/test_api.py

@@ -14,7 +14,7 @@ from patroni.dcs import ClusterConfig, Member
 from patroni.exceptions import PostgresConnectionException
 from patroni.ha import _MemberStatus
 from patroni.postgresql.config import get_param_diff
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 from patroni.psycopg import OperationalError
 from patroni.utils import RetryFailedError, tzutc
 
@@ -51,7 +51,7 @@ class MockPostgresql:
     connection_pool = MockConnectionPool()
     name = 'test'
     state = PostgresqlState.RUNNING
-    role = 'primary'
+    role = PostgresqlRole.PRIMARY
     server_version = 90625
     major_version = 90600
     sysid = 'dummysysid'
@@ -214,21 +214,22 @@ class TestRestApiHandler(unittest.TestCase):
         MockRestApiServer(RestApiHandler, 'GET /read-only')
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})):
             MockRestApiServer(RestApiHandler, 'GET /replica')
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})):
             MockRestApiServer(RestApiHandler, 'GET /replica')
         with patch.object(RestApiHandler, 'get_postgresql_status',
                           Mock(return_value={'state': PostgresqlState.RUNNING})):
             MockRestApiServer(RestApiHandler, 'GET /health')
         MockRestApiServer(RestApiHandler, 'GET /leader')
         with patch.object(RestApiHandler, 'get_postgresql_status',
-                          Mock(return_value={'role': 'replica', 'sync_standby': True})):
+                          Mock(return_value={'role': PostgresqlRole.REPLICA, 'sync_standby': True})):
             MockRestApiServer(RestApiHandler, 'GET /synchronous')
             MockRestApiServer(RestApiHandler, 'GET /read-only-sync')
         with patch.object(RestApiHandler, 'get_postgresql_status',
-                          Mock(return_value={'role': 'replica', 'quorum_standby': True})):
+                          Mock(return_value={'role': PostgresqlRole.REPLICA, 'quorum_standby': True})):
             MockRestApiServer(RestApiHandler, 'GET /quorum')
             MockRestApiServer(RestApiHandler, 'GET /read-only-quorum')
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status',
+                          Mock(return_value={'role': PostgresqlRole.REPLICA})):
             MockRestApiServer(RestApiHandler, 'GET /asynchronous')
         with patch.object(MockHa, 'is_leader', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /replica')
@@ -237,7 +238,7 @@ class TestRestApiHandler(unittest.TestCase):
             with patch.object(global_config.__class__, 'is_standby_cluster', Mock(return_value=True)):
                 MockRestApiServer(RestApiHandler, 'GET /standby_leader')
         MockPatroni.dcs.cluster = None
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})):
             MockRestApiServer(RestApiHandler, 'GET /primary')
         with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /primary')
@@ -264,7 +265,7 @@ class TestRestApiHandler(unittest.TestCase):
                                           'tag_key1=true&tag_key2=false&'
                                           'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
         #
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})):
             MockRestApiServer(RestApiHandler, 'GET /primary?lag=1M&'
                                               'tag_key1=true&tag_key2=false&'
                                               'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
@@ -278,7 +279,8 @@ class TestRestApiHandler(unittest.TestCase):
                                               'tag_key1=true&tag_key2=false&'
                                               'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
         #
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'standby_leader'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status',
+                          Mock(return_value={'role': PostgresqlRole.STANDBY_LEADER})):
             MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
                                               'tag_key1=true&tag_key2=false&'
                                               'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
@@ -305,7 +307,7 @@ class TestRestApiHandler(unittest.TestCase):
                                           'tag_key1=true&tag_key2=false&'
                                           'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
         #
-        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})):
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})):
             MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
                                               'tag_key1=true&tag_key2=false&'
                                               'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
@@ -461,27 +463,29 @@ class TestRestApiHandler(unittest.TestCase):
         request = make_request(schedule=future_restart_time.isoformat(), role='unknown', postgres_version='9.5.3')
         MockRestApiServer(RestApiHandler, request)
         # wrong version
-        request = make_request(schedule=future_restart_time.isoformat(), role='primary', postgres_version='9.5.3.1')
+        request = make_request(schedule=future_restart_time.isoformat(),
+                               role=PostgresqlRole.PRIMARY, postgres_version='9.5.3.1')
         MockRestApiServer(RestApiHandler, request)
         # unknown filter
         request = make_request(schedule=future_restart_time.isoformat(), batman='lives')
         MockRestApiServer(RestApiHandler, request)
         # incorrect schedule
-        request = make_request(schedule='2016-08-42 12:45TZ+1', role='primary')
+        request = make_request(schedule='2016-08-42 12:45TZ+1', role=PostgresqlRole.PRIMARY)
         MockRestApiServer(RestApiHandler, request)
         # everything fine, but the schedule is missing
-        request = make_request(role='primary', postgres_version='9.5.2')
+        request = make_request(role=PostgresqlRole.PRIMARY, postgres_version='9.5.2')
         MockRestApiServer(RestApiHandler, request)
         for retval in (True, False):
             with patch.object(MockHa, 'schedule_future_restart', Mock(return_value=retval)):
                 request = make_request(schedule=future_restart_time.isoformat())
                 MockRestApiServer(RestApiHandler, request)
             with patch.object(MockHa, 'restart', Mock(return_value=(retval, "foo"))):
-                request = make_request(role='primary', postgres_version='9.5.2')
+                request = make_request(role=PostgresqlRole.PRIMARY, postgres_version='9.5.2')
                 MockRestApiServer(RestApiHandler, request)
 
         with patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)):
-            MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='primary'))
+            MockRestApiServer(RestApiHandler,
+                              make_request(schedule='2016-08-42 12:45TZ+1', role=PostgresqlRole.PRIMARY))
             # Valid timeout
             MockRestApiServer(RestApiHandler, make_request(timeout='60s'))
             # Invalid timeout

tests/test_callback_executor.py

@@ -1,10 +1,12 @@
 import unittest
 
+from unittest import mock
 from unittest.mock import Mock, patch
 
 import psutil
 
-from patroni.postgresql.callback_executor import CallbackExecutor
+from patroni.postgresql.callback_executor import CallbackAction, CallbackExecutor
+from patroni.postgresql.misc import PostgresqlRole
 
 
 class TestCallbackExecutor(unittest.TestCase):
@@ -14,7 +16,7 @@ class TestCallbackExecutor(unittest.TestCase):
         mock_popen.return_value.children.return_value = []
         mock_popen.return_value.is_running.return_value = True
 
-        callback = ['test.sh', 'on_start', 'replica', 'foo']
+        callback = ['test.sh', CallbackAction.ON_START, PostgresqlRole.REPLICA, 'foo']
         ce = CallbackExecutor()
         ce._kill_children = Mock(side_effect=Exception)
         ce._invoke_excepthook = Mock()
@@ -22,6 +24,8 @@ class TestCallbackExecutor(unittest.TestCase):
         ce.join()
 
         self.assertIsNone(ce.call(callback))
+        mock_popen.assert_called_with(['test.sh', 'on_start', 'replica', 'foo'], close_fds=True)
+        mock_popen.reset_mock()
 
         mock_popen.return_value.kill.side_effect = psutil.AccessDenied()
         self.assertIsNone(ce.call(callback))
@@ -38,5 +42,7 @@ class TestCallbackExecutor(unittest.TestCase):
         self.assertIsNone(ce.call(callback))
 
         mock_popen.side_effect = [Mock()]
-        self.assertIsNone(ce.call(['test.sh', 'on_reload', 'replica', 'foo']))
+        self.assertIsNone(ce.call(['test.sh', CallbackAction.ON_RELOAD, PostgresqlRole.REPLICA, 'foo']))
+        self.assertEqual(mock_popen.call_args_list[-2],
+                         mock.call(['test.sh', 'on_reload', 'replica', 'foo'], close_fds=True))
         ce.join()

tests/test_consul.py

@@ -9,7 +9,7 @@ from consul import ConsulException, NotFound
 from patroni.dcs import get_dcs
 from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulClient, ConsulError, \
     ConsulInternalError, HTTPClient, InvalidSession, InvalidSessionTTL, RetryFailedError
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 from patroni.postgresql.mpp import get_mpp
 
 from . import SleepException
@@ -257,7 +257,8 @@ class TestConsul(unittest.TestCase):
     @patch.object(Agent.Service, 'register', Mock(side_effect=(False, True, True, True)))
     @patch.object(Agent.Service, 'deregister', Mock(return_value=True))
     def test_update_service(self):
-        d = {'role': 'replica', 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', 'state': PostgresqlState.RUNNING}
+        d = {'role': PostgresqlRole.REPLICA, 'api_url': 'http://a/t', 'conn_url': 'pg://c:1',
+             'state': PostgresqlState.RUNNING}
         self.assertIsNone(self.c.update_service({}, {}))
         self.assertFalse(self.c.update_service({}, d))
         self.assertTrue(self.c.update_service(d, d))
@@ -269,7 +270,7 @@ class TestConsul(unittest.TestCase):
         d['state'] = PostgresqlState.RUNNING
         d['role'] = 'bla'
         self.assertIsNone(self.c.update_service({}, d))
-        d['role'] = 'primary'
+        d['role'] = PostgresqlRole.PRIMARY
         self.assertTrue(self.c.update_service({}, d))
 
     @patch.object(KV, 'put', Mock(side_effect=ConsulException))
@@ -281,7 +282,7 @@ class TestConsul(unittest.TestCase):
 
         self.c.refresh_session = Mock(return_value=False)
 
-        d = {'role': 'replica', 'api_url': 'http://a/t',
+        d = {'role': PostgresqlRole.REPLICA, 'api_url': 'http://a/t',
              'conn_url': 'pg://c:1', 'state': PostgresqlState.RUNNING}
 
         # Changing register_service from True to False calls deregister()

tests/test_ctl.py

@@ -11,6 +11,7 @@ import etcd
 from click.testing import CliRunner
 from prettytable import PrettyTable
 
+from patroni.ctl import CtlPostgresqlRole
 from patroni.postgresql.misc import PostgresqlState
 
 try:
@@ -52,7 +53,7 @@ def get_default_config(*args):
 @patch('patroni.ctl.load_config', get_default_config)
 @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
 class TestCtl(unittest.TestCase):
-    TEST_ROLES = ('primary', 'leader')
+    TEST_ROLES = (CtlPostgresqlRole.PRIMARY, CtlPostgresqlRole.LEADER)
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     def setUp(self):
@@ -89,14 +90,16 @@ class TestCtl(unittest.TestCase):
                 self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role))
 
             # MockCursor returns pg_is_in_recovery as false
-            self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role='replica'))
+            self.assertIsNone(
+                get_cursor(get_cluster_initialized_with_leader(), None, {}, role=CtlPostgresqlRole.REPLICA))
 
-            self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role='any'))
+            self.assertIsNotNone(
+                get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role=CtlPostgresqlRole.ANY))
 
             # Mutually exclusive options
             with self.assertRaises(PatroniCtlException) as e:
                 get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, member_name='other',
-                           role='replica')
+                           role=CtlPostgresqlRole.REPLICA)
 
             self.assertEqual(str(e.exception), '--role and --member are mutually exclusive options')
 
@@ -301,7 +304,8 @@ class TestCtl(unittest.TestCase):
                 self.assertTrue('False' in str(rows))
 
             with patch.object(MockCursor, 'execute', Mock(side_effect=OperationalError('bla'))):
-                rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
+                rows = query_member(None, None, None, None,
+                                    CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {})
 
         with patch('patroni.ctl.get_cursor', Mock(return_value=None)):
             # No role nor member given -- generic message
@@ -313,11 +317,13 @@ class TestCtl(unittest.TestCase):
             self.assertTrue('No connection to member foo' in str(rows))
 
             # Role given -- message pointing to role
-            rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
+            rows = query_member(None, None, None, None,
+                                CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {})
             self.assertTrue('No connection to role replica' in str(rows))
 
         with patch('patroni.ctl.get_cursor', Mock(side_effect=OperationalError('bla'))):
-            rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
+            rows = query_member(None, None, None, None,
+                                CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {})
 
     def test_dsn(self):
         result = self.runner.invoke(ctl, ['dsn', 'alpha'])
@@ -471,12 +477,12 @@ class TestCtl(unittest.TestCase):
                 self.assertEqual(len(r), 1)
                 self.assertEqual(r[0].name, 'leader')
 
-            r = list(get_all_members(get_cluster_initialized_with_leader(), None, role='replica'))
+            r = list(get_all_members(get_cluster_initialized_with_leader(), None, role=CtlPostgresqlRole.REPLICA))
             self.assertEqual(len(r), 1)
             self.assertEqual(r[0].name, 'other')
 
             self.assertEqual(len(list(get_all_members(get_cluster_initialized_without_leader(),
-                                                      None, role='replica'))), 2)
+                                                      None, role=CtlPostgresqlRole.REPLICA))), 2)
 
     def test_members(self):
         result = self.runner.invoke(ctl, ['list'])

tests/test_ha.py

@@ -18,7 +18,7 @@ from patroni.postgresql.bootstrap import Bootstrap
 from patroni.postgresql.callback_executor import CallbackAction
 from patroni.postgresql.cancellable import CancellableSubprocess
 from patroni.postgresql.config import ConfigHandler
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.rewind import Rewind
 from patroni.postgresql.slots import SlotsHandler
@@ -59,7 +59,7 @@ def get_cluster_bootstrapping_without_leader(cluster_config=None):
 def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False):
     m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
                                   'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4,
-                                  'role': 'primary', 'state': 'running'})
+                                  'role': PostgresqlRole.PRIMARY, 'state': 'running'})
     leader = Leader(0, 0, m1 if leader else Member(0, '', 28, {}))
     m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
                                  'api_url': 'http://127.0.0.1:8011/patroni',
@@ -214,7 +214,7 @@ class TestHa(PostgresInit):
     def setUp(self):
         super(TestHa, self).setUp()
         self.p.set_state(PostgresqlState.RUNNING)
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time))
         self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
         self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
@@ -242,9 +242,9 @@ class TestHa(PostgresInit):
         with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')):
             self.ha.touch_member()
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
-        self.p.set_role('standby_leader')
+        self.p.set_role(PostgresqlRole.STANDBY_LEADER)
         self.ha.touch_member()
-        self.p.set_role('primary')
+        self.p.set_role(PostgresqlRole.PRIMARY)
         self.ha.dcs.touch_member = true
         self.ha.touch_member()
 
@@ -299,7 +299,7 @@ class TestHa(PostgresInit):
         self.p.follow = false
         self.p.is_running = false
         self.p.name = 'leader'
-        self.p.set_role('demoted')
+        self.p.set_role(PostgresqlRole.DEMOTED)
         self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID}
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
@@ -308,7 +308,7 @@ class TestHa(PostgresInit):
         self.p.start = false
         self.p.is_running = false
         self.p.name = 'leader'
-        self.p.set_role('primary')
+        self.p.set_role(PostgresqlRole.PRIMARY)
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'starting primary after failure')
@@ -346,7 +346,7 @@ class TestHa(PostgresInit):
     def test_recover_with_rewind(self):
         self.p.is_running = false
         self.ha.cluster = get_cluster_initialized_with_leader()
-        self.ha.cluster.leader.member.data.update(version='2.0.2', role='primary')
+        self.ha.cluster.leader.member.data.update(version='2.0.2', role=PostgresqlRole.PRIMARY)
         self.ha._rewind.pg_rewind = true
         self.ha._rewind.check_leader_is_not_in_recovery = true
         with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)):
@@ -414,7 +414,7 @@ class TestHa(PostgresInit):
     def test_long_promote(self):
         self.ha.has_lock = true
         self.p.is_primary = false
-        self.p.set_role('primary')
+        self.p.set_role(PostgresqlRole.PRIMARY)
         self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     def test_demote_after_failing_to_obtain_lock(self):
@@ -579,7 +579,7 @@ class TestHa(PostgresInit):
 
     def test_update_failsafe(self):
         self.assertRaises(Exception, self.ha.update_failsafe, {})
-        self.p.set_role('primary')
+        self.p.set_role(PostgresqlRole.PRIMARY)
         self.assertEqual(self.ha.update_failsafe({}), 'Running as a leader')
 
     def test_call_failsafe_member(self):
@@ -715,7 +715,7 @@ class TestHa(PostgresInit):
             self.assertEqual(self.ha.run_cycle(), 'updated leader lock during restart')
 
             self.ha.update_lock = false
-            self.p.set_role('primary')
+            self.p.set_role(PostgresqlRole.PRIMARY)
             with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \
                     patch('patroni.async_executor.CriticalTask.result',
                           PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \
@@ -907,7 +907,7 @@ class TestHa(PostgresInit):
 
     def test_manual_failover_process_no_leader(self):
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
 
         # failover to another member, fetch_node_status for candidate fails
         with patch('patroni.ha.logger.warning') as mock_warning:
@@ -917,7 +917,7 @@ class TestHa(PostgresInit):
                              ('%s: member %s is %s', 'manual failover', 'leader', 'not reachable'))
 
         # failover to another member, candidate is accessible, in_recovery
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.fetch_node_status = get_node_status()
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
@@ -928,7 +928,7 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
         # failover to me but I am set to nofailover. In no case I should be elected as a leader
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None))
         self.ha.patroni.nofailover = True
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
@@ -952,7 +952,7 @@ class TestHa(PostgresInit):
 
     def test_manual_switchover_process_no_leader(self):
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
 
         # I was the leader, other members are healthy
         self.ha.fetch_node_status = get_node_status()
@@ -989,7 +989,7 @@ class TestHa(PostgresInit):
 
         # manual failover to our node (postgresql0),
         # which name is not in sync nodes list (some sync nodes are available)
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
                                                                  sync=('leader1', 'other'))
         self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']),
@@ -1045,7 +1045,7 @@ class TestHa(PostgresInit):
 
         # switchover to me, I am not leader
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None))
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock')
 
@@ -1117,9 +1117,9 @@ class TestHa(PostgresInit):
     def test_post_recover(self, mock_call_nowait):
         self.p.is_running = false
         self.ha.has_lock = true
-        self.p.set_role('primary')
+        self.p.set_role(PostgresqlRole.PRIMARY)
         self.assertEqual(self.ha.post_recover(), 'removed leader key after trying and failing to start postgres')
-        self.assertEqual(self.p.role, 'demoted')
+        self.assertEqual(self.p.role, PostgresqlRole.DEMOTED)
         mock_call_nowait.assert_called_once_with(CallbackAction.ON_ROLE_CHANGE)
         self.ha.has_lock = false
         self.assertEqual(self.ha.post_recover(), 'failed to start postgres')
@@ -1167,7 +1167,7 @@ class TestHa(PostgresInit):
             self.assertEqual(self.ha.run_cycle(), "restart scheduled")
 
     def test_restart_matches(self):
-        self.p._role = 'replica'
+        self.p._role = PostgresqlRole.REPLICA
         self.p._connection.server_version = 90500
         self.p._pending_restart = True
         self.assertFalse(self.ha.restart_matches("primary", "9.5.0", True))
@@ -1195,7 +1195,7 @@ class TestHa(PostgresInit):
         self.ha._leader_timeline = 1
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
         self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader), the standby leader with the lock')
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
 
@@ -1221,7 +1221,7 @@ class TestHa(PostgresInit):
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
     def test_process_unhealthy_standby_cluster_as_cascade_replica(self):
         self.p.is_primary = false
-        self.p.name = 'replica'
+        self.p.name = PostgresqlRole.REPLICA
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_member:'))
 
@@ -1458,7 +1458,7 @@ class TestHa(PostgresInit):
 
         mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.has_lock = true
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         self.p.name = 'leader'
@@ -1472,7 +1472,7 @@ class TestHa(PostgresInit):
         mock_set_sync.reset_mock()
 
         # When we just became primary nobody is sync
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         mock_write_sync.return_value = False
         self.assertTrue(self.ha.enforce_primary_role('msg', 'promote msg') != 'promote msg')
         mock_set_sync.assert_not_called()
@@ -1481,7 +1481,7 @@ class TestHa(PostgresInit):
         self.ha.is_synchronous_mode = true
 
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.p.name = 'other'
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'other2'))
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
@@ -1512,7 +1512,7 @@ class TestHa(PostgresInit):
 
         self.p.name = 'other'
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         mock_restart = self.p.restart = Mock(return_value=True)
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
         self.ha.touch_member = Mock(return_value=True)
@@ -1627,7 +1627,7 @@ class TestHa(PostgresInit):
         self.p.data_directory_empty = Mock(side_effect=OSError(5, "Input/output error: '{}'".format(self.p.data_dir)))
         self.assertEqual(self.ha.run_cycle(),
                          'released leader key voluntarily as data dir not accessible and currently leader')
-        self.assertEqual(self.p.role, 'uninitialized')
+        self.assertEqual(self.p.role, PostgresqlRole.UNINITIALIZED)
 
         # as has_lock is mocked out, we need to fake the leader key release
         self.ha.has_lock = false
@@ -1651,7 +1651,7 @@ class TestHa(PostgresInit):
         self.p.is_primary = false
         self.ha.run_cycle()
         exit_mock.assert_called_once_with(1)
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.dcs.initialize = Mock()
         with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)):
             self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
@@ -1687,7 +1687,7 @@ class TestHa(PostgresInit):
         self.ha.is_paused = true
         self.p.data_directory_empty = true
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: running with empty data directory')
-        self.assertEqual(self.p.role, 'uninitialized')
+        self.assertEqual(self.p.role, PostgresqlRole.UNINITIALIZED)
 
     @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True))
     def test_sysid_no_match_in_pause(self):
@@ -1742,7 +1742,7 @@ class TestHa(PostgresInit):
         self.p._major_version = 90500
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('other', self.p.name + ',foo'))
         self.p.is_primary = false
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=None)
         # Postgres 9.5, write_sync_state to DCS failed
         self.assertEqual(self.ha.run_cycle(),
@@ -1764,7 +1764,7 @@ class TestHa(PostgresInit):
         self.p._major_version = 90600
         mock_set_sync.reset_mock()
         mock_write_sync.reset_mock()
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         # Postgres 9.6, with quorum commit we avoid updating /sync key and put some nodes to ssn
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
         self.assertEqual(mock_write_sync.call_count, 0)
@@ -1773,7 +1773,7 @@ class TestHa(PostgresInit):
 
         self.p._major_version = 150000
         mock_set_sync.reset_mock()
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.p.name = 'nonsync'
         self.ha.fetch_node_status = get_node_status()
         # Postgres 15, with quorum commit. Non-sync node promoted we avoid updating /sync key and put some nodes to ssn

tests/test_kubernetes.py

@@ -15,7 +15,7 @@ from patroni.dcs import get_dcs
 from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \
     K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, Retry, \
     RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 from patroni.postgresql.mpp import get_mpp
 
 from . import MockResponse, SleepException
@@ -322,19 +322,19 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         mock_patch_namespaced_pod.return_value.metadata.resource_version = '10'
 
         self.k._name = 'p-1'
-        self.k.touch_member({'role': 'replica', 'state': PostgresqlState.INITDB})
+        self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.INITDB})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar')
 
-        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'})
+        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None)
 
-        self.k.touch_member({'role': 'replica', 'state': PostgresqlState.CUSTOM_BOOTSTRAP})
+        self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.CUSTOM_BOOTSTRAP})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar')
 
-        self.k.touch_member({'role': 'replica', 'state': PostgresqlState.BOOTSTRAP_STARTING})
+        self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.BOOTSTRAP_STARTING})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar')
 
-        self.k.touch_member({'state': PostgresqlState.STOPPED, 'role': 'primary'})
+        self.k.touch_member({'state': PostgresqlState.STOPPED, 'role': PostgresqlRole.PRIMARY})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None)
 
         self.k._role_label = 'isMaster'
@@ -343,26 +343,26 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         self.k._standby_leader_label_value = 'false'
         self.k._tmp_role_label = 'tmp_role'
 
-        self.k.touch_member({'state': PostgresqlState.CREATING_REPLICA, 'role': 'replica'})
+        self.k.touch_member({'state': PostgresqlState.CREATING_REPLICA, 'role': PostgresqlRole.REPLICA})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar')
 
-        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'})
+        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA})
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None)
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false')
-        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'replica')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.REPLICA)
         mock_patch_namespaced_pod.rest_mock()
 
         self.k._name = 'p-0'
-        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'standby_leader'})
+        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.STANDBY_LEADER})
         mock_patch_namespaced_pod.assert_called()
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false')
-        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'primary')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.PRIMARY)
         mock_patch_namespaced_pod.rest_mock()
 
-        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'primary'})
+        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.PRIMARY})
         mock_patch_namespaced_pod.assert_called()
         self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true')
-        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'primary')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.PRIMARY)
 
     def test_initialize(self):
         self.k.initialize()
@@ -495,7 +495,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0])
         mock_logger_exception.reset_mock()
 
-        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'})
+        self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA})
         mock_logger_exception.assert_called_once()
         self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0])
 

tests/test_patroni.py

@@ -20,7 +20,7 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 
 from . import psycopg_connect, SleepException
 from .test_etcd import etcd_read, etcd_write
@@ -163,7 +163,7 @@ class TestPatroni(unittest.TestCase):
     @patch.object(Postgresql, 'state', PropertyMock(return_value=PostgresqlState.RUNNING))
     @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
     def test_run(self):
-        self.p.postgresql.set_role('replica')
+        self.p.postgresql.set_role(PostgresqlRole.REPLICA)
         self.p.sighup_handler()
         self.p.ha.dcs.watch = Mock(side_effect=SleepException)
         self.p.api.start = Mock()

tests/test_postgresql.py

@@ -22,7 +22,7 @@ from patroni.postgresql import PgIsReadyStatus, Postgresql
 from patroni.postgresql.bootstrap import Bootstrap
 from patroni.postgresql.callback_executor import CallbackAction
 from patroni.postgresql.config import _false_validator, get_param_diff
-from patroni.postgresql.misc import PostgresqlState
+from patroni.postgresql.misc import PostgresqlRole, PostgresqlState
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.validator import _get_postgres_guc_validators, _load_postgres_gucs_validators, \
     _read_postgres_gucs_validators_file, Bool, Enum, EnumBool, Integer, InvalidGucValidatorsFile, \
@@ -446,7 +446,7 @@ class TestPostgresql(BaseTestPostgresql):
         task = CriticalTask()
         self.assertTrue(self.p.promote(0, task))
 
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.p.config._config['pre_promote'] = 'test'
         with patch('patroni.postgresql.cancellable.CancellableSubprocess.is_cancelled', PropertyMock(return_value=1)):
             self.assertFalse(self.p.promote(0, task))
@@ -480,7 +480,7 @@ class TestPostgresql(BaseTestPostgresql):
 
     @patch('shlex.split', Mock(side_effect=OSError))
     def test_call_nowait(self):
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START))
         self.p.bootstrapping = True
         self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START))
@@ -508,7 +508,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('os.path.exists', Mock(return_value=True))
     @patch.object(Postgresql, 'controldata', Mock())
     def test_get_postgres_role_from_data_directory(self):
-        self.assertEqual(self.p.get_postgres_role_from_data_directory(), 'replica')
+        self.assertEqual(self.p.get_postgres_role_from_data_directory(), PostgresqlRole.REPLICA)
 
     @patch('os.remove', Mock())
     @patch('shutil.rmtree', Mock())
@@ -852,7 +852,8 @@ class TestPostgresql(BaseTestPostgresql):
     def test_get_primary_timeline(self):
         self.assertEqual(self.p.get_primary_timeline(), 1)
 
-    @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica'))
+    @patch.object(Postgresql, 'get_postgres_role_from_data_directory',
+                  Mock(return_value=PostgresqlRole.REPLICA))
     @patch.object(Postgresql, 'is_running', Mock(return_value=False))
     @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True))
     @patch('patroni.postgresql.config.logger')
@@ -893,7 +894,7 @@ class TestPostgresql(BaseTestPostgresql):
 
     @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError('')))
     def test_received_timeline(self):
-        self.p.set_role('standby_leader')
+        self.p.set_role(PostgresqlRole.STANDBY_LEADER)
         self.p.reset_cluster_info_state(None)
         self.assertRaises(PostgresConnectionException, self.p.received_timeline)
 

tests/test_slots.py

@@ -8,7 +8,7 @@ from unittest.mock import Mock, patch, PropertyMock
 from patroni import global_config, psycopg
 from patroni.dcs import Cluster, ClusterConfig, Member, Status, SyncState
 from patroni.postgresql import Postgresql
-from patroni.postgresql.misc import fsync_dir, PostgresqlState
+from patroni.postgresql.misc import fsync_dir, PostgresqlRole, PostgresqlState
 from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
 from patroni.tags import Tags
 
@@ -52,21 +52,22 @@ class TestSlotsHandler(BaseTestPostgresql):
         global_config.update(cluster)
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, self.tags)
-        self.p.set_role('standby_leader')
+        self.p.set_role(PostgresqlRole.STANDBY_LEADER)
         with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
                 patch.object(global_config.__class__, 'is_standby_cluster', PropertyMock(return_value=True)), \
                 patch('patroni.postgresql.slots.logger.debug') as mock_debug:
             self.s.sync_replication_slots(cluster, self.tags)
             mock_debug.assert_called_once()
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
             config.data['slots'].pop('ls')
             self.s.sync_replication_slots(cluster, self.tags)
             mock_drop.assert_not_called()
-        self.p.set_role('primary')
-        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
+        self.p.set_role(PostgresqlRole.PRIMARY)
+        with mock.patch('patroni.postgresql.Postgresql.role',
+                        new_callable=PropertyMock(return_value=PostgresqlRole.REPLICA)):
             self.s.sync_replication_slots(cluster, self.tags)
         with patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
             alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
@@ -79,7 +80,7 @@ class TestSlotsHandler(BaseTestPostgresql):
             self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
             with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
                 self.s.sync_replication_slots(cluster, self.tags)
-                self.p.set_role('replica')
+                self.p.set_role(PostgresqlRole.REPLICA)
                 self.s.sync_replication_slots(cluster, self.tags)
 
     def test_cascading_replica_sync_replication_slots(self):
@@ -91,7 +92,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         })
         cluster = Cluster(True, config, self.leader, Status(0, {'ls': 10}, []),
                           [self.me, self.other, self.leadermem, cascading_replica], None, SyncState.empty(), None, None)
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         with patch.object(Postgresql, '_query') as mock_query, \
                 patch.object(Postgresql, 'is_primary', Mock(return_value=False)):
             mock_query.return_value = [('ls', 'logical', 104, 'b', 'a', 5, 12345, 105)]
@@ -147,10 +148,10 @@ class TestSlotsHandler(BaseTestPostgresql):
         # sanity for primary
         self.p.name = self.leadermem.name
         self.assertEqual(
-            cluster._get_permanent_slots(self.p, self.leadermem, 'primary'),
+            cluster._get_permanent_slots(self.p, self.leadermem, PostgresqlRole.PRIMARY),
             {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}})
         self.assertEqual(
-            cluster._get_members_slots(self.p.name, 'primary', False, True),
+            cluster._get_members_slots(self.p.name, PostgresqlRole.PRIMARY, False, True),
             {'test_3': {'type': 'physical', 'lsn': 98, 'expected_active': False},
              'test_4': {'type': 'physical', 'lsn': 98, 'expected_active': True}})
 
@@ -158,7 +159,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.name = nostream_node.name
         # permanent logical slots are not allowed on nostream node
         self.assertEqual(
-            cluster._get_permanent_slots(self.p, nostream_node, 'replica'),
+            cluster._get_permanent_slots(self.p, nostream_node, PostgresqlRole.REPLICA),
             {'bar': {'type': 'physical'}})
         self.assertEqual(
             cluster.get_slot_name_on_primary(self.p.name, nostream_node),
@@ -166,7 +167,7 @@ class TestSlotsHandler(BaseTestPostgresql):
 
         # check cascade member-slot existence on nostream node
         self.assertEqual(
-            cluster._get_members_slots(nostream_node.name, 'replica', False, True),
+            cluster._get_members_slots(nostream_node.name, PostgresqlRole.REPLICA, False, True),
             {'leader': {'type': 'physical', 'lsn': 99, 'expected_active': False},
              'test_3': {'type': 'physical', 'lsn': 98, 'expected_active': True},
              'test_4': {'type': 'physical', 'lsn': 98, 'expected_active': False}})
@@ -174,7 +175,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         # cascade also does not entitled to have logical slot on itself ...
         self.p.name = cascade_node.name
         self.assertEqual(
-            cluster._get_permanent_slots(self.p, cascade_node, 'replica'),
+            cluster._get_permanent_slots(self.p, cascade_node, PostgresqlRole.REPLICA),
             {'bar': {'type': 'physical'}})
         # ... and member-slot on primary
         self.assertEqual(
@@ -184,7 +185,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         # simple replica must have every permanent slot ...
         self.p.name = stream_node.name
         self.assertEqual(
-            cluster._get_permanent_slots(self.p, stream_node, 'replica'),
+            cluster._get_permanent_slots(self.p, stream_node, PostgresqlRole.REPLICA),
             {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}})
         # ... and member-slot on primary
         self.assertEqual(
@@ -219,7 +220,7 @@ class TestSlotsHandler(BaseTestPostgresql):
 
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
-        self.p.set_role('replica')
+        self.p.set_role(PostgresqlRole.REPLICA)
         self.cluster.status.slots['ls'] = 12346
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
@@ -321,7 +322,7 @@ class TestSlotsHandler(BaseTestPostgresql):
             self.assertTrue(mock_query.call_args[0][0].startswith('WITH slots AS (SELECT slot_name, active'))
 
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
-    @patch.object(Postgresql, 'role', PropertyMock(return_value='replica'))
+    @patch.object(Postgresql, 'role', PropertyMock(return_value=PostgresqlRole.REPLICA))
     def test_advance_physical_slots(self):
         config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1)
         cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}, []),
@@ -363,7 +364,7 @@ class TestSlotsHandler(BaseTestPostgresql):
             mock_drop.assert_not_called()
 
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
-    @patch.object(Postgresql, 'role', PropertyMock(return_value='replica'))
+    @patch.object(Postgresql, 'role', PropertyMock(return_value=PostgresqlRole.REPLICA))
     @patch.object(TestTags, 'tags', PropertyMock(return_value={'nofailover': True}))
     def test_slots_nofailover_tag(self):
         self.p.name = self.leadermem.name