跳到主要内容

7 篇博文 含有标签「saas-development」

查看所有标签

Playwright page.route() 实现 API 全量 Mock

· 阅读需 4 分钟

在构建 AI Agent 平台时遇到此问题,记录根因与解法。

TL;DR

使用 page.route() 拦截所有 API 请求,返回预定义的 mock 数据。测试不依赖真实后端,可以在任何环境稳定运行,且避免创建/删除数据等副作用。

问题现象

E2E 测试调用真实 API:

// ❌ 依赖真实后端
test('create agent', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

// 点击创建按钮
await authenticatedPage.click('button:has-text("Create")')

// 填写表单
await authenticatedPage.fill('input[name="name"]', 'Test Agent')
await authenticatedPage.click('button[type="submit"]')

// 等待 API 响应
await authenticatedPage.waitForTimeout(2000)

// 验证... 但如果后端挂了?如果数据库连接失败?
})

问题:

  1. 依赖后端状态 — 后端挂了测试就失败
  2. 数据副作用 — 每次运行创建真实数据
  3. 不可重复 — 数据变化导致断言失败
  4. CI 环境问题 — 需要启动完整后端服务

解决方案

1. 定义 Mock 数据

// e2e/fixtures.ts
export const mockAgents = [
{
agent_id: 'agent-1',
user_id: 'test-user-id',
name: 'Test Agent 1',
skills: [],
mcp_tools: [],
llm_config: { provider: 'anthropic', model: 'claude-3-5-sonnet-20241022' },
risk_threshold: 'medium',
auto_confirm_low: true,
created_at: '2024-01-01T00:00:00Z',
},
]

export const mockSkills = [
{
skill_id: 'skill-1',
owner_id: 'test-user-id',
name: 'Test Skill',
system_prompt: 'You are helpful.',
is_public: false,
is_own: true,
},
]

2. 设置 API Mock

export async function setupMockApi(page: Page) {
// Mock GET /api/agents
await page.route('**/api/agents', async (route) => {
if (route.request().method() === 'GET') {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(mockAgents),
})
} else if (route.request().method() === 'POST') {
// 模拟创建
const body = route.request().postDataJSON()
const newAgent = {
agent_id: `agent-${Date.now()}`,
user_id: 'test-user-id',
name: body.name,
created_at: new Date().toISOString(),
...body,
}
mockAgents.push(newAgent)
await route.fulfill({
status: 201,
contentType: 'application/json',
body: JSON.stringify(newAgent),
})
}
})

// Mock GET/PATCH/DELETE /api/agents/:id
await page.route('**/api/agents/*', async (route) => {
const url = route.request().url()
const match = url.match(/\/api\/agents\/([^/]+)/)
const agentId = match?.[1]

if (route.request().method() === 'GET') {
const agent = mockAgents.find((a) => a.agent_id === agentId)
await route.fulfill({
status: agent ? 200 : 404,
contentType: 'application/json',
body: JSON.stringify(agent || { error: 'Not found' }),
})
} else if (route.request().method() === 'DELETE') {
const index = mockAgents.findIndex((a) => a.agent_id === agentId)
if (index !== -1) mockAgents.splice(index, 1)
await route.fulfill({ status: 204 })
}
})

// Mock /api/skills
await page.route('**/api/skills**', async (route) => {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(mockSkills),
})
})

// Mock /api/health
await page.route('**/api/health', async (route) => {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({ status: 'ok' }),
})
})
}

3. 在 Fixture 中使用

export const test = base.extend<{ authenticatedPage: Page }>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await setupMockApi(page) // 拦截所有 API
await use(page)
},
})

4. 测试中使用

// e2e/dashboard.spec.ts
import { test, expect, mockAgents } from './fixtures'

test('should display agent list', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

// API 被自动 mock,返回 mockAgents
const agentCards = authenticatedPage.locator('[data-testid="agent-card"]')

// 断言基于已知的 mock 数据
await expect(agentCards).toHaveCount(mockAgents.length)
})

test('should create new agent', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

await authenticatedPage.click('button:has-text("Create")')

const dialog = authenticatedPage.locator('[role="dialog"]')
await dialog.locator('input[name="name"]').fill('New Agent')
await dialog.locator('button[type="submit"]').click()

// POST /api/agents 被 mock,返回 201
await authenticatedPage.waitForTimeout(500)

// 验证 UI 更新
await expect(authenticatedPage.locator('text=New Agent')).toBeVisible()
})

