Flask HTTP服务器开发指南

Flask Web服务开发完全指南
1. 环境准备
首先需要安装必要的依赖:
# 安装Flask框架
pip install Flask
# 安装生产级服务器
pip install gunicorn
2. 基础HTTP服务开发
2.1 创建基础应用 (app.py)
from flask import Flask, request, jsonify
import wave
import io
app = Flask(__name__)
@app.route('/process_wav', methods=['POST'])
def process_wav():
"""处理WAV文件上传的端点"""
# 验证请求中的文件
if 'file' not in request.files:
return jsonify({'error': 'No file part'}), 400
file = request.files['file']
# 验证文件名
if file.filename == '':
return jsonify({'error': 'No selected file'}), 400
if file:
try:
# WAV文件处理逻辑
with wave.open(file, 'rb') as wav_file:
framerate = wav_file.getframerate()
nframes = wav_file.getnframes()
duration = nframes / float(framerate)
return jsonify({
'status': 'success',
'duration': duration,
'framerate': framerate
})
except Exception as e:
return jsonify({
'error': 'File processing failed',
'message': str(e)
}), 500
return jsonify({'error': 'Invalid request'}), 400
@app.route('/health', methods=['GET'])
def health_check():
"""健康检查端点"""
return jsonify({'status': 'healthy'})
if __name__ == '__main__':
app.run(
debug=True, # 开发模式
host="0.0.0.0", # 允许外部访问
port=3000 # 指定端口
)
2.2 API端点说明
-
POST /process_wav
- 功能:处理WAV音频文件
- 请求:multipart/form-data格式,包含文件
- 响应:JSON格式的处理结果
-
GET /health
- 功能:服务健康检查
- 响应:服务状态信息
3. 生产环境部署
3.1 Gunicorn配置 (gunicorn_config.py)
import os
# 工作进程数
workers = int(os.environ.get('GUNICORN_PROCESSES', '8'))
# 每个工作进程的线程数
threads = int(os.environ.get('GUNICORN_THREADS', '8'))
# 服务绑定地址
bind = os.environ.get('GUNICORN_BIND', '0.0.0.0:3000')
# 允许的代理IP
forwarded_allow_ips = '*'
# 超时设置(秒)
timeout = 120
# 访问日志格式
accesslog = 'access.log'
errorlog = 'error.log'
# 工作模式
worker_class = 'gthread'
# 最大请求数
max_requests = 1000
max_requests_jitter = 50
3.2 启动服务
# 开发环境启动
python app.py
# 生产环境启动
gunicorn --config gunicorn_config.py app:app
4. 最佳实践建议
-
错误处理
- 实现全局错误处理器
- 使用适当的HTTP状态码
- 提供清晰的错误信息
-
安全考虑
- 实现请求验证
- 限制上传文件大小
- 添加适当的CORS设置
-
性能优化
- 使用异步处理大文件
- 实现请求限流
- 添加缓存机制
-
监控与日志
- 集成日志系统
- 添加性能监控
- 实现请求追踪
5. 常见问题解决
-
文件上传失败
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 限制16MB
-
跨域访问问题
from flask_cors import CORS CORS(app)
-
请求超时处理
@app.route('/long_process') def long_process(): try: with timeout(seconds=30): # 长时间处理 pass except TimeoutError: return jsonify({'error': 'Operation timed out'}), 408