import argparse
import etcd
from wmflib.interactive import ask_confirmation
def set_mirror_index(
old_replica: tuple[str, int],
new_replica: tuple[str, int],
prefix: str,
protocol: str,
):
old_client = etcd.Client(
host=old_replica[0], port=old_replica[1], protocol=protocol
)
replica_index = old_client.read("/").etcd_index
replica_key = f"/__replication/{prefix}"
ask_confirmation(f"Set {replica_key} to {replica_index} on {new_replica[0]}?")
new_client = etcd.Client(
host=new_replica[0], port=new_replica[1], protocol=protocol
)
new_client.write(replica_key, replica_index)
def hostport(hp: str) -> tuple[str, int]:
h, p = hp.split(":")
return h, int(p)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--protocol", default="https")
parser.add_argument("--old-replica", required=True, type=hostport)
parser.add_argument("--new-replica", required=True, type=hostport)
parser.add_argument("--prefix", required=True)
args = parser.parse_args()
set_mirror_index(args.old_replica, args.new_replica, args.prefix, args.protocol)
if __name__ == "__main__":
main()