Hay más ejemplos de AWS SDK disponibles en el GitHub repositorio de ejemplos de AWS Doc SDK
Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
Ejemplos de Aurora usando SDK para Python (Boto3)
Los siguientes ejemplos de código muestran cómo realizar acciones e implementar escenarios comunes mediante el uso AWS SDK for Python (Boto3) de Aurora.
Los conceptos básicos son ejemplos de código que muestran cómo realizar las operaciones esenciales dentro de un servicio.
Las acciones son extractos de código de programas más grandes y deben ejecutarse en contexto. Mientras las acciones muestran cómo llamar a las distintas funciones de servicio, es posible ver las acciones en contexto en los escenarios relacionados.
Los escenarios son ejemplos de código que muestran cómo llevar a cabo una tarea específica a través de llamadas a varias funciones dentro del servicio o combinado con otros Servicios de AWS.
En cada ejemplo se incluye un enlace al código de origen completo, con instrucciones de configuración y ejecución del código en el contexto.
Introducción
En el siguiente ejemplo de código se muestra cómo empezar a utilizar Aurora.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. import boto3 # Create an RDS client rds = boto3.client("rds") # Create a paginator for the describe_db_clusters operation paginator = rds.get_paginator("describe_db_clusters") # Use the paginator to get a list of DB clusters response_iterator = paginator.paginate( PaginationConfig={ "PageSize": 50, # Adjust PageSize as needed "StartingToken": None, } ) # Iterate through the pages of the response clusters_found = False for page in response_iterator: if "DBClusters" in page and page["DBClusters"]: clusters_found = True print("Here are your RDS Aurora clusters:") for cluster in page["DBClusters"]: print( f"Cluster ID: {cluster['DBClusterIdentifier']}, Engine: {cluster['Engine']}" ) if not clusters_found: print("No clusters found!")-
Para obtener más información sobre la API, consulta la sección Describe DBClusters en la referencia de la API de AWS SDK for Python (Boto3).
-
Conceptos básicos
En el siguiente ejemplo de código, se muestra cómo:
Cree un grupo de parámetros de clúster de base de datos de Aurora y defina los valores de los parámetros.
Cree un clúster de base de datos que utilice el grupo de parámetros.
Cree una instancia de base de datos que contenga una base de datos.
Realice una instantánea del clúster de base de datos y luego limpie los recursos.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. Ejecutar un escenario interactivo en un símbolo del sistema
class AuroraClusterScenario: """Runs a scenario that shows how to get started using Aurora DB clusters.""" def __init__(self, aurora_wrapper): """ :param aurora_wrapper: An object that wraps Aurora DB cluster actions. """ self.aurora_wrapper = aurora_wrapper def create_parameter_group(self, db_engine, parameter_group_name): """ Shows how to get available engine versions for a specified database engine and create a DB cluster parameter group that is compatible with a selected engine family. :param db_engine: The database engine to use as a basis. :param parameter_group_name: The name given to the newly created parameter group. :return: The newly created parameter group. """ print( f"Checking for an existing DB cluster parameter group named {parameter_group_name}." ) parameter_group = self.aurora_wrapper.get_parameter_group(parameter_group_name) if parameter_group is None: print(f"Getting available database engine versions for {db_engine}.") engine_versions = self.aurora_wrapper.get_engine_versions(db_engine) families = list({ver["DBParameterGroupFamily"] for ver in engine_versions}) family_index = q.choose("Which family do you want to use? ", families) print(f"Creating a DB cluster parameter group.") self.aurora_wrapper.create_parameter_group( parameter_group_name, families[family_index], "Example parameter group." ) parameter_group = self.aurora_wrapper.get_parameter_group( parameter_group_name ) print(f"Parameter group {parameter_group['DBClusterParameterGroupName']}:") pp(parameter_group) print("-" * 88) return parameter_group def set_user_parameters(self, parameter_group_name): """ Shows how to get the parameters contained in a custom parameter group and update some of the parameter values in the group. :param parameter_group_name: The name of the parameter group to query and modify. """ print("Let's set some parameter values in your parameter group.") auto_inc_parameters = self.aurora_wrapper.get_parameters( parameter_group_name, name_prefix="auto_increment" ) update_params = [] for auto_inc in auto_inc_parameters: if auto_inc["IsModifiable"] and auto_inc["DataType"] == "integer": print(f"The {auto_inc['ParameterName']} parameter is described as:") print(f"\t{auto_inc['Description']}") param_range = auto_inc["AllowedValues"].split("-") auto_inc["ParameterValue"] = str( q.ask( f"Enter a value between {param_range[0]} and {param_range[1]}: ", q.is_int, q.in_range(int(param_range[0]), int(param_range[1])), ) ) update_params.append(auto_inc) self.aurora_wrapper.update_parameters(parameter_group_name, update_params) print( "You can get a list of parameters you've set by specifying a source of 'user'." ) user_parameters = self.aurora_wrapper.get_parameters( parameter_group_name, source="user" ) pp(user_parameters) print("-" * 88) def create_cluster(self, cluster_name, db_engine, db_name, parameter_group): """ Shows how to create an Aurora DB cluster that contains a database of a specified type. The database is also configured to use a custom DB cluster parameter group. :param cluster_name: The name given to the newly created DB cluster. :param db_engine: The engine of the created database. :param db_name: The name given to the created database. :param parameter_group: The parameter group that is associated with the DB cluster. :return: The newly created DB cluster. """ print("Checking for an existing DB cluster.") cluster = self.aurora_wrapper.get_db_cluster(cluster_name) if cluster is None: admin_username = q.ask( "Enter an administrator user name for the database: ", q.non_empty ) admin_password = q.ask( "Enter a password for the administrator (at least 8 characters): ", q.non_empty, ) engine_versions = self.aurora_wrapper.get_engine_versions( db_engine, parameter_group["DBParameterGroupFamily"] ) engine_choices = [ ver["EngineVersionDescription"] for ver in engine_versions ] print("The available engines for your parameter group are:") engine_index = q.choose("Which engine do you want to use? ", engine_choices) print( f"Creating DB cluster {cluster_name} and database {db_name}.\n" f"The DB cluster is configured to use\n" f"your custom parameter group {parameter_group['DBClusterParameterGroupName']}\n" f"and selected engine {engine_choices[engine_index]}.\n" f"This typically takes several minutes." ) cluster = self.aurora_wrapper.create_db_cluster( cluster_name, parameter_group["DBClusterParameterGroupName"], db_name, db_engine, engine_versions[engine_index]["EngineVersion"], admin_username, admin_password, ) while cluster.get("Status") != "available": wait(30) cluster = self.aurora_wrapper.get_db_cluster(cluster_name) print("Cluster created and available.\n") print("Cluster data:") pp(cluster) print("-" * 88) return cluster def create_instance(self, cluster): """ Shows how to create a DB instance in an existing Aurora DB cluster. A new DB cluster contains no DB instances, so you must add one. The first DB instance that is added to a DB cluster defaults to a read-write DB instance. :param cluster: The DB cluster where the DB instance is added. :return: The newly created DB instance. """ print("Checking for an existing database instance.") cluster_name = cluster["DBClusterIdentifier"] db_inst = self.aurora_wrapper.get_db_instance(cluster_name) if db_inst is None: print("Let's create a database instance in your DB cluster.") print("First, choose a DB instance type:") inst_opts = self.aurora_wrapper.get_orderable_instances( cluster["Engine"], cluster["EngineVersion"] ) inst_choices = list( { opt["DBInstanceClass"] + ", storage type: " + opt["StorageType"] for opt in inst_opts } ) inst_index = q.choose( "Which DB instance class do you want to use? ", inst_choices ) print( f"Creating a database instance. This typically takes several minutes." ) db_inst = self.aurora_wrapper.create_instance_in_cluster( cluster_name, cluster_name, cluster["Engine"], inst_opts[inst_index]["DBInstanceClass"], ) while db_inst.get("DBInstanceStatus") != "available": wait(30) db_inst = self.aurora_wrapper.get_db_instance(cluster_name) print("Instance data:") pp(db_inst) print("-" * 88) return db_inst @staticmethod def display_connection(cluster): """ Displays connection information about an Aurora DB cluster and tips on how to connect to it. :param cluster: The DB cluster to display. """ print( "You can now connect to your database using your favorite MySql client.\n" "One way to connect is by using the 'mysql' shell on an Amazon EC2 instance\n" "that is running in the same VPC as your database cluster. Pass the endpoint,\n" "port, and administrator user name to 'mysql' and enter your password\n" "when prompted:\n" ) print( f"\n\tmysql -h {cluster['Endpoint']} -P {cluster['Port']} -u {cluster['MasterUsername']} -p\n" ) print( "For more information, see the User Guide for Aurora:\n" "\thttps://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.Aurora.html#CHAP_GettingStartedAurora.Aurora.Connect" ) print("-" * 88) def create_snapshot(self, cluster_name): """ Shows how to create a DB cluster snapshot and wait until it's available. :param cluster_name: The name of a DB cluster to snapshot. """ if q.ask( "Do you want to create a snapshot of your DB cluster (y/n)? ", q.is_yesno ): snapshot_id = f"{cluster_name}-{uuid.uuid4()}" print( f"Creating a snapshot named {snapshot_id}. This typically takes a few minutes." ) snapshot = self.aurora_wrapper.create_cluster_snapshot( snapshot_id, cluster_name ) while snapshot.get("Status") != "available": wait(30) snapshot = self.aurora_wrapper.get_cluster_snapshot(snapshot_id) pp(snapshot) print("-" * 88) def cleanup(self, db_inst, cluster, parameter_group): """ Shows how to clean up a DB instance, DB cluster, and DB cluster parameter group. Before the DB cluster parameter group can be deleted, all associated DB instances and DB clusters must first be deleted. :param db_inst: The DB instance to delete. :param cluster: The DB cluster to delete. :param parameter_group: The DB cluster parameter group to delete. """ cluster_name = cluster["DBClusterIdentifier"] parameter_group_name = parameter_group["DBClusterParameterGroupName"] if q.ask( "\nDo you want to delete the database instance, DB cluster, and parameter " "group (y/n)? ", q.is_yesno, ): print(f"Deleting database instance {db_inst['DBInstanceIdentifier']}.") self.aurora_wrapper.delete_db_instance(db_inst["DBInstanceIdentifier"]) print(f"Deleting database cluster {cluster_name}.") self.aurora_wrapper.delete_db_cluster(cluster_name) print( "Waiting for the DB instance and DB cluster to delete.\n" "This typically takes several minutes." ) while db_inst is not None or cluster is not None: wait(30) if db_inst is not None: db_inst = self.aurora_wrapper.get_db_instance( db_inst["DBInstanceIdentifier"] ) if cluster is not None: cluster = self.aurora_wrapper.get_db_cluster( cluster["DBClusterIdentifier"] ) print(f"Deleting parameter group {parameter_group_name}.") self.aurora_wrapper.delete_parameter_group(parameter_group_name) def run_scenario(self, db_engine, parameter_group_name, cluster_name, db_name): print("-" * 88) print( "Welcome to the Amazon Relational Database Service (Amazon RDS) get started\n" "with Aurora DB clusters demo." ) print("-" * 88) parameter_group = self.create_parameter_group(db_engine, parameter_group_name) self.set_user_parameters(parameter_group_name) cluster = self.create_cluster(cluster_name, db_engine, db_name, parameter_group) wait(5) db_inst = self.create_instance(cluster) self.display_connection(cluster) self.create_snapshot(cluster_name) self.cleanup(db_inst, cluster, parameter_group) print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") try: scenario = AuroraClusterScenario(AuroraWrapper.from_client()) scenario.run_scenario( "aurora-mysql", "doc-example-cluster-parameter-group", "doc-example-aurora", "docexampledb", ) except Exception: logging.exception("Something went wrong with the demo.")Defina las funciones a las que llama el escenario para administrar las acciones de Aurora.
class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_parameter_group(self, parameter_group_name): """ Gets a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to retrieve. :return: The requested parameter group. """ try: response = self.rds_client.describe_db_cluster_parameter_groups( DBClusterParameterGroupName=parameter_group_name ) parameter_group = response["DBClusterParameterGroups"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBParameterGroupNotFound": logger.info("Parameter group %s does not exist.", parameter_group_name) else: logger.error( "Couldn't get parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return parameter_group def create_parameter_group( self, parameter_group_name, parameter_group_family, description ): """ Creates a DB cluster parameter group that is based on the specified parameter group family. :param parameter_group_name: The name of the newly created parameter group. :param parameter_group_family: The family that is used as the basis of the new parameter group. :param description: A description given to the parameter group. :return: Data about the newly created parameter group. """ try: response = self.rds_client.create_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name, DBParameterGroupFamily=parameter_group_family, Description=description, ) except ClientError as err: logger.error( "Couldn't create parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def delete_parameter_group(self, parameter_group_name): """ Deletes a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to delete. :return: Data about the parameter group. """ try: response = self.rds_client.delete_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name ) except ClientError as err: logger.error( "Couldn't delete parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def get_parameters(self, parameter_group_name, name_prefix="", source=None): """ Gets the parameters that are contained in a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to query. :param name_prefix: When specified, the retrieved list of parameters is filtered to contain only parameters that start with this prefix. :param source: When specified, only parameters from this source are retrieved. For example, a source of 'user' retrieves only parameters that were set by a user. :return: The list of requested parameters. """ try: kwargs = {"DBClusterParameterGroupName": parameter_group_name} if source is not None: kwargs["Source"] = source parameters = [] paginator = self.rds_client.get_paginator("describe_db_cluster_parameters") for page in paginator.paginate(**kwargs): parameters += [ p for p in page["Parameters"] if p["ParameterName"].startswith(name_prefix) ] except ClientError as err: logger.error( "Couldn't get parameters for %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return parameters def update_parameters(self, parameter_group_name, update_parameters): """ Updates parameters in a custom DB cluster parameter group. :param parameter_group_name: The name of the parameter group to update. :param update_parameters: The parameters to update in the group. :return: Data about the modified parameter group. """ try: response = self.rds_client.modify_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name, Parameters=update_parameters, ) except ClientError as err: logger.error( "Couldn't update parameters in %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def get_db_cluster(self, cluster_name): """ Gets data about an Aurora DB cluster. :param cluster_name: The name of the DB cluster to retrieve. :return: The retrieved DB cluster. """ try: response = self.rds_client.describe_db_clusters( DBClusterIdentifier=cluster_name ) cluster = response["DBClusters"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBClusterNotFoundFault": logger.info("Cluster %s does not exist.", cluster_name) else: logger.error( "Couldn't verify the existence of DB cluster %s. Here's why: %s: %s", cluster_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return cluster def create_db_cluster( self, cluster_name, parameter_group_name, db_name, db_engine, db_engine_version, admin_name, admin_password, ): """ Creates a DB cluster that is configured to use the specified parameter group. The newly created DB cluster contains a database that uses the specified engine and engine version. :param cluster_name: The name of the DB cluster to create. :param parameter_group_name: The name of the parameter group to associate with the DB cluster. :param db_name: The name of the database to create. :param db_engine: The database engine of the database that is created, such as MySql. :param db_engine_version: The version of the database engine. :param admin_name: The user name of the database administrator. :param admin_password: The password of the database administrator. :return: The newly created DB cluster. """ try: response = self.rds_client.create_db_cluster( DatabaseName=db_name, DBClusterIdentifier=cluster_name, DBClusterParameterGroupName=parameter_group_name, Engine=db_engine, EngineVersion=db_engine_version, MasterUsername=admin_name, MasterUserPassword=admin_password, ) cluster = response["DBCluster"] except ClientError as err: logger.error( "Couldn't create database %s. Here's why: %s: %s", db_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return cluster def delete_db_cluster(self, cluster_name): """ Deletes a DB cluster. :param cluster_name: The name of the DB cluster to delete. """ try: self.rds_client.delete_db_cluster( DBClusterIdentifier=cluster_name, SkipFinalSnapshot=True ) logger.info("Deleted DB cluster %s.", cluster_name) except ClientError: logger.exception("Couldn't delete DB cluster %s.", cluster_name) raise def create_cluster_snapshot(self, snapshot_id, cluster_id): """ Creates a snapshot of a DB cluster. :param snapshot_id: The ID to give the created snapshot. :param cluster_id: The DB cluster to snapshot. :return: Data about the newly created snapshot. """ try: response = self.rds_client.create_db_cluster_snapshot( DBClusterSnapshotIdentifier=snapshot_id, DBClusterIdentifier=cluster_id ) snapshot = response["DBClusterSnapshot"] except ClientError as err: logger.error( "Couldn't create snapshot of %s. Here's why: %s: %s", cluster_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return snapshot def get_cluster_snapshot(self, snapshot_id): """ Gets a DB cluster snapshot. :param snapshot_id: The ID of the snapshot to retrieve. :return: The retrieved snapshot. """ try: response = self.rds_client.describe_db_cluster_snapshots( DBClusterSnapshotIdentifier=snapshot_id ) snapshot = response["DBClusterSnapshots"][0] except ClientError as err: logger.error( "Couldn't get DB cluster snapshot %s. Here's why: %s: %s", snapshot_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return snapshot def create_instance_in_cluster( self, instance_id, cluster_id, db_engine, instance_class ): """ Creates a database instance in an existing DB cluster. The first database that is created defaults to a read-write DB instance. :param instance_id: The ID to give the newly created DB instance. :param cluster_id: The ID of the DB cluster where the DB instance is created. :param db_engine: The database engine of a database to create in the DB instance. This must be compatible with the configured parameter group of the DB cluster. :param instance_class: The DB instance class for the newly created DB instance. :return: Data about the newly created DB instance. """ try: response = self.rds_client.create_db_instance( DBInstanceIdentifier=instance_id, DBClusterIdentifier=cluster_id, Engine=db_engine, DBInstanceClass=instance_class, ) db_inst = response["DBInstance"] except ClientError as err: logger.error( "Couldn't create DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst def get_engine_versions(self, engine, parameter_group_family=None): """ Gets database engine versions that are available for the specified engine and parameter group family. :param engine: The database engine to look up. :param parameter_group_family: When specified, restricts the returned list of engine versions to those that are compatible with this parameter group family. :return: The list of database engine versions. """ try: kwargs = {"Engine": engine} if parameter_group_family is not None: kwargs["DBParameterGroupFamily"] = parameter_group_family response = self.rds_client.describe_db_engine_versions(**kwargs) versions = response["DBEngineVersions"] except ClientError as err: logger.error( "Couldn't get engine versions for %s. Here's why: %s: %s", engine, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return versions def get_orderable_instances(self, db_engine, db_engine_version): """ Gets DB instance options that can be used to create DB instances that are compatible with a set of specifications. :param db_engine: The database engine that must be supported by the DB instance. :param db_engine_version: The engine version that must be supported by the DB instance. :return: The list of DB instance options that can be used to create a compatible DB instance. """ try: inst_opts = [] paginator = self.rds_client.get_paginator( "describe_orderable_db_instance_options" ) for page in paginator.paginate( Engine=db_engine, EngineVersion=db_engine_version ): inst_opts += page["OrderableDBInstanceOptions"] except ClientError as err: logger.error( "Couldn't get orderable DB instances. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return inst_opts def get_db_instance(self, instance_id): """ Gets data about a DB instance. :param instance_id: The ID of the DB instance to retrieve. :return: The retrieved DB instance. """ try: response = self.rds_client.describe_db_instances( DBInstanceIdentifier=instance_id ) db_inst = response["DBInstances"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBInstanceNotFound": logger.info("Instance %s does not exist.", instance_id) else: logger.error( "Couldn't get DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst def delete_db_instance(self, instance_id): """ Deletes a DB instance. :param instance_id: The ID of the DB instance to delete. :return: Data about the deleted DB instance. """ try: response = self.rds_client.delete_db_instance( DBInstanceIdentifier=instance_id, SkipFinalSnapshot=True, DeleteAutomatedBackups=True, ) db_inst = response["DBInstance"] except ClientError as err: logger.error( "Couldn't delete DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst-
Para obtener información sobre la API, consulte los siguientes temas en la Referencia de la API de AWS SDK para Python (Boto3).
-
Acciones
En el siguiente ejemplo de código, se muestra cómo utilizar CreateDBCluster.
- SDK para Python (Boto3)
-
nota
Hay más información GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def create_db_cluster( self, cluster_name, parameter_group_name, db_name, db_engine, db_engine_version, admin_name, admin_password, ): """ Creates a DB cluster that is configured to use the specified parameter group. The newly created DB cluster contains a database that uses the specified engine and engine version. :param cluster_name: The name of the DB cluster to create. :param parameter_group_name: The name of the parameter group to associate with the DB cluster. :param db_name: The name of the database to create. :param db_engine: The database engine of the database that is created, such as MySql. :param db_engine_version: The version of the database engine. :param admin_name: The user name of the database administrator. :param admin_password: The password of the database administrator. :return: The newly created DB cluster. """ try: response = self.rds_client.create_db_cluster( DatabaseName=db_name, DBClusterIdentifier=cluster_name, DBClusterParameterGroupName=parameter_group_name, Engine=db_engine, EngineVersion=db_engine_version, MasterUsername=admin_name, MasterUserPassword=admin_password, ) cluster = response["DBCluster"] except ClientError as err: logger.error( "Couldn't create database %s. Here's why: %s: %s", db_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return cluster-
Para obtener más información sobre la API, consulta la referencia de la API Create DBCluster in AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar CreateDBClusterParameterGroup.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def create_parameter_group( self, parameter_group_name, parameter_group_family, description ): """ Creates a DB cluster parameter group that is based on the specified parameter group family. :param parameter_group_name: The name of the newly created parameter group. :param parameter_group_family: The family that is used as the basis of the new parameter group. :param description: A description given to the parameter group. :return: Data about the newly created parameter group. """ try: response = self.rds_client.create_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name, DBParameterGroupFamily=parameter_group_family, Description=description, ) except ClientError as err: logger.error( "Couldn't create parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response-
Para obtener más información sobre la API, consulta la referencia de la API Create DBCluster ParameterGroup in AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar CreateDBClusterSnapshot.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def create_cluster_snapshot(self, snapshot_id, cluster_id): """ Creates a snapshot of a DB cluster. :param snapshot_id: The ID to give the created snapshot. :param cluster_id: The DB cluster to snapshot. :return: Data about the newly created snapshot. """ try: response = self.rds_client.create_db_cluster_snapshot( DBClusterSnapshotIdentifier=snapshot_id, DBClusterIdentifier=cluster_id ) snapshot = response["DBClusterSnapshot"] except ClientError as err: logger.error( "Couldn't create snapshot of %s. Here's why: %s: %s", cluster_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return snapshot-
Para obtener más información sobre la API, consulta Crear DBCluster una instantánea en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar CreateDBInstance.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def create_instance_in_cluster( self, instance_id, cluster_id, db_engine, instance_class ): """ Creates a database instance in an existing DB cluster. The first database that is created defaults to a read-write DB instance. :param instance_id: The ID to give the newly created DB instance. :param cluster_id: The ID of the DB cluster where the DB instance is created. :param db_engine: The database engine of a database to create in the DB instance. This must be compatible with the configured parameter group of the DB cluster. :param instance_class: The DB instance class for the newly created DB instance. :return: Data about the newly created DB instance. """ try: response = self.rds_client.create_db_instance( DBInstanceIdentifier=instance_id, DBClusterIdentifier=cluster_id, Engine=db_engine, DBInstanceClass=instance_class, ) db_inst = response["DBInstance"] except ClientError as err: logger.error( "Couldn't create DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst-
Para obtener más información sobre la API, consulta la referencia de la API Create DBInstance in AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DeleteDBCluster.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def delete_db_cluster(self, cluster_name): """ Deletes a DB cluster. :param cluster_name: The name of the DB cluster to delete. """ try: self.rds_client.delete_db_cluster( DBClusterIdentifier=cluster_name, SkipFinalSnapshot=True ) logger.info("Deleted DB cluster %s.", cluster_name) except ClientError: logger.exception("Couldn't delete DB cluster %s.", cluster_name) raise-
Para obtener más información sobre la API, consulta Eliminar DBCluster en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DeleteDBClusterParameterGroup.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def delete_parameter_group(self, parameter_group_name): """ Deletes a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to delete. :return: Data about the parameter group. """ try: response = self.rds_client.delete_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name ) except ClientError as err: logger.error( "Couldn't delete parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response-
Para obtener más información sobre la API, consulta Eliminar DBCluster ParameterGroup en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DeleteDBInstance.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def delete_db_instance(self, instance_id): """ Deletes a DB instance. :param instance_id: The ID of the DB instance to delete. :return: Data about the deleted DB instance. """ try: response = self.rds_client.delete_db_instance( DBInstanceIdentifier=instance_id, SkipFinalSnapshot=True, DeleteAutomatedBackups=True, ) db_inst = response["DBInstance"] except ClientError as err: logger.error( "Couldn't delete DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst-
Para obtener más información sobre la API, consulta Eliminar DBInstance en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBClusterParameterGroups.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_parameter_group(self, parameter_group_name): """ Gets a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to retrieve. :return: The requested parameter group. """ try: response = self.rds_client.describe_db_cluster_parameter_groups( DBClusterParameterGroupName=parameter_group_name ) parameter_group = response["DBClusterParameterGroups"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBParameterGroupNotFound": logger.info("Parameter group %s does not exist.", parameter_group_name) else: logger.error( "Couldn't get parameter group %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return parameter_group-
Para obtener más información sobre la API, consulta la sección Describe DBCluster ParameterGroups en la referencia de la API de AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBClusterParameters.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_parameters(self, parameter_group_name, name_prefix="", source=None): """ Gets the parameters that are contained in a DB cluster parameter group. :param parameter_group_name: The name of the parameter group to query. :param name_prefix: When specified, the retrieved list of parameters is filtered to contain only parameters that start with this prefix. :param source: When specified, only parameters from this source are retrieved. For example, a source of 'user' retrieves only parameters that were set by a user. :return: The list of requested parameters. """ try: kwargs = {"DBClusterParameterGroupName": parameter_group_name} if source is not None: kwargs["Source"] = source parameters = [] paginator = self.rds_client.get_paginator("describe_db_cluster_parameters") for page in paginator.paginate(**kwargs): parameters += [ p for p in page["Parameters"] if p["ParameterName"].startswith(name_prefix) ] except ClientError as err: logger.error( "Couldn't get parameters for %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return parameters-
Para obtener más información sobre la API, consulta Describe DBCluster los parámetros en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBClusterSnapshots.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_cluster_snapshot(self, snapshot_id): """ Gets a DB cluster snapshot. :param snapshot_id: The ID of the snapshot to retrieve. :return: The retrieved snapshot. """ try: response = self.rds_client.describe_db_cluster_snapshots( DBClusterSnapshotIdentifier=snapshot_id ) snapshot = response["DBClusterSnapshots"][0] except ClientError as err: logger.error( "Couldn't get DB cluster snapshot %s. Here's why: %s: %s", snapshot_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return snapshot-
Para obtener más información sobre la API, consulta Describe DBCluster las instantáneas en la referencia de la API AWS del SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBClusters.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_db_cluster(self, cluster_name): """ Gets data about an Aurora DB cluster. :param cluster_name: The name of the DB cluster to retrieve. :return: The retrieved DB cluster. """ try: response = self.rds_client.describe_db_clusters( DBClusterIdentifier=cluster_name ) cluster = response["DBClusters"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBClusterNotFoundFault": logger.info("Cluster %s does not exist.", cluster_name) else: logger.error( "Couldn't verify the existence of DB cluster %s. Here's why: %s: %s", cluster_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return cluster-
Para obtener más información sobre la API, consulta la sección Describe DBClusters en la referencia de la API de AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBEngineVersions.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_engine_versions(self, engine, parameter_group_family=None): """ Gets database engine versions that are available for the specified engine and parameter group family. :param engine: The database engine to look up. :param parameter_group_family: When specified, restricts the returned list of engine versions to those that are compatible with this parameter group family. :return: The list of database engine versions. """ try: kwargs = {"Engine": engine} if parameter_group_family is not None: kwargs["DBParameterGroupFamily"] = parameter_group_family response = self.rds_client.describe_db_engine_versions(**kwargs) versions = response["DBEngineVersions"] except ClientError as err: logger.error( "Couldn't get engine versions for %s. Here's why: %s: %s", engine, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return versions-
Para obtener más información sobre la API, consulta Describir DBEngine las versiones en la referencia de la API de AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeDBInstances.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_db_instance(self, instance_id): """ Gets data about a DB instance. :param instance_id: The ID of the DB instance to retrieve. :return: The retrieved DB instance. """ try: response = self.rds_client.describe_db_instances( DBInstanceIdentifier=instance_id ) db_inst = response["DBInstances"][0] except ClientError as err: if err.response["Error"]["Code"] == "DBInstanceNotFound": logger.info("Instance %s does not exist.", instance_id) else: logger.error( "Couldn't get DB instance %s. Here's why: %s: %s", instance_id, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return db_inst-
Para obtener más información sobre la API, consulta la sección Describe DBInstances en la referencia de la API de AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar DescribeOrderableDBInstanceOptions.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def get_orderable_instances(self, db_engine, db_engine_version): """ Gets DB instance options that can be used to create DB instances that are compatible with a set of specifications. :param db_engine: The database engine that must be supported by the DB instance. :param db_engine_version: The engine version that must be supported by the DB instance. :return: The list of DB instance options that can be used to create a compatible DB instance. """ try: inst_opts = [] paginator = self.rds_client.get_paginator( "describe_orderable_db_instance_options" ) for page in paginator.paginate( Engine=db_engine, EngineVersion=db_engine_version ): inst_opts += page["OrderableDBInstanceOptions"] except ClientError as err: logger.error( "Couldn't get orderable DB instances. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return inst_opts-
Para obtener más información sobre la API, consulta DescribeOrderableDBInstanceOpciones en la referencia de la API del AWS SDK for Python (Boto3).
-
En el siguiente ejemplo de código, se muestra cómo utilizar ModifyDBClusterParameterGroup.
- SDK para Python (Boto3)
-
nota
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS
. class AuroraWrapper: """Encapsulates Aurora DB cluster actions.""" def __init__(self, rds_client): """ :param rds_client: A Boto3 Amazon Relational Database Service (Amazon RDS) client. """ self.rds_client = rds_client @classmethod def from_client(cls): """ Instantiates this class from a Boto3 client. """ rds_client = boto3.client("rds") return cls(rds_client) def update_parameters(self, parameter_group_name, update_parameters): """ Updates parameters in a custom DB cluster parameter group. :param parameter_group_name: The name of the parameter group to update. :param update_parameters: The parameters to update in the group. :return: Data about the modified parameter group. """ try: response = self.rds_client.modify_db_cluster_parameter_group( DBClusterParameterGroupName=parameter_group_name, Parameters=update_parameters, ) except ClientError as err: logger.error( "Couldn't update parameters in %s. Here's why: %s: %s", parameter_group_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response-
Para obtener más información sobre la API, consulta Modificar DBCluster ParameterGroup en el AWS SDK para la referencia de la API de Python (Boto3).
-
Escenarios
En el siguiente ejemplo de código se muestra cómo crear una biblioteca de préstamos en la que los usuarios puedan pedir prestados y devolver libros mediante una API de REST respaldada por una base de datos de Amazon Aurora.
- SDK para Python (Boto3)
-
Muestra cómo utilizarla AWS SDK for Python (Boto3) con la API de Amazon Relational Database Service (Amazon RDS) y AWS Chalice para crear una API REST respaldada por una base de datos de Amazon Aurora. El servicio web es totalmente sin servidor y representa una biblioteca de préstamos sencilla en la que los usuarios pueden pedir prestados libros y devolverlos. Aprenda cómo:
Crear y administrar un clúster de base de datos Aurora sin servidor.
Se utiliza para administrar AWS Secrets Manager las credenciales de la base de datos.
Implementar una capa de almacenamiento de datos que utilice Amazon RDS para mover datos dentro y fuera de la base de datos.
Use AWS Chalice para implementar una API REST sin servidor en Amazon API Gateway y. AWS Lambda
Utilice el paquete Requests para enviar solicitudes al servicio web.
Para obtener el código fuente completo y las instrucciones sobre cómo configurarla y ejecutarla, consulte el ejemplo completo en. GitHub
Servicios utilizados en este ejemplo
API Gateway
Aurora
Lambda
Secrets Manager
El siguiente ejemplo de código muestra cómo crear una aplicación web que haga un seguimiento de los elementos de trabajo en una base de datos Amazon Aurora Serverless y utilice Amazon Simple Email Service (Amazon SES) para enviar informes.
- SDK para Python (Boto3)
-
Muestra cómo usarlo AWS SDK for Python (Boto3) para crear un servicio REST que rastrea los elementos de trabajo en una base de datos Amazon Aurora Serverless y envía informes por correo electrónico mediante Amazon Simple Email Service (Amazon SES). En este ejemplo se utiliza el marco web de Flask para gestionar el enrutamiento HTTP y se integra con una página web de React para presentar una aplicación web completamente funcional.
Cree un servicio REST de Flask que se integre con. Servicios de AWS
Lea, escriba y actualice los elementos de trabajo almacenados en una base de datos de Aurora Serverless.
Cree un AWS Secrets Manager secreto que contenga las credenciales de la base de datos y utilícelo para autenticar las llamadas a la base de datos.
Utilice Amazon SES para enviar informes de elementos de trabajo por correo electrónico.
Para obtener el código fuente completo y las instrucciones sobre cómo configurarlo y ejecutarlo, consulte el ejemplo completo en GitHub
. Servicios utilizados en este ejemplo
Aurora
Amazon RDS
Servicio de datos de Amazon RDS
Amazon SES
