跳到主要内容

Flask

Flask 是一个轻量级、灵活的 Python Web 框架,以简洁和可扩展性为核心设计理念。它不强制依赖特定库或架构,仅提供核心功能(如路由、模板渲染),开发者可自由选配数据库、表单验证等组件。这种"微框架"特性使其学习成本极低,同时能通过扩展轻松构建复杂应用,特别适合快速开发小型项目或作为微服务基础。

本指南介绍如何在 CloudBase HTTP 云函数上部署 Flask 应用程序。

前置条件

在开始之前,请确保您已经:

  • 安装了 Python 3.10 或更高版本
  • 拥有腾讯云账号并开通了 CloudBase 服务
  • 了解基本的 Python 和 Flask 开发知识

第一步:创建 Flask 应用

💡 提示:如果您已经有一个 Flask 应用,可以跳过此步骤。

创建项目目录

mkdir flask-cloudbase
cd flask-cloudbase

设置虚拟环境

# 创建虚拟环境
python -m venv env

# 激活虚拟环境
# Linux/macOS
source env/bin/activate

# Windows
# env\Scripts\activate

安装 Flask

pip install flask

第二步:编写应用代码

创建 app.py 文件作为应用的入口文件:

⚠️ 重要提示:CloudBase HTTP 云函数要求应用监听 9000 端口。

import os
from flask import Flask, jsonify, request

app = Flask(__name__)

# 配置应用
app.config['DEBUG'] = os.environ.get('DEBUG', 'False').lower() == 'true'
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'your-secret-key-here')

# 模拟数据
users = [
{'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'},
{'id': 2, 'name': '李四', 'email': 'lisi@example.com'},
{'id': 3, 'name': '王五', 'email': 'wangwu@example.com'}
]

@app.route('/')
def hello():
"""根路径处理函数"""
return jsonify({
'message': 'Hello from Flask on CloudBase!',
'framework': 'Flask',
'version': '2.3.0'
})

@app.route('/health')
def health_check():
"""健康检查接口"""
return jsonify({
'status': 'healthy',
'framework': 'Flask',
'python_version': os.sys.version
})

@app.route('/api/users', methods=['GET'])
def get_users():
"""获取用户列表(支持分页)"""
page = request.args.get('page', 1, type=int)
limit = request.args.get('limit', 10, type=int)

# 简单分页逻辑
start_index = (page - 1) * limit
end_index = start_index + limit
paginated_users = users[start_index:end_index]

return jsonify({
'success': True,
'data': {
'total': len(users),
'page': page,
'limit': limit,
'items': paginated_users
}
})

@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""根据 ID 获取用户"""
user = next((u for u in users if u['id'] == user_id), None)
if not user:
return jsonify({
'success': False,
'message': 'User not found'
}), 404

return jsonify({
'success': True,
'data': user
})

@app.route('/api/users', methods=['POST'])
def create_user():
"""创建新用户"""
data = request.get_json()

if not data or 'name' not in data or 'email' not in data:
return jsonify({
'success': False,
'message': 'Name and email are required'
}), 400

# 检查邮箱是否已存在
if any(u['email'] == data['email'] for u in users):
return jsonify({
'success': False,
'message': 'Email already exists'
}), 400

# 创建新用户
new_user = {
'id': max(u['id'] for u in users) + 1 if users else 1,
'name': data['name'],
'email': data['email']
}
users.append(new_user)

return jsonify({
'success': True,
'data': new_user
}), 201

@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""更新用户信息"""
user_index = next((i for i, u in enumerate(users) if u['id'] == user_id), None)
if user_index is None:
return jsonify({
'success': False,
'message': 'User not found'
}), 404

data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': 'No data provided'
}), 400

# 检查邮箱是否被其他用户使用
if 'email' in data and any(u['email'] == data['email'] and u['id'] != user_id for u in users):
return jsonify({
'success': False,
'message': 'Email already exists'
}), 400

# 更新用户信息
if 'name' in data:
users[user_index]['name'] = data['name']
if 'email' in data:
users[user_index]['email'] = data['email']

return jsonify({
'success': True,
'data': users[user_index]
})

