**Approach (brief)** Promote the read replica via RDS promote_read_replica API, poll until DB instance status is "available" and its read-replica status removed, then update Route53 record (CNAME or Alias) to point to the new instance endpoint. Make operations idempotent by checking current primary endpoint and DNS before making changes.python
#!/usr/bin/env python3
import boto3, time
from botocore.exceptions import ClientError
rds = boto3.client('rds')
r53 = boto3.client('route53')
def promote_and_point(read_replica_id, hosted_zone_id, record_name, use_alias=False):
# 1) Describe instance to get endpoint and role
inst = rds.describe_db_instances(DBInstanceIdentifier=read_replica_id)['DBInstances'][0]
endpoint = inst.get('Endpoint', {}).get('Address')
if not endpoint:
raise RuntimeError("Replica has no endpoint yet")
# 2) Idempotent: if replica already primary (not a ReadReplica), skip promote
if not inst.get('ReadReplicaSourceDBInstanceIdentifier'):
print("Instance already primary")
else:
try:
print("Promoting read replica...")
rds.promote_read_replica(DBInstanceIdentifier=read_replica_id)
except ClientError as e:
if e.response['Error']['Code'] == 'InvalidDBInstanceState':
# maybe already promoting or promoted; continue
print("Already promoting or invalid state, will poll")
else:
raise
# 3) Wait until available and no ReadReplicaSourceDBInstanceIdentifier
for _ in range(60):
inst = rds.describe_db_instances(DBInstanceIdentifier=read_replica_id)['DBInstances'][0]
status = inst['DBInstanceStatus']
if status == 'available' and not inst.get('ReadReplicaSourceDBInstanceIdentifier'):
endpoint = inst['Endpoint']['Address']
break
time.sleep(10)
else:
raise TimeoutError("Timed out waiting for promotion")
# 4) Idempotent DNS update: check current record
rr = r53.list_resource_record_sets(HostedZoneId=hosted_zone_id, StartRecordName=record_name, MaxItems="1")['ResourceRecordSets'][0]
current = None
if rr['Name'].rstrip('.') == record_name.rstrip('.'):
if use_alias and 'AliasTarget' in rr:
current = rr['AliasTarget']['DNSName'].rstrip('.')
elif not use_alias and 'ResourceRecords' in rr:
current = rr['ResourceRecords'][0]['Value'].rstrip('.')
if current == endpoint:
print("DNS already points to desired endpoint")
return
change = {
'Action': 'UPSERT',
'ResourceRecordSet': {
'Name': record_name,
'Type': 'A' if use_alias else 'CNAME',
'TTL': 60
}
}
if use_alias:
change['ResourceRecordSet'].pop('TTL', None)
change['ResourceRecordSet']['AliasTarget'] = {
'HostedZoneId': 'Z35SXDOTRQ7X7K', # RDS endpoint hosted zone id varies by region; replace as needed
'DNSName': endpoint + '.',
'EvaluateTargetHealth': False
}
else:
change['ResourceRecordSet']['ResourceRecords'] = [{'Value': endpoint}]
try:
r53.change_resource_record_sets(HostedZoneId=hosted_zone_id, ChangeBatch={'Changes':[change]})
print("DNS updated to", endpoint)
except ClientError as e:
raise
if __name__ == '__main__':
# Example usage
promote_and_point('my-read-replica', 'Z123HOSTEDZONE', 'db.example.com', use_alias=False)