核心技巧

URL 匹配模式

// 精确匹配
await page.route('**/api/agents', handler)

// 通配符匹配
await page.route('**/api/agents/**', handler)

// 正则匹配
await page.route(/\/api\/agents\/\d+/, handler)

读取请求体

await page.route('**/api/agents', async (route) => {
const body = route.request().postDataJSON()
console.log('Request body:', body)

// 根据请求内容返回不同响应
if (body.name === 'error-test') {
await route.fulfill({ status: 400, body: JSON.stringify({ error: 'Bad request' }) })
} else {
await route.fulfill({ status: 201, body: JSON.stringify({ id: 'new-id', ...body }) })
}
})

模拟错误场景

// 模拟网络错误
await page.route('**/api/agents', (route) => route.abort('failed'))

// 模拟超时
await page.route('**/api/agents', async (route) => {
await new Promise((r) => setTimeout(r, 30000))
route.continue()
})

// 模拟 500 错误
await page.route('**/api/agents', (route) =>
route.fulfill({ status: 500, body: JSON.stringify({ error: 'Internal error' }) })
)

部分放行

// 只 mock 特定 API,其他放行
await page.route('**/api/**', async (route) => {
const url = route.request().url()

if (url.includes('/api/agents')) {
await route.fulfill({ status: 200, body: JSON.stringify(mockAgents) })
} else {
await route.continue() // 其他 API 走真实请求
}
})

Mock 数据管理

// 集中管理所有 mock 数据
// e2e/fixtures.ts
export const mockData = {
agents: [...],
skills: [...],
apiKeys: [...],
tasks: [...],
}

// 每个测试前重置
test.beforeEach(() => {
Object.assign(mockData, JSON.parse(JSON.stringify(originalMockData)))
})

关键原则

  1. Mock 所有外部依赖 — API、OAuth、第三方服务
  2. 模拟真实数据结构 — Mock 数据应与 API 契约一致
  3. 覆盖成功和失败场景 — 200/400/500 都要测试
  4. 隔离测试数据 — 每个测试用独立的 mock 数据副本

对类似需求感兴趣?联系合作

Playwright Custom Fixtures 实现免登录测试

· 阅读需 4 分钟

在构建 AI Agent 平台时遇到此问题,记录根因与解法。

TL;DR

使用 Playwright 的 test.extend() 创建自定义 fixture,通过 page.addInitScript() 在页面加载前注入 auth token 到 localStorage。测试用 authenticatedPage 替代 page,自动获得登录状态,无需每个测试重复登录。

问题现象

E2E 测试需要验证登录后才能访问的页面:

// ❌ 每个测试都要走登录流程
test('dashboard', async ({ page }) => {
await page.goto('/login')
await page.fill('input[name="email"]', 'test@example.com')
await page.fill('input[name="password"]', 'password')
await page.click('button[type="submit"]')
await page.waitForURL('/dashboard')
// 终于可以开始测试了...
await expect(page.locator('h1')).toBeVisible()
})

问题:

  1. 每个测试重复登录 — 浪费时间,拖慢 CI
  2. 依赖真实认证服务 — Supabase Auth 不可用时测试失败
  3. 测试间状态污染 — 登录状态可能互相影响

解决方案

1. 创建 Custom Fixture

// e2e/fixtures.ts
import { test as base, expect, Page } from '@playwright/test'

// Mock 用户数据
export const mockUser = {
id: 'test-user-id',
email: 'test@example.com',
aud: 'authenticated',
role: 'authenticated',
}

export const mockAccessToken = 'mock-access-token-for-testing'

// 注入 auth token 到 localStorage
export async function setupMockAuth(page: Page) {
await page.addInitScript(
({ user, accessToken }) => {
const mockSession = {
access_token: accessToken,
refresh_token: 'mock-refresh-token',
expires_in: 3600,
expires_at: Math.floor(Date.now() / 1000) + 3600,
token_type: 'bearer',
user,
}
// Supabase auth token key 格式: sb-{project}-auth-token
localStorage.setItem('sb-placeholder-auth-token', JSON.stringify(mockSession))
},
{ user: mockUser, accessToken: mockAccessToken }
)
}

// 扩展 fixture
export const test = base.extend<{
authenticatedPage: Page
}>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await use(page)
},
})

export { expect }

2. 测试中使用 authenticatedPage

