Create Read Replica for Aurora Database with custom endpoint

create-read-replica.py

import boto3

"""
Quick and dirty script - written as part of the AWS Data Lab Workshop
- More work required to formalize the below - untested code.

Move to CDK - when given a free moment
"""

class OnDemandReadReplica():
  
  def __init__(self, db_cluster_id,  # cluster id
               db_cluster_endpoint_id,  # cluster endpoint name you want to give to endpoint
               db_read_replica_instance_id_prefix_name, # prefix you want to give to read replicas will have number appended
               db_param_group_name, # param group name ie: default.aurora-mysql5.7
               engine, # db engine ie. 'aurora-mysql'
               engine_version, # engine version i.e '5.7.12'
               az, # availability zone the cluster is in
               db_instance_class, # instance class you want read replicas to be i.e 'db.t2.small'
               number_of_read_replicas, # number of read replicas you want to create
               read_replica_public_accessible=True,  # whether you want read replica to be publicly accessible
               storage_encrypted=False, # whether you want to encrypt storage
               promotion_tier=15, # specifies order aurora replica is promoted to the primary instance after failure
               excluded_members_from_endpoint=[], # names of any read replicas you want to explicitly exclude on the endpoint
               tags_for_endpoint=[{}]):  # tags you want to give the endpoint on creation
    
    self.rds = boto3.client('rds')
    self.db_cluster_id = db_cluster_id
    self.db_cluster_endpoint_id = db_cluster_endpoint_id
    self.db_param_group_name = db_param_group_name
    self.engine = engine
    self.engine_version
    self.az = az
    self.db_instance_class = db_instance_class
    self.number_of_read_replicas = number_of_read_replicas
    
  def create(self):
    list_of_read_repliacas = self.__create_n_read_replicas()
    endpoint_name = self.__create_reader_endpoint(list_of_read_repliacas)
    print("COMPLETE")
    
    
  def __create_reader_endpoint(static_members):
    response = self.rds.create_db_cluster_endpoint(
      DBClusterIdentifier=self.db_cluster_id,
      DBClusterEndpointIdentifier=self.db_cluster_endpoint_id
      EndpointType='READER',
      StaticMembers= static_members # list of db instance identifiers to include
      ExcludedMembers= self.excluded_members # list 
      Tags= self.tags # list of dictionaries - dict is a key value pair for a tag
    )
    
    
  def __create_n_read_replicas(self):
    if self.number_of_read_replicas < 1:
      raise ValueError("1 or more read replicase must be specified in number_of_replicas param")

    list_of_read_replicas = []
    for x in range(0,self.number_of_read_replicas):
      db_isntance_id_name = self.db_read_replica_instance_id_prefix_name+'-'+x
      list_of_read_replicas.append(db_isntance_id_name)
      self.rds.create_db_instance(
        DBParameterGroupName=self.db_param_group_name,
        Engine=self.engine,
        EngineVersion=self.engine_version,
        DBClusterIdentifier=self.db_cluster_id,
        AvailabilityZone=self.az,
        DBInstanceClass=self.db_instance_class,
        DBInstanceIdentifier=db_isntance_id_name,
        PubliclyAccessible=self.publically_accessible,
        StorageEncrypted=self.storage_encrypted,
        PromotionTier=self.promotion_tier)
      return list_of_read_replicas

required-permissions.json

    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "VisualEditor0",
                "Effect": "Allow",
                "Action": [
                    "rds:CreateDBClusterEndpoint",
                    "rds:CreateDBCluster",
                    "rds:CreateDBInstance"
                ],
                "Resource": "*"
            }
        ]
    }