Flask 是 Python 生态中最流行的轻量级 Web 框架之一。它简洁灵活,不预设项目结构,非常适合快速构建个人 API 服务。本文将带你从零搭建一个完整的 REST API,涵盖路由设计、JWT 认证与中间件开发。

Flask API开发

一、项目初始化

Flask 项目的起步非常简单,一个文件即可跑起来:

from flask import Flask, jsonify app = Flask(__name__) @app.route('/api/health') def health(): return jsonify({'status': 'ok', 'version': '1.0.0'}) if __name__ == '__main__': app.run(host='0.0.0.0', port=6789, debug=True)

推荐使用环境变量管理敏感配置,避免硬编码密码和密钥:

import os HOST = os.environ.get('APP_HOST', '127.0.0.1') PORT = int(os.environ.get('APP_PORT', 6789)) SECRET = os.environ.get('APP_SECRET', 'change-me')
安全提醒:生产环境必须修改默认密码!在服务器环境变量或 .env 文件中设置强密码,并确保 .env 文件列入 .gitignore。

二、RESTful 路由设计

良好的 API 设计遵循 RESTful 规范,URL 以资源名词为中心,HTTP 方法表示操作类型:

from flask import request @app.route('/api/signals', methods=['GET']) def get_signals(): code = request.args.get('code') strategy = request.args.get('strategy', 'default') # 查询数据库并返回 JSON results = query_signals(code, strategy) return jsonify({'code': 0, 'data': results}) @app.route('/api/signals', methods=['POST']) def create_signal(): data = request.get_json() # 数据验证 if not data.get('code'): return jsonify({'code': 10002, 'message': '缺少必填参数 code'}), 400 # 写入数据库 insert_signal(data) return jsonify({'code': 0, 'message': 'ok'}), 201

三、JWT 认证中间件

JWT(JSON Web Token)是实现无状态 API 认证的最佳方案。以下是一个手写的 HMAC-SHA256 JWT 实现:

JWT认证
import hmac, hashlib, base64, json, time def create_token(user_id, secret, expire=7200): header = base64.urlsafe_b64encode(json.dumps({'alg':'HS256','typ':'JWT'}).encode()).rstrip(b'=') payload = json.dumps({'user_id':user_id,'iat':int(time.time()), 'exp':int(time.time())+expire}).encode() payload_b64 = base64.urlsafe_b64encode(payload).rstrip(b'=') signing = f'{header.decode()}.{payload_b64.decode()}' sig = base64.urlsafe_b64encode( hmac.new(secret.encode(),signing.encode(),hashlib.sha256).digest() ).rstrip(b'=') return f'{signing}.{sig.decode()}' def verify_token(token, secret): try: parts = token.split('.') if len(parts) != 3: return None signing = f'{parts[0]}.{parts[1]}' expected = base64.urlsafe_b64encode( hmac.new(secret.encode(),signing.encode(),hashlib.sha256).digest() ).rstrip(b'=') if not hmac.compare_digest(expected, parts[2].encode()): return None padded = parts[1] + '='*(4-len(parts[1])%4) payload = json.loads(base64.urlsafe_b64decode(padded)) if payload['exp'] < time.time(): return None return payload except: return None

使用 before_request 钩子实现自动鉴权:

@app.before_request def authenticate(): if request.endpoint in ('health',): return # 白名单跳过 auth = request.headers.get('Authorization', '') if not auth.startswith('Bearer '): return jsonify({'code':11001,'message':'缺少 Token'}), 401 token = auth[7:] payload = verify_token(token, SHARED_SECRET) if not payload: return jsonify({'code':11002,'message':'Token 无效或已过期'}), 401 request.user_id = payload['user_id']

四、CORS 跨域配置

如果前端和 API 在不同域名,需要配置 CORS:

from flask_cors import CORS CORS(app, resources={r'/*': { 'origins': '*', 'supports_credentials': False, 'methods': ['GET','POST','PUT','DELETE','OPTIONS'], 'allow_headers': ['Content-Type','Authorization'] }})

五、请求日志与错误处理

使用 after_request 钩子记录每个 API 请求的耗时和状态,便于问题排查:

@app.before_request def start_timer(): request.start_time = time.time() @app.after_request def log_request(response): duration = int((time.time() - request.start_time) * 1000) print(f'{request.method} {request.path} {response.status_code} {duration}ms') return response

六、Docker 部署

用 Docker 部署 Flask 应用可以保证环境一致性:

FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 6789 CMD ["python", "signal_server.py"]

配合 Nginx 反向代理和 systemd 服务管理,即可实现生产级别的服务部署。

总结

Flask 的简洁设计使其成为个人项目 API 的首选框架。本文从项目初始化到 JWT 认证、CORS 配置到 Docker 部署,涵盖了一个完整 API 服务的开发流程。对于个人数据分析工具的后端服务,Flask + SQLite 的组合方案兼具开发效率和运行性能,推荐给所有个人开发者尝试。