如何在AWS AppSync中创建自定义硬编码健康检查GraphQL查询

核心结论 #

是的，你必须在GraphQL Schema中定义对应的healthCheck查询——GraphQL是强类型查询语言，AppSync所有可执行的操作都必须在Schema里明确声明，否则无法被客户端调用，也无法绑定解析器。

1. 定义GraphQL Schema中的Health-Check结构 #

首先要定义查询的返回类型和查询字段，根据你需要的信息调整字段：

# 定义健康检查的返回类型
type HealthCheckResult {
  version: String!       # 从pyproject.toml读取的API版本号
  branch: String!        # 代码分支信息
  status: String!        # 服务健康状态
  timestamp: String!     # 响应时间戳
}

# 在根Query类型中添加healthCheck查询入口
type Query {
  healthCheck: HealthCheckResult!
}

如果需要监控其他服务状态（比如数据库连通性），可以在HealthCheckResult中新增对应字段（如databaseStatus: String）。

2. 用Pulumi Python实现解析器 #

健康检查不需要调用外部数据源，直接用无数据源解析器即可，通过响应模板返回预注入的信息：

步骤1：读取pyproject.toml的版本信息

先安装依赖库（Python 3.11+可直接用标准库tomllib）：

pip install tomli

步骤2：Pulumi代码实现

import pulumi
import pulumi_aws as aws
import tomli
from datetime import datetime

# 读取pyproject.toml中的版本信息（以Poetry管理为例，根据实际配置调整）
with open("pyproject.toml", "rb") as f:
    pyproject_data = tomli.load(f)
api_version = pyproject_data["tool"]["poetry"]["version"]

# 从Pulumi配置或环境变量获取分支信息（建议CI/CD注入）
branch_name = pulumi.Config().get("branch") or "local-dev"

# 创建AppSync API
appsync_api = aws.appsync.GraphqlApi(
    "health-check-api",
    authentication_type="API_KEY",
    schema=pulumi.FileAsset("schema.graphql")  # 加载上面定义的Schema文件
)

# 绑定healthCheck查询的无数据源解析器
health_check_resolver = aws.appsync.Resolver(
    "health-check-resolver",
    api_id=appsync_api.id,
    type_name="Query",
    field_name="healthCheck",
    data_source_name="NONE",  # 无数据源标识
    # 响应模板：直接返回预注入的静态数据
    response_template=f"""
    {{
      "version": "{api_version}",
      "branch": "{branch_name}",
      "status": "healthy",
      "timestamp": "{datetime.utcnow().isoformat()}Z"
    }}
    """
)

# 输出测试用的API端点和密钥
pulumi.export("appsync_api_endpoint", appsync_api.uris["GRAPHQL"])
pulumi.export("appsync_api_key", appsync_api.api_key)

3. 测试Health-Check查询 #

部署完成后，在AppSync控制台的查询编辑器发送请求：

query GetHealthCheck {
  healthCheck {
    version
    branch
    status
    timestamp
  }
}

会返回类似结果：

{
  "data": {
    "healthCheck": {
      "version": "1.0.0",
      "branch": "main",
      "status": "healthy",
      "timestamp": "2024-05-20T12:34:56.789Z"
    }
  }
}

扩展说明 #

如果需要动态检查其他服务状态（比如数据库连通性），可以将解析器绑定到Lambda数据源，在Lambda中执行健康检查逻辑后返回结果；分支信息建议通过CI/CD流水线注入到Pulumi配置中，避免硬编码。

内容的提问来源于stack exchange，提问作者bolino