工具函数
约 2219 字大约 7 分钟
2025-08-11
概述
shared 目录包含了边缘函数的共享工具模块，提供了 CORS 处理、响应格式化、数据验证、Supabase 客户端创建等通用功能。
文件结构
supabase/functions/shared/
├── index.ts # 模块统一导出
├── cors.ts # CORS 跨域处理
├── response.ts # 统一响应格式
├── supabase.ts # Supabase 客户端
└── validation.ts # 数据验证工具模块详情
1. CORS 处理模块 (cors.ts)
corsHeaders
默认 CORS 响应头配置。
export const corsHeaders = {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Headers': 'authorization, x-client-info, apikey, content-type',
'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
}handleCors(request)
处理 CORS 预检请求。
参数:
request: Request- 请求对象
返回值:
Response | null- OPTIONS 请求返回 CORS 响应，否则返回 null
使用示例:
const corsResponse = handleCors(req)
if (corsResponse) return corsResponsecreateCorsHeaders(additionalHeaders)
创建带有 CORS 头的响应头对象。
参数:
additionalHeaders?: Record<string, string>- 额外的响应头
返回值:
Record<string, string>- 合并了 CORS 头的响应头对象
2. 响应格式模块 (response.ts)
ApiResponse 接口
统一的 API 响应数据接口。
export interface ApiResponse<T = any> {
success: boolean
data?: T
error?: string
message?: string
}成功响应函数
createSuccessResponse(data, message, status)
创建成功响应。
参数:
data: T- 响应数据message?: string- 成功消息status?: number- HTTP 状态码，默认 200
返回值:
Response- 格式化的成功响应
使用示例:
return createSuccessResponse({
user: userData,
session: sessionData
}, '登录成功')错误响应函数
createErrorResponse(error, status)
创建基础错误响应。
参数:
error: string- 错误消息status?: number- HTTP 状态码，默认 400
createValidationError(error)
创建验证错误响应（400）。
createUnauthorizedResponse(error)
创建未授权响应（401）。
createForbiddenResponse(error)
创建禁止访问响应（403）。
createNotFoundResponse(error)
创建资源不存在响应（404）。
createConflictResponse(error)
创建冲突响应（409）。
createServerErrorResponse(error)
创建服务器错误响应（500）。
safeAsyncHandler(operation, errorMessage)
安全地处理异步操作并返回适当的响应。
参数:
operation: () => Promise<Response>- 要执行的异步操作errorMessage?: string- 自定义错误消息，默认 '操作失败'
返回值:
Promise<Response>- 安全包装的响应
使用示例:
return await safeAsyncHandler(async () => {
// 业务逻辑
return createSuccessResponse(data)
}, '登录失败')3. Supabase 客户端模块 (supabase.ts)
createSupabaseClient()
创建具有完整权限的 Supabase 客户端实例。
环境变量:
SUPABASE_URL- Supabase 项目 URLSUPABASE_SERVICE_ROLE_KEY- 服务角色密钥
返回值:
SupabaseClient- 具有完整数据库访问权限的客户端
使用示例:
const supabase = createSupabaseClient()
const { data, error } = await supabase
.from('user_info')
.select('*')
.eq('id', userId)createSupabaseAnonClient()
创建匿名权限的 Supabase 客户端实例。
环境变量:
SUPABASE_URL- Supabase 项目 URLSUPABASE_ANON_KEY- 匿名密钥
返回值:
SupabaseClient- 受 RLS 策略限制的客户端
4. 数据验证模块 (validation.ts)
isValidEmail(email)
邮箱格式验证。
参数:
email: string- 邮箱地址
返回值:
boolean- 是否为有效邮箱格式
正则表达式:
const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/isValidPassword(password, minLength)
密码强度验证。
参数:
password: string- 密码minLength?: number- 最小长度，默认 6
返回值:
boolean- 是否满足强度要求
isValidUserRole(role)
用户角色验证。
参数:
role: string- 用户角色
返回值:
role is 'admin' | 'student' | 'teacher'- 是否为有效角色
有效角色:
admin- 管理员student- 学生teacher- 教师
validateRequiredFields(fields)
必填字段验证。
参数:
fields: Record<string, any>- 字段对象
返回值:
{
isValid: boolean
missingFields: string[]
}使用示例:
const { isValid, missingFields } = validateRequiredFields({
email,
password,
full_name
})
if (!isValid) {
return createValidationError(`缺少必填字段: ${missingFields.join(', ')}`)
}isValidPhoneNumber(phone)
手机号格式验证（中国大陆）。
参数:
phone: string- 手机号
返回值:
boolean- 是否为有效手机号格式
正则表达式:
const phoneRegex = /^1[3-9]\d{9}$/isValidUUID(uuid)
UUID 格式验证。
参数:
uuid: string- UUID 字符串
返回值:
boolean- 是否为有效 UUID 格式
正则表达式:
const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i使用指南
导入模块
import {
createSupabaseClient,
handleCors,
createSuccessResponse,
createValidationError,
safeAsyncHandler,
validateRequiredFields,
isValidEmail
} from '../shared/index.ts'标准边缘函数模式
Deno.serve(async (req) => {
// 1. 处理 CORS
const corsResponse = handleCors(req)
if (corsResponse) return corsResponse
// 2. 安全处理业务逻辑
return await safeAsyncHandler(async () => {
// 3. 创建客户端
const supabase = createSupabaseClient()
// 4. 解析请求数据
const requestData = await req.json()
// 5. 验证必填字段
const { isValid, missingFields } = validateRequiredFields(requestData)
if (!isValid) {
return createValidationError(`缺少必填字段: ${missingFields.join(', ')}`)
}
// 6. 具体业务逻辑
// ...
// 7. 返回成功响应
return createSuccessResponse(result, '操作成功')
}, '操作失败')
})设计原则
1. 模块化设计
- 每个模块负责特定功能
- 通过 index.ts 统一导出
- 便于维护和测试
2. 类型安全
- 使用 TypeScript 严格类型检查
- 定义明确的接口和类型
- 提供类型保护函数
3. 错误处理
- 统一的错误响应格式
- 详细的错误分类
- 安全的异常处理
4. 可复用性
- 通用的工具函数
- 标准化的处理流程
- 一致的代码风格
5. 安全性
- 输入数据验证
- CORS 安全配置
- 敏感信息保护
使用示例
cURL 请求
# 示例：测试 CORS 处理
curl -X OPTIONS http://127.0.0.1:54321/functions/v1/your-function \
-H "Access-Control-Request-Method: POST" \
-H "Access-Control-Request-Headers: authorization, content-type" \
-H "Origin: http://localhost:3000"
# 示例：测试边缘函数响应格式
curl -X POST http://127.0.0.1:54321/functions/v1/your-function \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ANON_KEY" \
-d '{"test": "data"}'JavaScript 客户端
使用共享工具函数创建边缘函数
// 在边缘函数中使用共享工具
import {
createSupabaseClient,
handleCors,
createSuccessResponse,
createValidationError,
safeAsyncHandler,
validateRequiredFields,
isValidEmail
} from '../shared/index.ts'
Deno.serve(async (req) => {
// 1. 处理 CORS
const corsResponse = handleCors(req)
if (corsResponse) return corsResponse
// 2. 安全处理业务逻辑
return await safeAsyncHandler(async () => {
const supabase = createSupabaseClient()
const requestData = await req.json()
// 验证必填字段
const { isValid, missingFields } = validateRequiredFields({
email: requestData.email,
name: requestData.name
})
if (!isValid) {
return createValidationError(`缺少必填字段: ${missingFields.join(', ')}`)
}
// 验证邮箱格式
if (!isValidEmail(requestData.email)) {
return createValidationError('邮箱格式不正确')
}
// 业务逻辑处理
const result = { message: 'Success', data: requestData }
return createSuccessResponse(result, '操作成功')
}, '操作失败')
})使用 Supabase 客户端调用带有工具函数的边缘函数
import { createClient } from '@supabase/supabase-js'
const supabase = createClient(
'your-project-url',
'your-anon-key'
)
// 调用使用了工具函数的边缘函数
const testUtilityFunction = async () => {
try {
const { data, error } = await supabase.functions.invoke('your-function', {
body: {
email: 'test@example.com',
name: '测试用户'
}
})
if (error) {
console.error('调用失败:', error)
return
}
// 处理标准化的响应格式
if (data.success) {
console.log('操作成功:', data.data)
console.log('消息:', data.message)
} else {
console.error('业务错误:', data.error)
}
} catch (err) {
console.error('网络错误:', err)
}
}
testUtilityFunction()Nuxt.js 中使用工具函数规范
<script setup>
const supabase = useSupabaseClient()
// 调用遵循工具函数规范的边缘函数
const callEdgeFunction = async (functionName, payload) => {
try {
const { data, error } = await supabase.functions.invoke(functionName, {
body: payload
})
if (error) {
throw error
}
// 所有使用共享工具的边缘函数都会返回标准格式
return {
success: data.success,
data: data.data,
message: data.message,
error: data.error
}
} catch (error) {
return {
success: false,
error: error.message,
data: null,
message: null
}
}
}
// 使用示例
const handleSubmit = async () => {
const result = await callEdgeFunction('your-function', {
email: 'user@example.com',
name: '用户姓名'
})
if (result.success) {
console.log('成功:', result.message)
} else {
console.error('失败:', result.error)
}
}
</script>Python 客户端
使用 Supabase Python 客户端
from supabase import create_client, Client
from typing import Dict, Any, Optional
class EdgeFunctionClient:
def __init__(self, url: str, key: str):
self.supabase: Client = create_client(url, key)
def call_function(self, function_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
调用边缘函数的通用方法
Args:
function_name: 边缘函数名称
payload: 请求数据
Returns:
标准化的响应格式
"""
try:
response = self.supabase.functions.invoke(
function_name,
invoke_options={"body": payload}
)
if response.status_code == 200:
result = response.json()
# 所有使用共享工具的边缘函数都返回标准格式
return {
"success": result.get("success", False),
"data": result.get("data"),
"message": result.get("message"),
"error": result.get("error")
}
else:
return {
"success": False,
"error": f"HTTP错误: {response.status_code}",
"data": None,
"message": None
}
except Exception as e:
return {
"success": False,
"error": str(e),
"data": None,
"message": None
}
def validate_and_call(self, function_name: str, payload: Dict[str, Any],
required_fields: list) -> Dict[str, Any]:
"""
验证字段并调用边缘函数
"""
# 本地验证（可选）
missing_fields = [field for field in required_fields if not payload.get(field)]
if missing_fields:
return {
"success": False,
"error": f"缺少必填字段: {', '.join(missing_fields)}",
"data": None,
"message": None
}
return self.call_function(function_name, payload)
# 使用示例
def main():
client = EdgeFunctionClient("your-project-url", "your-anon-key")
# 调用使用了工具函数的边缘函数
result = client.validate_and_call(
"your-function",
{
"email": "test@example.com",
"name": "测试用户"
},
["email", "name"]
)
if result["success"]:
print(f"✅ 成功: {result['message']}")
print(f"数据: {result['data']}")
else:
print(f"❌ 失败: {result['error']}")
if __name__ == "__main__":
main()FastAPI 应用集成工具函数规范
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr
from supabase import create_client, Client
from typing import Dict, Any, Optional
import os
app = FastAPI()
# 初始化 Supabase 客户端
supabase: Client = create_client(
os.getenv("SUPABASE_URL"),
os.getenv("SUPABASE_ANON_KEY")
)
class EdgeFunctionRequest(BaseModel):
email: EmailStr
name: str
additional_data: Optional[Dict[str, Any]] = None
class StandardResponse(BaseModel):
success: bool
message: Optional[str] = None
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
class EdgeFunctionService:
"""边缘函数调用服务"""
@staticmethod
def call_edge_function(function_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
"""调用边缘函数的通用方法"""
try:
response = supabase.functions.invoke(
function_name,
invoke_options={"body": payload}
)
if response.status_code == 200:
return response.json()
else:
return {
"success": False,
"error": f"边缘函数调用失败: HTTP {response.status_code}"
}
except Exception as e:
return {
"success": False,
"error": f"调用异常: {str(e)}"
}
@app.post("/api/test-utility", response_model=StandardResponse)
async def test_utility_function(request: EdgeFunctionRequest):
"""测试使用工具函数的边缘函数"""
# 构建请求数据
payload = {
"email": request.email,
"name": request.name
}
if request.additional_data:
payload.update(request.additional_data)
# 调用边缘函数
result = EdgeFunctionService.call_edge_function("your-function", payload)
if result.get("success"):
return StandardResponse(
success=True,
message=result.get("message", "操作成功"),
data=result.get("data")
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=result.get("error", "操作失败")
)
@app.get("/api/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "service": "Evoliant API"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)版权所有
版权归属：Evoliant
许可证：MIT