// e2e/dashboard.spec.ts
import { test, expect } from './fixtures'

test.describe('Dashboard', () => {
test('should display welcome message', async ({ authenticatedPage }) => {
// 直接访问受保护页面,无需登录
await authenticatedPage.goto('/dashboard')

await expect(authenticatedPage.locator('h1')).toContainText('Welcome')
})

test('should show agent list', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/dashboard')

const agents = authenticatedPage.locator('[data-testid="agent-card"]')
await expect(agents.first()).toBeVisible()
})
})

3. 对比:未认证 vs 已认证

// 未认证测试(会重定向到登录页)
test('redirects to login when not authenticated', async ({ page }) => {
await page.goto('/dashboard')
await page.waitForTimeout(500)

expect(page.url()).toContain('/login')
})

// 已认证测试(直接访问 dashboard)
test('allows access when authenticated', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/dashboard')
await authenticatedPage.waitForTimeout(500)

expect(authenticatedPage.url()).not.toContain('/login')
})

核心原理

addInitScript vs 页面加载后设置

// ❌ 错误方式:页面加载后设置(可能已重定向)
await page.goto('/dashboard')
await page.evaluate(() => {
localStorage.setItem('auth-token', '...')
})
// 此时 auth guard 已经检测到未登录并重定向了

// ✅ 正确方式:页面加载前注入
await page.addInitScript(() => {
localStorage.setItem('auth-token', '...')
})
await page.goto('/dashboard') // 页面加载时 auth guard 检测到 token

addInitScript 在以下时机执行:

  1. 页面 DOM 开始解析前
  2. React/Vue 等框架初始化前
  3. Auth guard 检查前

Fixture 生命周期

test('dashboard', async ({ authenticatedPage }) => {})

base.extend<{ authenticatedPage }>()

authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page) // 1. 设置 auth
await use(page) // 2. 执行测试
} // 3. 自动清理

完整配置

// playwright.config.ts
export default defineConfig({
testDir: './e2e',
fullyParallel: false, // 串行执行避免状态污染
workers: 1,
use: {
baseURL: 'http://localhost:5173',
trace: 'on-first-retry',
},
webServer: {
command: 'pnpm dev',
url: 'http://localhost:5173',
reuseExistingServer: !process.env.CI,
},
})

扩展:结合 API Mock

export const test = base.extend<{
authenticatedPage: Page
}>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await setupMockApi(page) // 同时 mock API
await use(page)
},
})

关键原则

  1. Fixture 优于 beforeEach — 自动复用,代码更简洁
  2. addInitScript 避免竞态 — 在 auth guard 检查前注入 token
  3. 隔离测试数据 — Mock 用户和 token 不要与生产混淆
  4. 串行执行避免污染workers: 1fullyParallel: false

对类似需求感兴趣?联系合作

绕过 Supabase Auth 实现 Playwright E2E 测试免登录

· 阅读需 4 分钟

在为客户构建 AI Agent SaaS 平台时遇到此问题,记录根因与解法。

TL;DR

E2E 测试不应该依赖真实的 OAuth 登录流程。通过在 useAuth hook 中检测 localStorage 的测试标记,直接注入 mock 认证状态,跳过 Supabase 初始化。同时将 Zustand store 的 loading 默认值改为 false,避免 AuthGuard 卡在无限 spinner。

问题现象

使用 Playwright 测试 React SPA 时,页面被 AuthGuard 组件保护,需要 Supabase 认证才能访问。测试启动后,页面一直显示 loading spinner,无法进入业务流程。

// AuthGuard 组件 - 测试时卡在这里
export function AuthGuard({ children }: AuthGuardProps) {
const { isAuthenticated, loading } = useAuth()

if (loading) {
return <Spinner /> // 永远显示 spinner
}

if (!isAuthenticated) {
return <Navigate to="/login" />
}

return <>{children}</>
}

测试代码尝试模拟登录,但 Supabase Auth SDK 内部状态无法通过简单的 API mock 控制。

根因

1. Supabase Auth 是异步初始化的

useAuth hook 在 useEffect 中调用 supabase.auth.getSession(),这是异步操作。测试环境下网络请求可能失败或超时,导致状态永远停留在 loading: true

2. Zustand Store 的默认值问题

// authStore.ts - 问题代码
export const useAuthStore = create<AuthState>()(
persist(
(set) => ({
user: null,
token: null,
loading: true, // 👈 默认值是 true
// ...
}),
{ name: 'auth-storage' }
)
)

