def get_postgresql_status(self, retry: bool = False) -> Dict[str, Any]:
        """Builds an object representing a status of "postgres".

        Some of the values are collected by executing a query and other are taken from the state stored in memory.

        :param retry: whether the query should be retried if failed or give up immediately

        :returns: a dict with the status of Postgres/Patroni. The keys are:

            * ``state``: one of :class:`~patroni.postgresql.misc.PostgresqlState` or ``unknown``;
            * ``postmaster_start_time``: ``pg_postmaster_start_time()``;
            * ``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 :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY`:

                    * ``location``: ``pg_current_wal_flush_lsn()``

                * 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)``;
                    * ``replayed_timestamp``: ``pg_last_xact_replay_timestamp``;
                    * ``paused``: ``pg_is_wal_replay_paused()``;

            * ``sync_standby``: ``True`` if replication mode is synchronous and this is a sync standby;
            * ``quorum_standby``: ``True`` if replication mode is quorum and this is a quorum standby;
            * ``timeline``: PostgreSQL primary node timeline;
            * ``replication``: :class:`list` of :class:`dict` entries, one for each replication connection. Each entry
                contains the following keys:

                * ``application_name``: ``pg_stat_activity.application_name``;
                * ``client_addr``: ``pg_stat_activity.client_addr``;
                * ``state``: ``pg_stat_replication.state``;
                * ``sync_priority``: ``pg_stat_replication.sync_priority``;
                * ``sync_state``: ``pg_stat_replication.sync_state``;
                * ``usename``: ``pg_stat_activity.usename``.

            * ``pause``: ``True`` if cluster is in maintenance mode;
            * ``cluster_unlocked``: ``True`` if cluster has no node holding the leader lock;
            * ``failsafe_mode_is_active``: ``True`` if DCS failsafe mode is currently active;
            * ``dcs_last_seen``: epoch timestamp DCS was last reached by Patroni.

        """
        postgresql = self.server.patroni.postgresql
        cluster = self.server.patroni.dcs.cluster
        config = global_config.from_cluster(cluster)
        try:

            if postgresql.state not in (PostgresqlState.RUNNING, PostgresqlState.RESTARTING,
                                        PostgresqlState.STARTING):
                raise RetryFailedError('')
            replication_state = ("pg_catalog.pg_{0}_{1}_diff(wr.latest_end_lsn, '0/0')::bigint, wr.status"
                                 if postgresql.major_version >= 90600 else "NULL, NULL") + ", " +\
                ("pg_catalog.current_setting('restore_command')" if postgresql.major_version >= 120000 else "NULL") +\
                ", " + ("pg_catalog.pg_wal_lsn_diff(wr.written_lsn, '0/0')::bigint"
                        if postgresql.major_version >= 130000 else "NULL")
            stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
                    " pg_catalog.pg_last_xact_replay_timestamp(), " + replication_state + ","
                    " (SELECT pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
                    "FROM (SELECT (SELECT rolname FROM pg_catalog.pg_authid WHERE oid = usesysid) AS usename,"
                    " application_name, client_addr, w.state, sync_state, sync_priority"
                    " FROM pg_catalog.pg_stat_get_wal_senders() w, pg_catalog.pg_stat_get_activity(pid)) AS ri)") +\
                (" FROM pg_catalog.pg_stat_get_wal_receiver() AS wr" if postgresql.major_version >= 90600 else "")

            row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name,
                                         postgresql.wal_flush), retry=retry)[0]
            result = {
                'state': postgresql.state,
                'postmaster_start_time': row[0],
                '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],
                    'replayed_location': row[3],
                    'replayed_timestamp': row[6],
                    'paused': row[5]} if row[1] == 0 else {
                    'location': row[2]
                })
            }

            if result['role'] == PostgresqlRole.REPLICA and config.is_standby_cluster:
                result['role'] = postgresql.role

            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

            if row[1] > 0:
                result['timeline'] = row[1]
            else:
                leader_timeline = None\
                    if not cluster or cluster.is_unlocked() or not cluster.leader else cluster.leader.timeline
                result['timeline'] = postgresql.replica_cached_timeline(leader_timeline)

            if row[7]:
                result['latest_end_lsn'] = row[7]

            replication_state = postgresql.replication_state_from_parameters(row[1] > 0, row[8], row[9])
            if replication_state:
                result['replication_state'] = replication_state

            if row[11]:
                result['replication'] = row[11]

        except (psycopg.Error, RetryFailedError, PostgresConnectionException):
            state = postgresql.state
            if state == PostgresqlState.RUNNING:
                logger.exception('get_postgresql_status')
                state = 'unknown'
            result: Dict[str, Any] = {'state': state, 'role': postgresql.role}

        if config.is_paused:
            result['pause'] = True
        if not cluster or cluster.is_unlocked():
            result['cluster_unlocked'] = True
        if self.server.patroni.ha.failsafe_is_active():
            result['failsafe_mode_is_active'] = True
        result['dcs_last_seen'] = self.server.patroni.dcs.last_seen
        return result
Builds an object representing a status of "postgres".

        Some of the values are collected by executing a query and other are taken from the state stored in memory.

        :param retry: whether the query should be retried if failed or give up immediately

        :returns: a dict with the status of Postgres/Patroni. The keys are:

            * ``state``: one of :class:`~patroni.postgresql.misc.PostgresqlState` or ``unknown``;
            * ``postmaster_start_time``: ``pg_postmaster_start_time()``;
            * ``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 :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY`:

                    * ``location``: ``pg_current_wal_flush_lsn()``

                * 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)``;
                    * ``replayed_timestamp``: ``pg_last_xact_replay_timestamp``;
                    * ``paused``: ``pg_is_wal_replay_paused()``;

            * ``sync_standby``: ``True`` if replication mode is synchronous and this is a sync standby;
            * ``quorum_standby``: ``True`` if replication mode is quorum and this is a quorum standby;
            * ``timeline``: PostgreSQL primary node timeline;
            * ``replication``: :class:`list` of :class:`dict` entries, one for each replication connection. Each entry
                contains the following keys:

                * ``application_name``: ``pg_stat_activity.application_name``;
                * ``client_addr``: ``pg_stat_activity.client_addr``;
                * ``state``: ``pg_stat_replication.state``;
                * ``sync_priority``: ``pg_stat_replication.sync_priority``;
                * ``sync_state``: ``pg_stat_replication.sync_state``;
                * ``usename``: ``pg_stat_activity.usename``.

            * ``pause``: ``True`` if cluster is in maintenance mode;
            * ``cluster_unlocked``: ``True`` if cluster has no node holding the leader lock;
            * ``failsafe_mode_is_active``: ``True`` if DCS failsafe mode is currently active;
            * ``dcs_last_seen``: epoch timestamp DCS was last reached by Patroni.

        
patroni/api.py