Há mais exemplos de AWS SDK disponíveis no repositório [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub .
As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Exemplos do Amazon RDS usando o SDK para Python (Boto3)
Os exemplos de código a seguir mostram como realizar ações e implementar cenários comuns usando o AWS SDK para Python (Boto3) com o Amazon RDS.
As *noções básicas* são exemplos de código que mostram como realizar as operações essenciais em um serviço.
*Ações* são trechos de código de programas maiores e devem ser executadas em contexto. Embora as ações mostrem como chamar perfis de serviço individuais, você pode ver as ações no contexto em seus cenários relacionados.
*Cenários* são exemplos de código que mostram como realizar tarefas específicas chamando várias funções dentro de um serviço ou combinadas com outros Serviços da AWS.
Cada exemplo inclui um link para o código-fonte completo, em que você pode encontrar instruções sobre como configurar e executar o código.
**Topics**
+ [Conceitos básicos](#get_started)
+ [Conceitos básicos](#basics)
+ [Ações](#actions)
+ [Cenários](#scenarios)
+ [Exemplos sem servidor](#serverless_examples)
## Conceitos básicos
### Olá, Amazon RDS
O exemplo de código a seguir mostra como começar a usar o Amazon RDS.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
"""
Purpose
Shows how to use the AWS SDK for Python (Boto3) with the Amazon Relational Database Service
(Amazon RDS) to list the databases in your account.
"""
import boto3
from botocore.exceptions import ClientError
# Create an RDS client
rds_client = boto3.client("rds")
# Create a paginator for the describe_db_instances operation
paginator = rds_client.get_paginator("describe_db_instances")
try:
# Use the paginator to get a list of DB instances
response_iterator = paginator.paginate(
PaginationConfig={
"MaxItems": 123,
"PageSize": 50, # Adjust PageSize as needed
"StartingToken": None,
}
)
# Iterate through the pages of the response
instances_found = False
for page in response_iterator:
if "DBInstances" in page and page["DBInstances"]:
instances_found = True
print("Your RDS instances are:")
for db in page["DBInstances"]:
print(db["DBInstanceIdentifier"])
if not instances_found:
print("No RDS instances found!")
except ClientError as e:
print(f"Couldn't list RDS instances. Here's why: {e.response['Error']['Message']}")
```
+ Para obter detalhes da API, consulte a Referência da API [Descrever DBInstances](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBInstances) no *AWS SDK for Python (Boto3*).
## Conceitos básicos
### Conheça os conceitos básicos
O exemplo de código a seguir mostra como:
+ Criar um grupo de parâmetros de banco de dados e definir os valores dos parâmetros.
+ Criar uma instância de banco de dados configurada para usar o grupo de parâmetros. A instância de banco de dados também contém um banco de dados.
+ Criar um snapshot da instância.
+ Exclua a instância e o grupo de parâmetros.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
Execute um cenário interativo em um prompt de comando.
```
class RdsInstanceScenario:
"""Runs a scenario that shows how to get started using Amazon RDS DB instances."""
def __init__(self, instance_wrapper):
"""
:param instance_wrapper: An object that wraps Amazon RDS DB instance actions.
"""
self.instance_wrapper = instance_wrapper
def create_parameter_group(self, parameter_group_name, db_engine):
"""
Shows how to get available engine versions for a specified database engine and
create a DB parameter group that is compatible with a selected engine family.
:param parameter_group_name: The name given to the newly created parameter group.
:param db_engine: The database engine to use as a basis.
:return: The newly created parameter group.
"""
print(
f"Checking for an existing DB instance parameter group named {parameter_group_name}."
)
parameter_group = self.instance_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.instance_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 parameter group.")
self.instance_wrapper.create_parameter_group(
parameter_group_name, families[family_index], "Example parameter group."
)
parameter_group = self.instance_wrapper.get_parameter_group(
parameter_group_name
)
print(f"Parameter group {parameter_group['DBParameterGroupName']}:")
pp(parameter_group)
print("-" * 88)
return parameter_group
def update_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.instance_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.instance_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.instance_wrapper.get_parameters(
parameter_group_name, source="user"
)
pp(user_parameters)
print("-" * 88)
def create_instance(self, instance_name, db_name, db_engine, parameter_group):
"""
Shows how to create a DB instance that contains a database of a specified
type and is configured to use a custom DB parameter group.
:param instance_name: The name given to the newly created DB instance.
:param db_name: The name given to the created database.
:param db_engine: The engine of the created database.
:param parameter_group: The parameter group that is associated with the DB instance.
:return: The newly created DB instance.
"""
print("Checking for an existing DB instance.")
db_inst = self.instance_wrapper.get_db_instance(instance_name)
if db_inst is None:
print("Let's create a DB instance.")
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.instance_wrapper.get_engine_versions(
db_engine, parameter_group["DBParameterGroupFamily"]
)
engine_choices = [ver["EngineVersion"] 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)
engine_selection = engine_versions[engine_index]
print(
"The available micro DB instance classes for your database engine are:"
)
inst_opts = self.instance_wrapper.get_orderable_instances(
engine_selection["Engine"], engine_selection["EngineVersion"]
)
inst_choices = list(
{
opt["DBInstanceClass"]
for opt in inst_opts
if "micro" in opt["DBInstanceClass"]
}
)
inst_index = q.choose(
"Which micro DB instance class do you want to use? ", inst_choices
)
group_name = parameter_group["DBParameterGroupName"]
storage_type = "standard"
allocated_storage = 5
print(
f"Creating a DB instance named {instance_name} and database {db_name}.\n"
f"The DB instance is configured to use your custom parameter group {group_name},\n"
f"selected engine {engine_selection['EngineVersion']},\n"
f"selected DB instance class {inst_choices[inst_index]},"
f"and {allocated_storage} GiB of {storage_type} storage.\n"
f"This typically takes several minutes."
)
db_inst = self.instance_wrapper.create_db_instance(
db_name,
instance_name,
group_name,
engine_selection["Engine"],
engine_selection["EngineVersion"],
inst_choices[inst_index],
storage_type,
allocated_storage,
admin_username,
admin_password,
)
while db_inst.get("DBInstanceStatus") != "available":
wait(10)
db_inst = self.instance_wrapper.get_db_instance(instance_name)
print("Instance data:")
pp(db_inst)
print("-" * 88)
return db_inst
@staticmethod
def display_connection(db_inst):
"""
Displays connection information about a DB instance and tips on how to
connect to it.
:param db_inst: The DB instance 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 DB instance. Pass the endpoint,\n"
"port, and administrator user name to 'mysql' and enter your password\n"
"when prompted:\n"
)
print(
f"\n\tmysql -h {db_inst['Endpoint']['Address']} -P {db_inst['Endpoint']['Port']} "
f"-u {db_inst['MasterUsername']} -p\n"
)
print(
"For more information, see the User Guide for Amazon RDS:\n"
"\thttps://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_GettingStarted.CreatingConnecting.MySQL.html#CHAP_GettingStarted.Connecting.MySQL"
)
print("-" * 88)
def create_snapshot(self, instance_name):
"""
Shows how to create a DB instance snapshot and wait until it's available.
:param instance_name: The name of a DB instance to snapshot.
"""
if q.ask(
"Do you want to create a snapshot of your DB instance (y/n)? ", q.is_yesno
):
snapshot_id = f"{instance_name}-{uuid.uuid4()}"
print(
f"Creating a snapshot named {snapshot_id}. This typically takes a few minutes."
)
snapshot = self.instance_wrapper.create_snapshot(snapshot_id, instance_name)
while snapshot.get("Status") != "available":
wait(10)
snapshot = self.instance_wrapper.get_snapshot(snapshot_id)
pp(snapshot)
print("-" * 88)
def cleanup(self, db_inst, parameter_group_name):
"""
Shows how to clean up a DB instance and parameter group.
Before the parameter group can be deleted, all associated DB instances must first
be deleted.
:param db_inst: The DB instance to delete.
:param parameter_group_name: The DB parameter group to delete.
"""
if q.ask(
"\nDo you want to delete the DB instance and parameter group (y/n)? ",
q.is_yesno,
):
print(f"Deleting DB instance {db_inst['DBInstanceIdentifier']}.")
self.instance_wrapper.delete_db_instance(db_inst["DBInstanceIdentifier"])
print(
"Waiting for the DB instance to delete. This typically takes several minutes."
)
while db_inst is not None:
wait(10)
db_inst = self.instance_wrapper.get_db_instance(
db_inst["DBInstanceIdentifier"]
)
print(f"Deleting parameter group {parameter_group_name}.")
self.instance_wrapper.delete_parameter_group(parameter_group_name)
def run_scenario(self, db_engine, parameter_group_name, instance_name, db_name):
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
print("-" * 88)
print(
"Welcome to the Amazon Relational Database Service (Amazon RDS)\n"
"get started with DB instances demo."
)
print("-" * 88)
parameter_group = self.create_parameter_group(parameter_group_name, db_engine)
self.update_parameters(parameter_group_name)
db_inst = self.create_instance(
instance_name, db_name, db_engine, parameter_group
)
self.display_connection(db_inst)
self.create_snapshot(instance_name)
self.cleanup(db_inst, parameter_group_name)
print("\nThanks for watching!")
print("-" * 88)
if __name__ == "__main__":
try:
scenario = RdsInstanceScenario(InstanceWrapper.from_client())
scenario.run_scenario(
"mysql",
"doc-example-parameter-group",
"doc-example-instance",
"docexampledb",
)
except Exception:
logging.exception("Something went wrong with the demo.")
```
Defina as funções que são chamadas pelo cenário para gerenciar as ações do Amazon RDS.
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 parameter group.
:param parameter_group_name: The name of the parameter group to retrieve.
:return: The parameter group.
"""
try:
response = self.rds_client.describe_db_parameter_groups(
DBParameterGroupName=parameter_group_name
)
parameter_group = response["DBParameterGroups"][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 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_parameter_group(
DBParameterGroupName=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 parameter group.
:param parameter_group_name: The name of the parameter group to delete.
:return: Data about the parameter group.
"""
try:
self.rds_client.delete_db_parameter_group(
DBParameterGroupName=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
def get_parameters(self, parameter_group_name, name_prefix="", source=None):
"""
Gets the parameters that are contained in a DB 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 = {"DBParameterGroupName": parameter_group_name}
if source is not None:
kwargs["Source"] = source
parameters = []
paginator = self.rds_client.get_paginator("describe_db_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 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_parameter_group(
DBParameterGroupName=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 create_snapshot(self, snapshot_id, instance_id):
"""
Creates a snapshot of a DB instance.
:param snapshot_id: The ID to give the created snapshot.
:param instance_id: The ID of the DB instance to snapshot.
:return: Data about the newly created snapshot.
"""
try:
response = self.rds_client.create_db_snapshot(
DBSnapshotIdentifier=snapshot_id, DBInstanceIdentifier=instance_id
)
snapshot = response["DBSnapshot"]
except ClientError as err:
logger.error(
"Couldn't create snapshot of %s. Here's why: %s: %s",
instance_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return snapshot
def get_snapshot(self, snapshot_id):
"""
Gets a DB instance snapshot.
:param snapshot_id: The ID of the snapshot to retrieve.
:return: The retrieved snapshot.
"""
try:
response = self.rds_client.describe_db_snapshots(
DBSnapshotIdentifier=snapshot_id
)
snapshot = response["DBSnapshots"][0]
except ClientError as err:
logger.error(
"Couldn't get snapshot %s. Here's why: %s: %s",
snapshot_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return snapshot
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 create_db_instance(
self,
db_name,
instance_id,
parameter_group_name,
db_engine,
db_engine_version,
instance_class,
storage_type,
allocated_storage,
admin_name,
admin_password,
):
"""
Creates a DB instance.
:param db_name: The name of the database that is created in the DB instance.
:param instance_id: The ID to give the newly created DB instance.
:param parameter_group_name: A parameter group to associate with the DB instance.
:param db_engine: The database engine of a database to create in the DB instance.
:param db_engine_version: The engine version for the created database.
:param instance_class: The DB instance class for the newly created DB instance.
:param storage_type: The storage type of the DB instance.
:param allocated_storage: The amount of storage allocated on the DB instance, in GiBs.
:param admin_name: The name of the admin user for the created database.
:param admin_password: The admin password for the created database.
:return: Data about the newly created DB instance.
"""
try:
response = self.rds_client.create_db_instance(
DBName=db_name,
DBInstanceIdentifier=instance_id,
DBParameterGroupName=parameter_group_name,
Engine=db_engine,
EngineVersion=db_engine_version,
DBInstanceClass=instance_class,
StorageType=storage_type,
AllocatedStorage=allocated_storage,
MasterUsername=admin_name,
MasterUserPassword=admin_password,
)
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 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 obter detalhes da API, consulte os tópicos a seguir na *Referência da API AWS SDK para Python (Boto3)*.
+ [CriarDBInstance](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBInstance)
+ [Criar DBParameter grupo](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBParameterGroup)
+ [CriarDBSnapshot](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBSnapshot)
+ [ExcluirDBInstance](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DeleteDBInstance)
+ [Excluir DBParameter grupo](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DeleteDBParameterGroup)
+ [Descreva DBEngine as versões](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBEngineVersions)
+ [DescreverDBInstances](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBInstances)
+ [Descrever DBParameter grupos](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBParameterGroups)
+ [DescreverDBParameters](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBParameters)
+ [DescreverDBSnapshots](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBSnapshots)
+ [DescribeOrderableDBInstanceOpções](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeOrderableDBInstanceOptions)
+ [Modificar DBParameter grupo](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/ModifyDBParameterGroup)
## Ações
### `CreateDBInstance`
O código de exemplo a seguir mostra como usar `CreateDBInstance`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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_instance(
self,
db_name,
instance_id,
parameter_group_name,
db_engine,
db_engine_version,
instance_class,
storage_type,
allocated_storage,
admin_name,
admin_password,
):
"""
Creates a DB instance.
:param db_name: The name of the database that is created in the DB instance.
:param instance_id: The ID to give the newly created DB instance.
:param parameter_group_name: A parameter group to associate with the DB instance.
:param db_engine: The database engine of a database to create in the DB instance.
:param db_engine_version: The engine version for the created database.
:param instance_class: The DB instance class for the newly created DB instance.
:param storage_type: The storage type of the DB instance.
:param allocated_storage: The amount of storage allocated on the DB instance, in GiBs.
:param admin_name: The name of the admin user for the created database.
:param admin_password: The admin password for the created database.
:return: Data about the newly created DB instance.
"""
try:
response = self.rds_client.create_db_instance(
DBName=db_name,
DBInstanceIdentifier=instance_id,
DBParameterGroupName=parameter_group_name,
Engine=db_engine,
EngineVersion=db_engine_version,
DBInstanceClass=instance_class,
StorageType=storage_type,
AllocatedStorage=allocated_storage,
MasterUsername=admin_name,
MasterUserPassword=admin_password,
)
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 obter detalhes da API, consulte Referência da API [Create DBInstance](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBInstance) in *AWS SDK for Python (Boto3*).
### `CreateDBParameterGroup`
O código de exemplo a seguir mostra como usar `CreateDBParameterGroup`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 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_parameter_group(
DBParameterGroupName=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 obter detalhes da API, consulte Referência da API [Create DBParameter Group](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBParameterGroup) in *AWS SDK for Python (Boto3*).
### `CreateDBSnapshot`
O código de exemplo a seguir mostra como usar `CreateDBSnapshot`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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_snapshot(self, snapshot_id, instance_id):
"""
Creates a snapshot of a DB instance.
:param snapshot_id: The ID to give the created snapshot.
:param instance_id: The ID of the DB instance to snapshot.
:return: Data about the newly created snapshot.
"""
try:
response = self.rds_client.create_db_snapshot(
DBSnapshotIdentifier=snapshot_id, DBInstanceIdentifier=instance_id
)
snapshot = response["DBSnapshot"]
except ClientError as err:
logger.error(
"Couldn't create snapshot of %s. Here's why: %s: %s",
instance_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return snapshot
```
+ Para obter detalhes da API, consulte Referência da API [Create DBSnapshot](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/CreateDBSnapshot) in *AWS SDK for Python (Boto3*).
### `DeleteDBInstance`
O código de exemplo a seguir mostra como usar `DeleteDBInstance`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 obter detalhes da API, consulte a Referência da API [Excluir DBInstance](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DeleteDBInstance) no *AWS SDK for Python (Boto3*).
### `DeleteDBParameterGroup`
O código de exemplo a seguir mostra como usar `DeleteDBParameterGroup`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 parameter group.
:param parameter_group_name: The name of the parameter group to delete.
:return: Data about the parameter group.
"""
try:
self.rds_client.delete_db_parameter_group(
DBParameterGroupName=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
```
+ Para obter detalhes da API, consulte a Referência da API [Excluir DBParameter grupo](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DeleteDBParameterGroup) no *AWS SDK for Python (Boto3*).
### `DescribeDBEngineVersions`
O código de exemplo a seguir mostra como usar `DescribeDBEngineVersions`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 obter detalhes da API, consulte [Descrição das DBEngine versões](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBEngineVersions) na referência da API *AWS SDK for Python (Boto3*).
### `DescribeDBInstances`
O código de exemplo a seguir mostra como usar `DescribeDBInstances`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 obter detalhes da API, consulte a Referência da API [Descrever DBInstances](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBInstances) no *AWS SDK for Python (Boto3*).
### `DescribeDBParameterGroups`
O código de exemplo a seguir mostra como usar `DescribeDBParameterGroups`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 parameter group.
:param parameter_group_name: The name of the parameter group to retrieve.
:return: The parameter group.
"""
try:
response = self.rds_client.describe_db_parameter_groups(
DBParameterGroupName=parameter_group_name
)
parameter_group = response["DBParameterGroups"][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 obter detalhes da API, consulte a Referência da API [Descrever DBParameter grupos](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBParameterGroups) no *AWS SDK for Python (Boto3*).
### `DescribeDBParameters`
O código de exemplo a seguir mostra como usar `DescribeDBParameters`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 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 = {"DBParameterGroupName": parameter_group_name}
if source is not None:
kwargs["Source"] = source
parameters = []
paginator = self.rds_client.get_paginator("describe_db_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 obter detalhes da API, consulte a Referência da API [Descrever DBParameters](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBParameters) no *AWS SDK for Python (Boto3*).
### `DescribeDBSnapshots`
O código de exemplo a seguir mostra como usar `DescribeDBSnapshots`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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_snapshot(self, snapshot_id):
"""
Gets a DB instance snapshot.
:param snapshot_id: The ID of the snapshot to retrieve.
:return: The retrieved snapshot.
"""
try:
response = self.rds_client.describe_db_snapshots(
DBSnapshotIdentifier=snapshot_id
)
snapshot = response["DBSnapshots"][0]
except ClientError as err:
logger.error(
"Couldn't get snapshot %s. Here's why: %s: %s",
snapshot_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return snapshot
```
+ Para obter detalhes da API, consulte a Referência da API [Descrever DBSnapshots](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeDBSnapshots) no *AWS SDK for Python (Boto3*).
### `DescribeOrderableDBInstanceOptions`
O código de exemplo a seguir mostra como usar `DescribeOrderableDBInstanceOptions`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 obter detalhes da API, consulte [DescribeOrderableDBInstanceReferência da API Opções](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/DescribeOrderableDBInstanceOptions) no *AWS SDK for Python (Boto3*).
### `ModifyDBParameterGroup`
O código de exemplo a seguir mostra como usar `ModifyDBParameterGroup`.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rds#code-examples).
```
class InstanceWrapper:
"""Encapsulates Amazon RDS DB instance actions."""
def __init__(self, rds_client):
"""
:param rds_client: A Boto3 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 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_parameter_group(
DBParameterGroupName=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 obter detalhes da API, consulte Referência da API [Modificar DBParameter grupo](https://docs.aws.amazon.com/goto/boto3/rds-2014-10-31/ModifyDBParameterGroup) no *AWS SDK for Python (Boto3*).
## Cenários
### Crie um rastreador de itens de trabalho do Aurora Sem Servidor
O exemplo de código a seguir mostra como criar uma aplicação Web que rastreia os itens de trabalho em um banco de dados do Amazon Aurora Sem Servidor e usa o Amazon Simple Email Service (Amazon SES) para enviar relatórios.
**SDK para Python (Boto3)**
Mostra como usar o AWS SDK para Python (Boto3) para criar um serviço REST que rastreia itens de trabalho em um banco de dados Amazon Aurora Serverless e envia relatórios por e-mail usando o Amazon Simple Email Service (Amazon SES). Este exemplo usa a estrutura web Flask para lidar com o roteamento HTTP e se integra a uma página da Web do React para apresentar uma aplicação Web totalmente funcional.
+ Crie um serviço Flask REST que se integre com o. Serviços da AWS
+ Leia, grave e atualize itens de trabalho armazenados em um banco de dados do Aurora Sem Servidor.
+ Crie um AWS Secrets Manager segredo que contenha as credenciais do banco de dados e use-o para autenticar chamadas para o banco de dados.
+ Use o Amazon SES para enviar relatórios por e-mail de itens de trabalho.
Para obter o código-fonte completo e instruções sobre como configurar e executar, veja o exemplo completo em [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/aurora_item_tracker).
**Serviços usados neste exemplo**
+ Aurora
+ Amazon RDS
+ Serviços de dados do Amazon RDS
+ Amazon SES
## Exemplos sem servidor
### Como se conectar a um banco de dados do Amazon RDS em uma função do Lambda
O exemplo de código a seguir mostra como implementar uma função do Lambda que se conecte a um banco de dados do RDS. A função faz uma solicitação simples ao banco de dados e exibe o resultado.
**SDK para Python (Boto3)**
Tem mais sobre GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam).
Conectar-se a um banco de dados do Amazon RDS em uma função do Lambda usando Python.
```
import json
import os
import boto3
import pymysql
# RDS settings
proxy_host_name = os.environ['PROXY_HOST_NAME']
port = int(os.environ['PORT'])
db_name = os.environ['DB_NAME']
db_user_name = os.environ['DB_USER_NAME']
aws_region = os.environ['AWS_REGION']
# Fetch RDS Auth Token
def get_auth_token():
client = boto3.client('rds')
token = client.generate_db_auth_token(
DBHostname=proxy_host_name,
Port=port
DBUsername=db_user_name
Region=aws_region
)
return token
def lambda_handler(event, context):
token = get_auth_token()
try:
connection = pymysql.connect(
host=proxy_host_name,
user=db_user_name,
password=token,
db=db_name,
port=port,
ssl={'ca': 'Amazon RDS'} # Ensure you have the CA bundle for SSL connection
)
with connection.cursor() as cursor:
cursor.execute('SELECT %s + %s AS sum', (3, 2))
result = cursor.fetchone()
return result
except Exception as e:
return (f"Error: {str(e)}") # Return an error message if an exception occurs
```