测试启动时,loading: true + 异步初始化失败 = 永远 loading。

3. OAuth 流程无法自动化

即使能 mock API,OAuth 的重定向流程需要真实浏览器交互,E2E 测试无法可靠模拟。

解决方案

步骤 1:在 useAuth hook 中添加测试模式检测

// hooks/useAuth.ts
export function useAuth() {
const { user, token, loading, setUser, setToken, setLoading } = useAuthStore()

useEffect(() => {
const initAuth = async () => {
// 👇 优先检测测试模式
const testAuthUser = localStorage.getItem('test-auth-user')
const testAuthToken = localStorage.getItem('test-auth-token')

if (testAuthUser && testAuthToken) {
try {
const userData = JSON.parse(testAuthUser) as User
setUser(userData)
setToken(testAuthToken)
setLoading(false)
console.log('[useAuth] Using test mode auth')
return // 👈 直接返回,跳过 Supabase 初始化
} catch (e) {
console.error('Failed to parse test auth user:', e)
}
}

// 👇 正常模式:走 Supabase Auth
try {
const { data: { session } } = await supabase.auth.getSession()
if (session) {
setUser(session.user as User)
setToken(session.access_token)
}
} catch (error) {
console.error('Auth init failed:', error)
} finally {
setLoading(false)
}
}

initAuth()

// 👇 测试模式下跳过 auth state listener
if (localStorage.getItem('test-auth-user')) {
return
}

const { data: { subscription } } = supabase.auth.onAuthStateChange(
async (event, session) => {
// ... 正常的 auth state 处理
}
)

return () => subscription.unsubscribe()
}, [])
}

步骤 2:修改 Zustand Store 默认值

// stores/authStore.ts
export const useAuthStore = create<AuthState>()(
persist(
(set) => ({
user: null,
token: null,
loading: false, // 👈 改为 false,让 useAuth hook 控制状态
setUser: (user) => set({ user }),
setToken: (token) => set({ token }),
setLoading: (loading) => set({ loading }),
logout: () => set({ user: null, token: null, loading: false }),
}),
{
name: 'auth-storage',
partialize: (state) => ({ user: state.user, token: state.token }),
}
)
)

步骤 3:在 Playwright Fixture 中注入测试认证

// e2e/fixtures.ts
import { test as base } from '@playwright/test'

export const mockUser = {
id: 'test-user-id',
email: 'test@example.com',
created_at: '2024-01-01T00:00:00Z',
}

export const test = base.extend({
authenticatedPage: async ({ page }, use) => {
// 先访问页面以设置 localStorage 的 origin
await page.goto('/login')

// 👇 注入测试认证状态到 localStorage
await page.evaluate(
({ user, token }) => {
localStorage.setItem('test-auth-user', JSON.stringify(user))
localStorage.setItem('test-auth-token', token)
},
{ user: mockUser, token: 'mock-access-token' }
)

// 导航到受保护页面,useAuth 会检测到测试模式
await page.goto('/dashboard')

await use(page)
},
})

步骤 4:在测试中使用

// e2e/dashboard.spec.ts
import { test, expect } from './fixtures'

test('dashboard shows user agents', async ({ authenticatedPage }) => {
// authenticatedPage 已经通过认证,无需登录
await expect(authenticatedPage.getByText('Test Agent')).toBeVisible()
})

完整代码结构

agent-frontend/
├── e2e/
│ ├── fixtures.ts # Playwright fixture + mock 数据
│ ├── dashboard.spec.ts # 测试用例
│ └── ...
├── src/
│ ├── hooks/
│ │ └── useAuth.ts # 测试模式检测
│ └── stores/
│ └── authStore.ts # loading: false 默认值
└── playwright.config.ts

关键要点

  1. 测试模式 key 使用特殊前缀test-auth-* 不会在生产环境中出现
  2. 检测优先于初始化:先检查 localStorage,再走 Supabase Auth
  3. 跳过 auth listener:测试模式下不需要监听 auth state 变化
  4. loading 默认值改为 false:让 hook 显式控制 loading 状态

对类似需求感兴趣?联系合作

在 Zustand Store 中实现数据缓存

· 阅读需 3 分钟

在构建 AI Agent 平台时遇到此问题,记录根因与解法。

TL;DR

