如何检查 DynamoDB 表是否存在？

我是 boto3的一个新用户，我正在使用 DynamoDB。

我查看了 DynamoDB api，但是找不到任何能够告诉我表是否已经存在的方法。

处理这个问题的最佳方法是什么？

我是否应该尝试创建一个新表并使用 try catch 包装它？

78675 次浏览
小开

可以使用 描述表 API 确定表是否存在。

示例代码:

from __future__ import print_function # Python 2/3 compatibility
import os
os.environ["TZ"] = "UTC"
import boto3


client = boto3.client('dynamodb', region_name='us-west-2', endpoint_url="http://localhost:8000")






response = client.describe_table(
TableName='Movies'
)


print(response)

如果表存在:-

  • 你会得到回应的

如果表不存在:-

  • 你会得到 ResourceNotFoundException

    异常: 发生错误(ResourceNotF 调用 DescribeTable 操作时: 不能对 一张不存在的桌子

另一种方式:-

或者，您可以使用文档中的 table.wait_until_exists():

等待直到此表存在。此方法调用 Table _ vis. wait () ，它轮询。 每20秒执行一次 DynamoDB.Client.description _ table () ，直到成功执行 状态。在25次检查失败后返回一个错误。

参见: AWS API 文档

请求语法

table.wait_until_exists()

报税表

None

小开
最佳答案

通过阅读文档，我可以看到有三种方法可以检查表是否存在。

  1. 如果表已经存在，则 CreateTable API抛出错误 ResourceInUseException。用 try 包装 create _ table 方法，但是要捕捉这个
  2. 可以使用 清单表 API获取与当前帐户和端点关联的表名列表。检查表名是否出现在响应中获得的表名列表中。
  3. 如果请求的表名不存在，则 描述表 API将抛出错误 ResourceNotFoundException。

对我来说，如果您只想创建一个表，第一个选项听起来更好。

编辑: 我发现有些人发现很难捕捉到例外。我将在下面提供一些代码，让您了解如何在 boto3中处理异常。

例子一

import boto3


dynamodb_client = boto3.client('dynamodb')


try:
response = dynamodb_client.create_table(
AttributeDefinitions=[
{
'AttributeName': 'Artist',
'AttributeType': 'S',
},
{
'AttributeName': 'SongTitle',
'AttributeType': 'S',
},
],
KeySchema=[
{
'AttributeName': 'Artist',
'KeyType': 'HASH',
},
{
'AttributeName': 'SongTitle',
'KeyType': 'RANGE',
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
},
TableName='test',
)
except dynamodb_client.exceptions.ResourceInUseException:
# do something here as you require
pass

例子2

import boto3


dynamodb_client = boto3.client('dynamodb')




table_name = 'test'
existing_tables = dynamodb_client.list_tables()['TableNames']


if table_name not in existing_tables:
response = dynamodb_client.create_table(
AttributeDefinitions=[
{
'AttributeName': 'Artist',
'AttributeType': 'S',
},
{
'AttributeName': 'SongTitle',
'AttributeType': 'S',
},
],
KeySchema=[
{
'AttributeName': 'Artist',
'KeyType': 'HASH',
},
{
'AttributeName': 'SongTitle',
'KeyType': 'RANGE',
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
},
TableName=table_name,
)

例子3

import boto3


dynamodb_client = boto3.client('dynamodb')


try:
response = dynamodb_client.describe_table(TableName='test')
except dynamodb_client.exceptions.ResourceNotFoundException:
# do something here as you require
pass
小开 
import boto3


from botocore.exceptions import ClientError


TABLE_NAME = "myTableName"
dynamodb = boto3.resource('dynamodb', endpoint_url="https://dynamodb.us-east-1.amazonaws.com")


table = dynamodb.Table(TABLE_NAME)


try:
response = client.describe_table(TableName=TABLE_NAME)


except ClientError as ce:
if ce.response['Error']['Code'] == 'ResourceNotFoundException':
print "Table " + TABLE_NAME + " does not exist. Create the table first and try again."
else:
print "Unknown exception occurred while querying for the " + TABLE_NAME + " table. Printing full error:"
pprint.pprint(ce.response)
小开

您可以使用任何 boto3 Table 实例对象的 . table _ status attr。如果存在，则返回其状态(CREATING、 UPDATING、 DELEING、 Aactive)或抛出异常 botocore.exceptions.ClientError: Requested resource not found: Table: <YOUR_TABLE_NAME> not found。您可以将这些条件包装到 try/but 中，以获得关于当前表状态的完整信息。

import boto3
from botocore.exceptions import ClientError


dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
table = dynamodb.Table('your_table_name_str')


try:
is_table_existing = table.table_status in ("CREATING", "UPDATING",
"DELETING", "ACTIVE")
except ClientError:
is_table_existing = False
print "Table %s doesn't exist." % table.name
小开

如果您不想使用 boto3.client而只使用 boto3.resource，可以选择其他方法:

import boto3


database = boto3.resource('dynamodb', endpoint_url="http://localhost:8000")


table_name  = 'MyTable'
table_names = [table.name for table in database.tables.all()]


if table_name in table_names:
print('table', table_name, 'exists')
小开

请注意，这取决于您是使用客户端还是使用资源。如果您使用 boto3.client()，您可以使用3个方法的接受答案建议。如果使用 boto3.resource()，则只能使用 dynamodb_resource.create_table()并检查异常。

try:
table = dynamodb_resource.create_table(
...
)
table.meta.client.get_waiter('table_exists').wait(TableName=your_table_name)
except ResourceInUseException:
# do sth here
小开

我知道，如果在 table.split (0，-1)和 table 之间有超过10个表，那么这样做的风险会很小。但是，它确实保存了抛出异常等等。 遗憾的是，文档 https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_ListTables.html(示例)暗示返回列表中的第一个表将是搜索项，但事实并非如此。.

类 Dynamo {

private m_db!: DynamoDB;


private async ensure_table(name: string) {
const search = await this.db().listTables({
ExclusiveStartTableName: name.slice(0, -1),
Limit: 10 });
const exists = search.TableNames?.includes(name);
exists || await this.create_table(name);
}


private async create_table(name: string) {
// create the table here
}


 

private db(): DynamoDB {
return this.m_db || (this.m_db = this.create_db());
}




private create_db = (): DynamoDB => {
return new DynamoDB({apiVersion: "2012-08-10"}); }
}

}

小开

您可以使用方便的资源 API，同时仍然处理和捕获客户端 API 级别的异常，因为您可以从资源访问客户端！这使得检查 table 是否存在的方法成为我所发现的最优雅的方法:

resource = boto3.resource('dynamodb', region_name='eu-north-1')


def table_exists(table_name: str) -> bool:
try:
resource.Table(table_name).table_status
except resource.meta.client.exceptions.ResourceNotFoundException:
return False
return True


print(table_exists('dummy_table'))

提交答案