如何使用Boto3检查AWS Athena数据库是否已存在？

你遇到的这个异步执行问题太常见了——Athena的start_query_execution是异步操作，提交创建数据库的请求后不会立刻等它完成，直接跑建表代码自然会找不到数据库。别担心，用Boto3完全可以检查数据库的存在性，甚至可以等创建任务完成后再继续，下面给你两种实用方法：

方法1：用get_database API（精准检查） #

这个方法直接针对目标数据库查询，存在就返回数据库详情，不存在会抛出ResourceNotFoundException异常，我们可以用try-except来捕获判断：

import boto3
from botocore.exceptions import ClientError

client = boto3.client('athena')
db_name = 'your_db_name'

def check_database_exists(db_name):
    try:
        response = client.get_database(DatabaseName=db_name)
        # 数据库存在
        print(f"数据库 {db_name} 已存在")
        return True
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            # 数据库不存在
            print(f"数据库 {db_name} 不存在")
            return False
        else:
            # 其他错误，比如权限问题
            raise e

方法2：用list_databases API（批量查询后过滤） #

如果需要一次性查看多个数据库，或者不想处理异常，可以用这个方法列出所有数据库，再检查目标是否在列表中：

def check_database_exists_via_list(db_name):
    paginator = client.get_paginator('list_databases')
    for page in paginator.paginate(CatalogName='AwsDataCatalog'):
        for db in page['DatabaseList']:
            if db['Name'] == db_name:
                print(f"数据库 {db_name} 已存在")
                return True
    print(f"数据库 {db_name} 不存在")
    return False

结合你的创建流程优化 #

既然你是先创建数据库再建表，更稳妥的方式是：先检查数据库是否存在，不存在则提交创建请求，然后等待请求执行完成，再开始建表。这里给你补充等待查询完成的逻辑：

import time

def wait_for_query_completion(query_execution_id):
    while True:
        response = client.get_query_execution(QueryExecutionId=query_execution_id)
        status = response['QueryExecution']['Status']['State']
        if status in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
            return status
        # 每隔2秒轮询一次，可根据需求调整
        time.sleep(2)

# 你的原有创建数据库代码
config = {'OutputLocation': 's3://your-query-results-bucket/'}
db_name = 'db_name'

if not check_database_exists(db_name):
    # 提交创建数据库请求
    response = client.start_query_execution(
        QueryString=f'create database {db_name}',
        ResultConfiguration=config
    )
    query_execution_id = response['QueryExecutionId']
    # 等待创建完成
    status = wait_for_query_completion(query_execution_id)
    if status == 'SUCCEEDED':
        print("数据库创建完成，现在可以创建表了")
        # 在这里执行你的创建表逻辑
        # client.start_query_execution(QueryString='create table ...', ...)
    else:
        print(f"数据库创建失败，状态：{status}")
else:
    print("数据库已存在，直接创建表")
    # 执行创建表逻辑

这样就能彻底避免因为数据库创建未完成导致的建表失败问题啦。

内容的提问来源于stack exchange，提问作者Carlos P Ceballos