在 Zustand store 中添加 lastFetchTime 字段和 TTL 常量,请求前检查缓存是否过期。10 行代码实现简单有效的数据缓存,避免多页面重复请求同一数据。

问题现象

MCP 工具列表被多个页面使用(Agent 设置页、工具市场页、聊天页),每次进入页面都触发 API 请求:

GET /api/mcp-tools  → 200  (AgentSettingsPage)
GET /api/mcp-tools → 200 (McpToolsPage)
GET /api/mcp-tools → 200 (ChatPage tool selector)

工具列表变化频率很低(管理员手动配置),频繁请求浪费带宽且影响页面加载速度。

根因

直接在组件中调用 API,没有缓存层:

// ❌ 无缓存:每次挂载都请求
function McpToolsPage() {
const [tools, setTools] = useState([])

useEffect(() => {
mcpToolsApi.list().then(setTools)
}, [])

return <ToolList tools={tools} />
}

问题:

  1. 每个页面独立请求 — 没有全局状态共享
  2. 短时间内重复请求 — 用户在页面间跳转时触发多次
  3. 无法控制刷新频率 — 即使数据未变化也重新获取

解决方案

Zustand Store + TTL 缓存

// src/stores/mcpToolsStore.ts
import { create } from 'zustand'
import { mcpToolsApi, type McpTool } from '@/services/api'

const CACHE_TTL = 10 * 60 * 1000 // 10 分钟

interface McpToolsState {
tools: McpTool[]
lastFetchTime: number | null
loading: boolean
error: string | null

fetchTools: (force?: boolean) => Promise<void>
clearError: () => void
}

export const useMcpToolsStore = create<McpToolsState>((set, get) => ({
tools: [],
lastFetchTime: null,
loading: false,
error: null,

fetchTools: async (force = false) => {
const { tools, lastFetchTime } = get()

// 有缓存且未过期且非强制 → 跳过
if (tools.length && lastFetchTime && !force) {
if (Date.now() - lastFetchTime < CACHE_TTL) {
return // 缓存命中,直接返回
}
}

set({ loading: true, error: null })
try {
const data = await mcpToolsApi.list()
set({ tools: data, lastFetchTime: Date.now(), loading: false })
} catch (err) {
const message = err instanceof Error ? err.message : 'Failed to fetch tools'
set({ error: message, loading: false })
}
},

clearError: () => set({ error: null }),
}))

组件中使用

// ✅ 使用 store 缓存
function McpToolsPage() {
const { tools, loading, fetchTools } = useMcpToolsStore()

useEffect(() => {
fetchTools() // 自动检查缓存
}, [fetchTools])

if (loading) return <Spinner />
return <ToolList tools={tools} />
}

// 强制刷新
function RefreshButton() {
const { fetchTools } = useMcpToolsStore()
return <button onClick={() => fetchTools(true)}>刷新</button>
}

核心逻辑解析

// 缓存检查逻辑
if (tools.length && lastFetchTime && !force) {
if (Date.now() - lastFetchTime < CACHE_TTL) {
return // 缓存有效,跳过请求
}
}
条件说明
tools.length已有数据(空数组不算有效缓存)
lastFetchTime记录了上次请求时间
!force非强制刷新
Date.now() - lastFetchTime < CACHE_TTL未超过过期时间

适用场景

场景是否适合原因
工具列表、配置字典✅ 适合变化频率低,多页面共享
用户权限、角色✅ 适合会话内基本不变
实时数据(消息、通知)❌ 不适合需要最新状态
分页列表❌ 不适合数据量大,缓存策略复杂

扩展:更精细的缓存控制

interface CacheOptions {
ttl: number // 过期时间
staleWhileRevalidate: boolean // 过期后先返回旧数据再更新
}

// 过期后后台刷新,先返回缓存数据
if (tools.length && lastFetchTime) {
const age = Date.now() - lastFetchTime
if (age < CACHE_TTL) {
return // 缓存新鲜
}
if (options.staleWhileRevalidate && age < CACHE_TTL * 2) {
// 缓存过期但可接受,后台刷新
mcpToolsApi.list().then(data => set({ tools: data, lastFetchTime: Date.now() }))
return
}
}

关键原则

  1. 缓存时间要合理 — 根据数据变化频率设置 TTL
  2. 提供强制刷新入口 — 用户可手动刷新最新数据
  3. 首次加载要有 loading 状态 — 空数据时不应跳过请求

