def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[int], candidate: Optional[str],
                               force: bool, switchover_leader: Optional[str] = None,
                               switchover_scheduled: Optional[str] = None) -> None:
    """Perform a failover or a switchover operation in the cluster.

    Informational messages are printed in the console during the operation, as well as the list of members before and
    after the operation, so the user can follow the operation status.

    .. note::
        If not able to perform the operation through the REST API, write directly to the DCS as a fall back.

    :param action: action to be taken -- ``failover`` or ``switchover``.
    :param cluster_name: name of the Patroni cluster.
    :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be
        prompted for filling it -- unless *force* is ``True``, in which case an exception is raised.
    :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used.
    :param force: perform the failover or switchover without asking for confirmations.
    :param switchover_leader: name of the leader passed to the switchover command if any.
    :param switchover_scheduled: timestamp when the switchover should be scheduled to occur. If ``now``,
        perform immediately.

    :raises:
        :class:`PatroniCtlException`: if:
            * Patroni is running on a Citus cluster, but no *group* was specified; or
            * a switchover was requested by the cluster has no leader; or
            * *switchover_leader* does not match the current leader of the cluster; or
            * cluster has no candidates available for the operation; or
            * no *candidate* is given for a failover operation; or
            * current leader and *candidate* are the same; or
            * *candidate* is tagged as nofailover; or
            * *candidate* is not a member of the cluster; or
            * trying to schedule a switchover in a cluster that is in maintenance mode; or
            * user aborts the operation.
    """
    dcs = get_dcs(cluster_name, group)
    cluster = dcs.get_cluster()
    click.echo('Current cluster topology')
    output_members(cluster, cluster_name, group=group)

    if is_citus_cluster() and group is None:
        if force:
            raise PatroniCtlException('For Citus clusters the --group must me specified')
        else:
            group = click.prompt('Citus group', type=int)
            dcs = get_dcs(cluster_name, group)
            cluster = dcs.get_cluster()

    config = global_config.from_cluster(cluster)

    cluster_leader = cluster.leader and cluster.leader.name
    # leader has to be be defined for switchover only
    if action == 'switchover':
        if not cluster_leader:
            raise PatroniCtlException('This cluster has no leader')

        if switchover_leader is None:
            if force:
                switchover_leader = cluster_leader
            else:
                prompt = 'Standby Leader' if config.is_standby_cluster else 'Primary'
                switchover_leader = click.prompt(prompt, type=str, default=cluster_leader)

        if cluster_leader != switchover_leader:
            raise PatroniCtlException(f'Member {switchover_leader} is not the leader of cluster {cluster_name}')

    # excluding members with nofailover tag
    candidate_names = [str(m.name) for m in cluster.members if m.name != cluster_leader and not m.nofailover]
    # We sort the names for consistent output to the client
    candidate_names.sort()

    if not candidate_names:
        raise PatroniCtlException('No candidates found to {0} to'.format(action))

    if candidate is None and not force:
        candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='')

    if action == 'failover' and not candidate:
        raise PatroniCtlException('Failover could be performed only to a specific candidate')

    if candidate and candidate not in candidate_names:
        if candidate == cluster_leader:
            raise PatroniCtlException(
                f'Member {candidate} is already the leader of cluster {cluster_name}')
        raise PatroniCtlException(
            f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')

    if all((not force,
            action == 'failover',
            config.is_synchronous_mode,
            not cluster.sync.is_empty,
            not cluster.sync.matches(candidate, True))):
        if not click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}?'):
            raise PatroniCtlException('Aborting ' + action)

    scheduled_at_str = None
    scheduled_at = None

    if action == 'switchover':
        if switchover_scheduled is None and not force:
            next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
            switchover_scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ' ) ',
                                                type=str, default='now')

        scheduled_at = parse_scheduled(switchover_scheduled)
        if scheduled_at:
            if config.is_paused:
                raise PatroniCtlException("Can't schedule switchover in the paused state")
            scheduled_at_str = scheduled_at.isoformat()

    failover_value = {'candidate': candidate}
    if action == 'switchover':
        failover_value['leader'] = switchover_leader
    if scheduled_at_str:
        failover_value['scheduled_at'] = scheduled_at_str

    logging.debug(failover_value)

    # By now we have established that the leader exists and the candidate exists
    if not force:
        demote_msg = f', demoting current leader {cluster_leader}' if cluster_leader else ''
        if scheduled_at_str:
            # only switchover can be scheduled
            if not click.confirm(f'Are you sure you want to schedule switchover of cluster '
                                 f'{cluster_name} at {scheduled_at_str}{demote_msg}?'):
                # action as a var to catch a regression in the tests
                raise PatroniCtlException('Aborting scheduled ' + action)
        else:
            if not click.confirm(f'Are you sure you want to {action} cluster {cluster_name}{demote_msg}?'):
                raise PatroniCtlException('Aborting ' + action)

    r = None
    try:
        member = cluster.leader.member if cluster.leader else candidate and cluster.get_member(candidate, False)
        if TYPE_CHECKING:  # pragma: no cover
            assert isinstance(member, Member)
        r = request_patroni(member, 'post', action, failover_value)

        # probably old patroni, which doesn't support switchover yet
        if r.status == 501 and action == 'switchover' and b'Server does not support this operation' in r.data:
            r = request_patroni(member, 'post', 'failover', failover_value)

        if r.status in (200, 202):
            logging.debug(r)
            cluster = dcs.get_cluster()
            logging.debug(cluster)
            click.echo('{0} {1}'.format(timestamp(), r.data.decode('utf-8')))
        else:
            click.echo('{0} failed, details: {1}, {2}'.format(action.title(), r.status, r.data.decode('utf-8')))
            return
    except Exception:
        logging.exception(r)
        logging.warning('Failing over to DCS')
        click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action))
        dcs.manual_failover(switchover_leader, candidate, scheduled_at=scheduled_at)

    output_members(cluster, cluster_name, group=group)
Perform a failover or a switchover operation in the cluster.

    Informational messages are printed in the console during the operation, as well as the list of members before and
    after the operation, so the user can follow the operation status.

    .. note::
        If not able to perform the operation through the REST API, write directly to the DCS as a fall back.

    :param action: action to be taken -- ``failover`` or ``switchover``.
    :param cluster_name: name of the Patroni cluster.
    :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be
        prompted for filling it -- unless *force* is ``True``, in which case an exception is raised.
    :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used.
    :param force: perform the failover or switchover without asking for confirmations.
    :param switchover_leader: name of the leader passed to the switchover command if any.
    :param switchover_scheduled: timestamp when the switchover should be scheduled to occur. If ``now``,
        perform immediately.

    :raises:
        :class:`PatroniCtlException`: if:
            * Patroni is running on a Citus cluster, but no *group* was specified; or
            * a switchover was requested by the cluster has no leader; or
            * *switchover_leader* does not match the current leader of the cluster; or
            * cluster has no candidates available for the operation; or
            * no *candidate* is given for a failover operation; or
            * current leader and *candidate* are the same; or
            * *candidate* is tagged as nofailover; or
            * *candidate* is not a member of the cluster; or
            * trying to schedule a switchover in a cluster that is in maintenance mode; or
            * user aborts the operation.
    
patroni/ctl.py