介绍
# Senior Backend Engineer
后端开发模式、API 设计、数据库优化和安全实践。
## 目录
- [快速开始](#quick-start) - [工具概览](#tools-overview) - [API 脚手架](#1-api-scaffolder) - [数据库迁移工具](#2-database-migration-tool) - [API 负载测试工具](#3-api-load-tester) - [后端开发工作流](#backend-development-workflows) - [API 设计工作流](#api-design-workflow) - [数据库优化工作流](#database-optimization-workflow) - [安全加固工作流](#security-hardening-workflow) - [参考文档](#reference-documentation) - [常用模式快速参考](#common-patterns-quick-reference)
---
## 快速开始
```bash # Generate API routes from OpenAPI spec python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/
# Analyze database schema and generate migrations python scripts/database_migration_tool.py --connection postgres://localhost/mydb --analyze
# Load test an API endpoint python scripts/api_load_tester.py https://api.example.com/users --concurrency 50 --duration 30 ```
---
## 工具概览
### 1. API 脚手架
根据模式定义生成 API 路由处理程序、中间件和 OpenAPI 规范。
**输入：** OpenAPI 规范 (YAML/JSON) 或数据库模式 **输出：** 路由处理程序、验证中间件、TypeScript 类型
**用法：** ```bash # Generate Express routes from OpenAPI spec python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/
# Output: # Generated 12 route handlers in src/routes/ # - GET /users (listUsers) # - POST /users (createUser) # - GET /users/{id} (getUser) # - PUT /users/{id} (updateUser) # - DELETE /users/{id} (deleteUser) # ... # Created validation middleware: src/middleware/validators.ts # Created TypeScript types: src/types/api.ts
# Generate from database schema python scripts/api_scaffolder.py --from-db postgres://localhost/mydb --output src/routes/
# Generate OpenAPI spec from existing routes python scripts/api_scaffolder.py src/routes/ --generate-spec --output openapi.yaml ```
**支持的框架：** - Express.js (`--framework express`) - Fastify (`--framework fastify`) - Koa (`--framework koa`)
---
### 2. 数据库迁移工具
分析数据库模式，检测更改，并生成带有回滚支持的迁移文件。
**输入：** 数据库连接字符串或模式文件 **输出：** 迁移文件、模式差异报告、优化建议
**用法：** ```bash # Analyze current schema and suggest optimizations python scripts/database_migration_tool.py --connection postgres://localhost/mydb --analyze
# Output: # === Database Analysis Report === # Tables: 24 # Total rows: 1,247,832 # # MISSING INDEXES (5 found): # orders.user_id - 847ms avg query time, ADD INDEX recommended # products.category_id - 234ms avg query time, ADD INDEX recommended # # N+1 QUERY RISKS (3 found): # users -> orders relationship (no eager loading) # # SUGGESTED MIGRATIONS: # 1. Add index on orders(user_id) # 2. Add index on products(category_id) # 3. Add composite index on order_items(order_id, product_id)
# Generate migration from schema diff python scripts/database_migration_tool.py --connection postgres://localhost/mydb \ --compare schema/v2.sql --output migrations/
# Output: # Generated migration: migrations/20240115_add_user_indexes.sql # Generated rollback: migrations/20240115_add_user_indexes_rollback.sql
# Dry-run a migration python scripts/database_migration_tool.py --connection postgres://localhost/mydb \ --migrate migrations/20240115_add_user_indexes.sql --dry-run ```
---
### 3. API 负载测试工具
执行可配置并发数的 HTTP 负载测试，测量延迟百分位和吞吐量。
**输入：** API 端点 URL 和测试配置 **输出：** 包含延迟分布、错误率和吞吐量指标的性能报告
**用法：** ```bash # Basic load test python scripts/api_load_tester.py https://api.example.com/users --concurrency 50 --duration 30
# Output: # === Load Test Results === # Target: https://api.example.com/users # Duration: 30s | Concurrency: 50 # # THROUGHPUT: # Total requests: 15,247 # Requests/sec: 508.2 # Successful: 15,102 (99.0%) # Failed: 145 (1.0%) # # LATENCY (ms): # Min: 12 # Avg: 89 # P50: 67 # P95: 198 # P99: 423 # Max: 1,247 # # ERRORS: # Connection timeout: 89 # HTTP 503: 56 # # RECOMMENDATION: P99 latency (423ms) exceeds 200ms target. # Consider: connection pooling, query optimization, or horizontal scaling.
# Test with custom headers and body python scripts/api_load_tester.py https://api.example.com/orders \ --method POST \ --header "Authorization: Bearer token123" \ --body '{"product_id": 1, "quantity": 2}' \ --concurrency 100 \ --duration 60
# Compare two endpoints python scripts/api_load_tester.py https://api.example.com/v1/users https://api.example.com/v2/users \ --compare --concurrency 50 --duration 30 ```
---
## 后端开发工作流
### API 设计工作流
在设计新 API 或重构现有端点时使用。
**步骤 1：定义资源和操作** ```yaml # openapi.yaml openapi: 3.0.3 info: title: User Service API version: 1.0.0 paths: /users: get: summary: List users parameters: - name: limit in: query schema: type: integer default: 20 post: summary: Create user requestBody: required: true content: application/json: schema: $ref: '#/components/schemas/CreateUser' ```
**步骤 2：生成路由脚手架** ```bash python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/ ```
**步骤 3：实现业务逻辑** ```typescript // src/routes/users.ts (generated, then customized) export const createUser = async (req: Request, res: Response) => { const { email, name } = req.body;
// Add business logic const user = await userService.create({ email, name });
res.status(201).json(user); }; ```
**步骤 4：添加验证中间件** ```bash # Validation is auto-generated from OpenAPI schema # src/middleware/validators.ts includes: # - Request body validation # - Query parameter validation # - Path parameter validation ```
**步骤 5：生成更新的 OpenAPI 规范** ```bash python scripts/api_scaffolder.py src/routes/ --generate-spec --output openapi.yaml ```
---
### 数据库优化工作流
当查询缓慢或数据库性能需要改进时使用。
**步骤 1：分析当前性能** ```bash python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze ```
**步骤 2：识别慢查询** ```sql -- Check query execution plans EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 123 ORDER BY created_at DESC LIMIT 10;
-- Look for: Seq Scan (bad), Index Scan (good) ```
**步骤 3：生成索引迁移** ```bash python scripts/database_migration_tool.py --connection $DATABASE_URL \ --suggest-indexes --output migrations/ ```
**步骤 4：测试迁移（试运行）** ```bash python scripts/database_migration_tool.py --connection $DATABASE_URL \ --migrate migrations/add_indexes.sql --dry-run ```
**步骤 5：应用并验证** ```bash # Apply migration python scripts/database_migration_tool.py --connection $DATABASE_URL \ --migrate migrations/add_indexes.sql
# Verify improvement python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze ```
---
### 安全加固工作流
在准备 API 上线或进行安全审查后使用。
**步骤 1：审查身份验证设置** ```typescript // Verify JWT configuration const jwtConfig = { secret: process.env.JWT_SECRET, // Must be from env, never hardcoded expiresIn: '1h', // Short-lived tokens algorithm: 'RS256' // Prefer asymmetric }; ```
**步骤 2：添加速率限制** ```typescript import rateLimit from 'express-rate-limit';
const apiLimiter = rateLimit({ windowMs: 15 * 60 * 1000, // 15 minutes max: 100, // 100 requests per window standardHeaders: true, legacyHeaders: false, });
app.use('/api/', apiLimiter); ```
**步骤 3：验证所有输入** ```typescript import { z } from 'zod';
const CreateUserSchema = z.object({ email: z.string().email().max(255), name: z.string().min(1).max(100), age: z.number().int().positive().optional() });
// Use in route handler const data = CreateUserSchema.parse(req.body); ```
**步骤 4：使用攻击模式进行负载测试** ```bash # Test rate limiting python scripts/api_load_tester.py https://api.example.com/login \ --concurrency 200 --duration 10 --expect-rate-limit
# Test input validation python scripts/api_load_tester.py https://api.example.com/users \ --method POST \ --body '{"email": "not-an-email"}' \ --expect-status 400 ```
**步骤 5：审查安全标头** ```typescript import helmet from 'helmet';
app.use(helmet({ contentSecurityPolicy: true, crossOriginEmbedderPolicy: true, crossOriginOpenerPolicy: true, crossOriginResourcePolicy: true, hsts: { maxAge: 31536000, includeSubDomains: true }, })); ```
---
## 参考文档
| 文件 | 内容 | 使用场景 | |------|----------|----------| | `references/api_design_patterns.md` | REST vs GraphQL、版本控制、错误处理、分页 | 设计新 API | | `references/database_optimization_guide.md` | 索引策略、查询优化、N+1 解决方案 | 修复慢查询 | | `references/backend_security_practices.md` | OWASP Top 10、身份验证模式、输入验证 | 安全加固 |
---
## 常用模式快速参考
### REST API 响应格式 ```json { "data": { "id": 1, "name": "John" }, "meta": { "requestId": "abc-123" } } ```
### 错误响应格式 ```json { "error": { "code": "VALIDATION_ERROR", "message": "Invalid email format", "details": [{ "field": "email", "message": "must be valid email" }] }, "meta": { "requestId": "abc-123" } } ```
### HTTP 状态代码 | 代码 | 使用场景 | |------|----------| | 200 | 成功 (GET, PUT, PATCH) | | 201 | 已创建 (POST) | | 204 | 无内容 (DELETE) | | 400 | 验证错误 | | 401 | 需要身份验证 | | 403 | 权限被拒绝 | | 404 | 资源未找到 | | 429 | 超过速率限制 | | 500 | 内部服务器错误 |
### 数据库索引策略 ```sql -- Single column (equality lookups) CREATE INDEX idx_users_email ON users(email);
-- Composite (multi-column queries) CREATE INDEX idx_orders_user_status ON orders(user_id, status);
-- Partial (filtered queries) CREATE INDEX idx_orders_active ON orders(created_at) WHERE status = 'active';
-- Covering (avoid table lookup) CREATE INDEX idx_users_email_name ON users(email) INCLUDE (name); ```
---
## 常用命令
```bash # API Development python scripts/api_scaffolder.py openapi.yaml --framework express python scripts/api_scaffolder.py src/routes/ --generate-spec
# Database Operations python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze python scripts/database_migration_tool.py --connection $DATABASE_URL --migrate file.sql
# Performance Testing python scripts/api_load_tester.py https://api.example.com/endpoint --concurrency 50 python scripts/api_load_tester.py https://api.example.com/endpoint --compare baseline.json ```