对类似需求感兴趣?联系合作

修复 httpx async with client.post() 的隐藏坑

· 阅读需 2 分钟

在构建多服务协作的 SaaS 系统时遇到此问题,记录根因与解法。

TL;DR

httpx.AsyncClient 不要用 async with client.post() 模式,应该先创建 client 再调用方法:response = await client.post()

问题现象

import httpx

async def call_api():
async with httpx.AsyncClient() as client:
async with client.post(url, json=data) as response: # 问题代码
return response.json()

这段代码有时正常,有时报错:

httpx.RemoteProtocolError: cannot write to closing transport
RuntimeError: Session is closed

根因

async with client.post() 的陷阱

client.post() 返回的是 Response 对象,不是上下文管理器。用 async with 包装会导致:

  1. 连接过早关闭async with 块结束时立即关闭连接,但响应可能还在读取
  2. 资源竞争:多个并发请求时,连接池状态混乱

正确理解 httpx 上下文管理器

# ✅ 正确:client 是上下文管理器
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data)
return response.json()

# ❌ 错误:把 response 当上下文管理器
async with client.post(url) as response:
...

解决方案

方案 1:单次请求(推荐简单场景)

async def call_api(url: str, data: dict) -> dict:
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data)
response.raise_for_status()
return response.json()

方案 2:复用 client(推荐高频请求)

# 全局或依赖注入
_client = httpx.AsyncClient(timeout=30.0)

async def call_api(url: str, data: dict) -> dict:
response = await _client.post(url, json=data)
response.raise_for_status()
return response.json()

# 应用关闭时
async def shutdown():
await _client.aclose()

方案 3:FastAPI 依赖注入

from fastapi import Depends
from httpx import AsyncClient

async def get_http_client() -> AsyncClient:
async with AsyncClient(timeout=30.0) as client:
yield client

@router.post("/proxy")
async def proxy(
data: dict,
client: AsyncClient = Depends(get_http_client)
):
response = await client.post("https://external.api/endpoint", json=data)
return response.json()

FAQ

Q: httpx async with 怎么用才对?

A: async with 只用于管理 AsyncClient 生命周期,不是包装单个请求。正确模式:async with AsyncClient() as client: response = await client.post(...)

Q: 为什么有时 async with client.post() 也能跑?

A: 单线程、低并发时可能碰巧正常,但高并发或网络延迟时会暴露问题。这是隐藏 bug,不要侥幸。

Q: httpx 超时怎么配置?

A: AsyncClient(timeout=30.0)AsyncClient(timeout=httpx.Timeout(connect=5.0, read=30.0))

集成 Supabase Auth 到 FastAPI 的三个坑

· 阅读需 4 分钟

在为客户构建 SaaS 认证系统时遇到此问题,记录根因与解法。

TL;DR

Supabase Auth + FastAPI 集成有三个常见坑:JWKS 路径不是标准路径、ES256 签名需转换为 DER 格式、用户首次登录时本地数据库无记录。本文提供完整解决方案。

问题现象

坑 1:JWKS 路径 404

GET https://xxx.supabase.co/.well-known/jwks.json
# 404 Not Found

所有 JWT 验证请求返回 401 Invalid Token。

坑 2:ES256 签名验证失败

from jose import jwt
payload = jwt.decode(token, key, algorithms=["ES256"])
# JWTError: Signature verification failed

明明公钥是对的,但签名验证总是失败。

坑 3:用户首次登录无本地记录

# 创建 Agent 时
agent = Agent(user_id=current_user["user_id"], ...)
db.add(agent)
# ForeignKeyViolation: user_id 不存在

Supabase Auth 用户通过了 JWT 验证,但本地 agent_users 表没有该用户记录。

根因

坑 1:Supabase 非标准 JWKS 路径

标准 OAuth/OIDC 服务器 JWKS 在 /.well-known/jwks.json,但 Supabase 把认证服务放在 /auth/v1/ 子路径下:

标准路径Supabase 路径
/.well-known/jwks.json/auth/v1/.well-known/jwks.json

坑 2:ES256 原始签名 vs DER 格式

Supabase JWT 使用 ES256(P-256 曲线)签名。JWT 中的签名是 raw 格式r || s 拼接,64 字节),但 Python cryptography 库的 verify() 方法需要 DER-encoded ASN.1 格式

