How to use the patroni.utils.uri function in patroni

cmd = config.get('post_bootstrap') or config.get('post_init')
        if cmd:
            r = self._postgresql.config.local_connect_kwargs

            if 'host' in r:
                # '/tmp' => '%2Ftmp' for unix socket path
                host = quote_plus(r['host']) if r['host'].startswith('/') else r['host']
            else:
                host = ''

                # https://www.postgresql.org/docs/current/static/libpq-pgpass.html
                # A host name of localhost matches both TCP (host name localhost) and Unix domain socket
                # (pghost empty or the default socket directory) connections coming from the local machine.
                r['host'] = 'localhost'  # set it to localhost to write into pgpass

            connstring = uri('postgres', (host, r['port']), r['database'], r.get('user'))
            env = self._postgresql.config.write_pgpass(r) if 'password' in r else None

            try:
                ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)
            except OSError:
                logger.error('post_init script %s failed', cmd)
                return False
            if ret != 0:
                logger.error('post_init script %s returned non-zero code %d', cmd, ret)
                return False
        return True

def _get_machines_cache_from_config(self):
        if 'proxy' in self._config:
            return [uri(self.protocol, (self._config['host'], self._config['port']))]

        machines_cache = []
        if 'srv' in self._config:
            machines_cache = self._get_machines_cache_from_srv(self._config['srv'])

        if not machines_cache and 'hosts' in self._config:
            machines_cache = list(self._config['hosts'])

        if not machines_cache and 'host' in self._config:
            machines_cache = self._get_machines_cache_from_dns(self._config['host'], self._config['port'])
        return machines_cache

def _query_exhibitors(self, exhibitors):
        random.shuffle(exhibitors)
        for host in exhibitors:
            try:
                response = requests.get(uri('http', (host, self._exhibitor_port), self._uri_path), timeout=self.TIMEOUT)
                return response.json()
            except RequestException:
                pass
        return None

def conn_url(self):
        conn_url = self.data.get('conn_url')
        conn_kwargs = self.data.get('conn_kwargs')
        if conn_url:
            return conn_url

        if conn_kwargs:
            conn_url = uri('postgresql', (conn_kwargs.get('host'), conn_kwargs.get('port', 5432)))
            self.data['conn_url'] = conn_url
            return conn_url

if self._config.get('use_unix_socket'):
            unix_socket_directories = self._server_parameters.get('unix_socket_directories')
            if unix_socket_directories is not None:
                # fallback to tcp if unix_socket_directories is set, but there are no sutable values
                local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address

            # if unix_socket_directories is not specified, but use_unix_socket is set to true - do our best
            # to use default value, i.e. don't specify a host neither in connection url nor arguments
        else:
            local_address['host'] = tcp_local_address

        self._local_address = local_address
        self.local_replication_address = {'host': tcp_local_address, 'port': port}

        netloc = self._config.get('connect_address') or tcp_local_address + ':' + port
        self._postgresql.connection_string = uri('postgres', netloc, self._postgresql.database)

        self._postgresql.set_connection_kwargs(self.local_connect_kwargs)

def _get_machines_cache_from_dns(self, host, port):
        """One host might be resolved into multiple ip addresses. We will make list out of it"""
        if self.protocol == 'http':
            ret = map(lambda res: uri(self.protocol, res[-1][:2]), self._dns_resolver.resolve(host, port))
            if ret:
                return list(set(ret))
        return [uri(self.protocol, (host, port))]

"""

        self._postgresql.set_state('creating replica')
        self._postgresql.schedule_sanity_checks_after_pause()

        is_remote_master = isinstance(clone_member, RemoteMember)

        # get list of replica methods either from clone member or from
        # the config. If there is no configuration key, or no value is
        # specified, use basebackup
        replica_methods = (clone_member.create_replica_methods if is_remote_master
                           else self._postgresql.create_replica_methods) or ['basebackup']

        if clone_member and clone_member.conn_url:
            r = clone_member.conn_kwargs(self._postgresql.config.replication)
            connstring = uri('postgres', (r['host'], r['port']), r['database'], r['user'])
            # add the credentials to connect to the replica origin to pgpass.
            env = self._postgresql.config.write_pgpass(r)
        else:
            connstring = ''
            env = os.environ.copy()
            # if we don't have any source, leave only replica methods that work without it
            replica_methods = [r for r in replica_methods
                               if self._postgresql.replica_method_can_work_without_replication_connection(r)]

        # go through them in priority order
        ret = 1
        for replica_method in replica_methods:
            if self._postgresql.cancellable.is_cancelled:
                break

            method_config = self._postgresql.replica_method_options(replica_method)