@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""删除用户"""
user_index = next((i for i, u in enumerate(users) if u['id'] == user_id), None)
if user_index is None:
return jsonify({
'success': False,
'message': 'User not found'
}), 404

deleted_user = users.pop(user_index)
return jsonify({
'success': True,
'message': f'User {deleted_user["name"]} deleted successfully'
})

# 错误处理
@app.errorhandler(404)
def not_found(error):
return jsonify({
'success': False,
'message': 'Resource not found'
}), 404

@app.errorhandler(500)
def internal_error(error):
return jsonify({
'success': False,
'message': 'Internal server error'
}), 500

@app.errorhandler(400)
def bad_request(error):
return jsonify({
'success': False,
'message': 'Bad request'
}), 400

if __name__ == '__main__':
# CloudBase HTTP 云函数要求监听 9000 端口
app.run(host='0.0.0.0', port=9000, debug=False)

代码说明

让我们详细解释上述代码的关键部分:

  1. Flask 应用创建

    app = Flask(__name__)

    创建 Flask 应用实例,__name__ 参数帮助 Flask 识别应用程序的位置。

  2. 路由装饰器

    @app.route('/')
    def hello():
    return jsonify({'message': 'Hello from Flask!'})

    @app.route('/') 装饰器定义 URL 路由,当访问根 URL 时执行对应函数。

  3. HTTP 方法支持

    @app.route('/api/users', methods=['GET', 'POST'])

    通过 methods 参数指定路由支持的 HTTP 方法。

  4. 错误处理

    @app.errorhandler(404)
    def not_found(error):
    return jsonify({'error': 'Not Found'}), 404

    使用 @app.errorhandler 装饰器处理特定的 HTTP 错误。

第三步:本地测试

启动应用

python app.py

测试 API 接口

应用启动后,您可以通过以下方式测试:

# 测试根路径
curl http://localhost:9000/

# 测试健康检查
curl http://localhost:9000/health

# 测试用户列表
curl http://localhost:9000/api/users

# 测试分页
curl "http://localhost:9000/api/users?page=1&limit=2"

# 测试获取单个用户
curl http://localhost:9000/api/users/1

# 测试创建用户
curl -X POST http://localhost:9000/api/users \
-H "Content-Type: application/json" \
-d '{"name":"新用户","email":"newuser@example.com"}'

# 测试更新用户
curl -X PUT http://localhost:9000/api/users/1 \
-H "Content-Type: application/json" \
-d '{"name":"更新用户","email":"updated@example.com"}'

# 测试删除用户
curl -X DELETE http://localhost:9000/api/users/1

第四步:配置依赖项

生成 requirements.txt 文件:

pip freeze > requirements.txt

⚠️ 注意:只有在虚拟环境中运行上述命令才是安全的,否则它将生成系统上所有已安装的 Python 包,可能导致云函数无法正常启动。

典型的 requirements.txt 内容:

blinker==1.7.0
click==8.1.7
Flask==2.3.3
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.3
Werkzeug==2.3.7

第五步:创建启动脚本

创建 scf_bootstrap 文件(无扩展名):

#!/bin/bash
export PYTHONPATH="./env/lib/python3.10/site-packages:$PYTHONPATH"
/var/lang/python310/bin/python3.10 app.py

为启动脚本添加执行权限:

chmod +x scf_bootstrap

💡 说明

  • scf_bootstrap 是 CloudBase 云函数的启动脚本
  • 设置 Python 路径并启动 Flask 应用
  • 确保应用监听 9000 端口

第六步:准备部署文件

确保您的项目目录结构如下:

flask-cloudbase/
├── env/ # 虚拟环境(可选)
├── app.py # 应用主文件
├── requirements.txt # 依赖列表
└── scf_bootstrap # 启动脚本

第七步:部署到 CloudBase HTTP 云函数

通过控制台部署

  1. 登录 CloudBase 控制台
  2. 选择您的环境,进入「云函数」页面
  3. 点击「新建云函数」
  4. 选择「HTTP 云函数」
  5. 填写函数名称(如:flask-app
  6. 选择运行时:Python 3.10
  7. 提交方法选择:本地上传文件夹
  8. 函数代码选择项目根目录进行上传
  9. 自动安装依赖:开启此选项
  10. 点击「创建」按钮等待部署完成

通过 CLI 部署(敬请期待)

第八步:访问您的应用

部署成功后,您可以参考通过 HTTP 访问云函数设置自定义域名访问HTTP 云函数

您可以测试以下接口:

  • 根路径:/ - 欢迎信息
  • 健康检查:/health - 应用状态
  • 用户列表:/api/users - 获取用户列表
  • 用户详情:/api/users/1 - 获取特定用户
  • 创建用户:POST /api/users - 创建新用户

常见问题

Q: 为什么必须使用 9000 端口?

A: CloudBase HTTP 云函数要求应用监听 9000 端口,这是平台的标准配置。

Q: 如何处理静态文件?

A: Flask 可以通过 static 文件夹处理静态文件,或者使用 CDN 服务。

Q: 如何查看应用日志?

A: 在 CloudBase 控制台的云函数页面,点击函数名称进入详情页查看运行日志。

Q: 支持哪些 Python 版本?

A: CloudBase 支持 Python 3.6、3.7、3.9、3.10 等版本,建议使用 Python 3.10。

Q: 如何处理 CORS 跨域问题?

A: 可以使用 flask-cors 扩展或手动设置响应头来处理跨域请求。

最佳实践

1. 环境变量管理

import os
from flask import Flask

app = Flask(__name__)

# 使用环境变量
app.config['DEBUG'] = os.environ.get('DEBUG', 'False').lower() == 'true'
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'fallback-secret-key')
app.config['DATABASE_URL'] = os.environ.get('DATABASE_URL')

2. 添加 CORS 支持

pip install flask-cors
from flask_cors import CORS

app = Flask(__name__)
CORS(app, origins=['https://yourdomain.com'])

3. 日志配置

import logging
from flask import Flask

app = Flask(__name__)

# 配置日志
if not app.debug:
logging.basicConfig(level=logging.INFO)
app.logger.info('Flask app startup')

4. 蓝图组织代码

from flask import Blueprint

# 创建蓝图
api_bp = Blueprint('api', __name__, url_prefix='/api')

@api_bp.route('/users')
def get_users():
return jsonify({'users': []})

# 注册蓝图
app.register_blueprint(api_bp)

5. 数据库集成

pip install flask-sqlalchemy
from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)

进阶功能

1. 表单处理

pip install flask-wtf

2. 身份验证

pip install flask-login flask-jwt-extended

3. API 文档

pip install flask-restx

4. 缓存支持

pip install flask-caching

下一步