Raw:     r (32 bytes) || s (32 bytes) = 64 bytes
DER: 0x30 <len> 0x02 <r_len> <r> 0x02 <s_len> <s>

python-josejwt.decode() 在处理 ES256 时有兼容性问题,需要手动验证签名。

坑 3:认证与数据分离

Supabase Auth 是独立服务,用户注册/登录后只存在于 Supabase 的 auth.users 表。本地数据库的 agent_users 表需要手动同步。

解决方案

1. 正确的 JWKS URL

# config.py
class Settings(BaseSettings):
supabase_url: str = "https://xxx.supabase.co"

@property
def jwks_url(self) -> str:
# 关键:/auth/v1/ 前缀
return f"{self.supabase_url}/auth/v1/.well-known/jwks.json"

2. ES256 签名验证(完整代码)

import json
import base64
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature

def _base64url_decode(data: str) -> bytes:
"""Base64url 解码,自动补 padding"""
rem = len(data) % 4
if rem > 0:
data += "=" * (4 - rem)
return base64.urlsafe_b64decode(data)

def _raw_to_der_signature(raw_sig: bytes) -> bytes:
"""将 raw ECDSA 签名 (r||s) 转为 DER 格式"""
# P-256: r 和 s 各 32 字节
r = int.from_bytes(raw_sig[:32], "big")
s = int.from_bytes(raw_sig[32:], "big")
return encode_dss_signature(r, s)

def verify_es256_signature(token: str, public_key_jwk: dict) -> dict:
"""验证 ES256 JWT 签名,返回 payload"""
parts = token.split(".")
if len(parts) != 3:
raise ValueError("Invalid JWT format")

header_b64, payload_b64, signature_b64 = parts

# 1. 构建 EC 公钥
x = _base64url_decode(public_key_jwk["x"])
y = _base64url_decode(public_key_jwk["y"])
x_int = int.from_bytes(x, "big")
y_int = int.from_bytes(y, "big")

public_key = ec.EllipticCurvePublicNumbers(
x_int, y_int, ec.SECP256R1()
).public_key(default_backend())

# 2. 验证签名
message = f"{header_b64}.{payload_b64}".encode()
raw_signature = _base64url_decode(signature_b64)
der_signature = _raw_to_der_signature(raw_signature)

public_key.verify(
der_signature,
message,
ec.ECDSA(hashes.SHA256())
)

# 3. 返回 payload
return json.loads(_base64url_decode(payload_b64))

3. 用户同步服务

# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import AgentUser

async def ensure_user_exists(
db: AsyncSession,
user_id: str,
email: str,
plan: str = "free"
) -> AgentUser:
"""确保用户存在于本地数据库(从 Supabase Auth 同步)"""
# 检查是否存在
result = await db.execute(
select(AgentUser).where(AgentUser.user_id == user_id)
)
user = result.scalar_one_or_none()

if user:
return user

# 创建新用户
user = AgentUser(
user_id=user_id,
email=email,
plan=plan,
role="user"
)
db.add(user)
await db.commit()
await db.refresh(user)
return user

4. 在创建资源前调用

# app/routers/agents.py
@router.post("/")
async def create_agent(
input: CreateAgentInput,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
# 关键:确保用户存在
user = await ensure_user_exists(
db,
user_id=current_user["user_id"],
email=current_user["email"],
plan=current_user["plan"]
)

# 现在可以安全创建 Agent
agent = Agent(
user_id=user.user_id,
name=input.name,
llm_config=input.llm_config.model_dump()
)
...

FAQ

Q: Supabase JWT 验证返回 404 怎么办?

A: Supabase 的 JWKS 路径是 /auth/v1/.well-known/jwks.json,不是标准的 /.well-known/jwks.json。检查你的 JWKS URL 配置。

Q: python-jose 验证 ES256 签名失败怎么解决?

A: python-jose 对 ES256 支持不完善。使用 cryptography 库手动验证,需要将 JWT 的 raw 签名(r||s 64字节)转换为 DER 格式。

Q: FastAPI 如何同步 Supabase Auth 用户到本地数据库?

A: 在需要用户记录的 API(如创建资源)入口处调用 ensure_user_exists(),从 JWT 提取用户信息并同步到本地表。

Q: Supabase JWT 中的 user_id 在哪个字段?

A: sub 字段包含用户 UUID,email 字段包含邮箱,app_metadata.plan 包含订阅计划(自定义字段)。