Flask 是 Python 生态中最流行的轻量级 Web 框架之一。它简洁灵活,不预设项目结构,非常适合快速构建个人 API 服务。本文将带你从零搭建一个完整的 REST API,涵盖路由设计、JWT 认证与中间件开发。
一、项目初始化
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 实现:
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 的组合方案兼具开发效率和运行性能,推荐给所有个人开发者尝试。