文档简介：
操作场景
前提条件
操作步骤
步骤1：创建修改响应体的函数
步骤2：创建自定义响应体插件
|
参数
|
是否必填
|
说明
|
|
选择函数
|
必填
|
选择修改响应体的函数所在的命名空间、名称和版本。
|
|
Base64 编码
|
必填
|
业务后端响应内容传到云函数中时是否要经过 Base64 编码，一般适用于响应内容是二进制的情况。
|
|
后端超时
|
必填
|
设置 API 网关转发到修改响应体的函数的后端超时时间，超时时间的最大限制为30分钟。在 API 网关调用修改响应体的函数，未在超时时间内获得响应时，API 网关将终止此次调用，并返回相应的错误信息。
|
|
自定义内容
|
必填
|
设置 API 网关发送给修改响应体的函数的响应内容，支持选择 Header、Body、状态码。未选择的响应内容部分将不被修改，直接转发给客户端。
|
|
参数
|
是否必填
|
说明
|
|
请求方法
|
必填
|
请求自定义响应体函数的方法，支持 GET、POST、PUT、DELETE、HEAD、ANY。
|
|
公网服务
|
必填
|
自定义响应体改写服务访问地址，支持 HTTP 和 HTTPS 协议。
|
|
路径匹配模式
|
必填
|
支持后端路径匹配和全路径匹配两种方式。
后端路径匹配：直接使用配置的路径请求服务。
全路径匹配：使用去除请求路径的路径请求服务，如 API 路径配置为 /a/，请求路径为 /a/b，开启全路径匹配后，传输给服务的为 /b。
|
|
参数
|
是否必填
|
说明
|
|
选择 VPC
|
必填
|
选择响应体改写服务所属的 VPC。
|
|
请求方法
|
必填
|
请求响应体改写函数的方法，支持 GET、POST、PUT、DELETE、HEAD、ANY。
|
|
后端地址
|
必填
|
响应体改写服务访问地址，支持 HTTP 和 HTTPS 协议。
|
步骤3：绑定 API
pluginData
{"endpoint_timeout":15, // 后端超时时间，单位秒，合法值：0 ~ 60 秒"func_name":"test_name", // 自定义函数名称"func_namespace":"test_namespace", // 自定义函数命名空间"func_qualifier":"$LATEST", // 自定义函数版本"is_base64_encoded":true, // 业务后端响应内容传到云函数中时是否要经过Base64编码"is_custom_status":true, // 是否将响应的状态码内容发送到函数"is_custom_headers":true, // 是否将响应的Header内容发送到函数"is_custom_body":true, // 是否将响应的Body内容发送到函数"user_id":1253970226 // appid}
注意事项
自定义服务类型规则
示例代码
from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse, parse_qs import json
PORT_NUMBER = 8022 def handle_header(header, response): replace_headers = {}
# 需要替换或者新增的header remove_headers = [] # 需要删除的header # 示例：
替换或新增 header1 replace_headers["header1"] = "header1" # 示例：删除 X-Api-
Serviceid remove_headers.append("X-Api-Serviceid") response["replace_headers"] = replace_headers response
response["remove_headers"] = remove_headers def handle_body(body, header, response): response["replace_body"] = "replace_body" # 是否需要对replace_body 进行base64解码
response["is_base64_encoded"] = header.get("is_base64_encoded") def handle_status
(header, response): status = header.get("status") # 示例：后端错误处理 if status
is not None and status != 200: response["replace_status"] = 200 response["replace_body"]
= "custom response, upstream_status is " + str(status) response["is_base64_encoded"]
= False class MyHandler(BaseHTTPRequestHandler): def _set_headers(self): self.send_
response(200) self.send_header('Content-type', 'application/json') self.end_headers()
def do_POST(self): response = {} try: request_body = self.rfile.read(int(self.headers.
getheader("Content-length"))) except: request_body = "no body." # 重写header handle_header
(self.headers.__dict__, response) # 重写status handle_status(self.headers.__dict__, response)
# 重写body handle_body(request_body, self.headers.__dict__, response) self._set_headers()
self.wfile.write(json.dumps(response).encode('utf-8')) return def do_GET(self): response =
{ 'api-auth': True } # header认证 header_auth_key = self.headers.get("header-auth")
if header_auth_key == "apigw": response['api-auth'] = False # query认证 parsed_url =
urlparse(self.path) query_params = parse_qs(parsed_url.query) query_auth_key = query_
params.get("query-auth", [None])[0] if query_auth_key == "apigw": response['api-auth']
= False self._set_headers() self.wfile.write(json.dumps(response).encode('utf-8'))
return try: server = HTTPServer(('', PORT_NUMBER), MyHandler) print(f'Started HTTP
server on port {PORT_NUMBER}') # Wait forever for incoming http requests server.serve_
forever() except KeyboardInterrupt: print('^C received, shutting down the web server')
server.socket.close()
