方案二：怎么用Flask封装SQLite的API接口？

作者

使用 Flask 封装 SQLite 的 API 接口是一个非常经典且实用的方案。为了让你能够高效、规范地开发，我将推荐使用 Flask-SQLAlchemy（ORM 框架）配合 Flask-RESTful（RESTful API 扩展）来实现。

这种方式可以让你用 Python 对象来操作数据库，而不是写繁琐的原生 SQL 语句，同时能构建出结构清晰的 RESTful 接口。

下面是一个完整的、可直接运行的示例，演示如何构建一个管理“用户”数据的 API。

🛠️ 第一步：环境准备

你需要安装以下 Python 库：

pip install flask flask-sqlalchemy flask-restful

💻 第二步：编写 API 服务代码 (app.py)

我们将创建一个简单的应用，包含数据库模型定义和两个核心资源接口：UserList（用于获取列表和创建用户）和 UserItem（用于获取、修改和删除单个用户）。

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Resource, Api, reqparse
from werkzeug.security import generate_password_hash, check_password_hash

# 1. 初始化应用和扩展
app = Flask(__name__)
# 配置 SQLite 数据库文件路径
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.secret_key = 'your-secret-key' # 用于加密等安全操作

db = SQLAlchemy(app)
api = Api(app)

# 2. 定义数据库模型 (Model)
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

# 3. 定义 API 资源 (Resources)

# --- 用户列表资源 (对应 /users) ---
class UserList(Resource):
    # GET: 获取所有用户
    def get(self):
        users = User.query.all()
        # 将对象列表转换为字典列表以便 JSON 序列化
        return jsonify([
            {'id': user.id, 'username': user.username, 'email': user.email} 
            for user in users
        ])

    # POST: 创建一个新用户
    def post(self):
        json_data = request.get_json() # 获取前端传来的 JSON 数据

        # 简单校验
        if not json_data or not json_data.get('username') or not json_data.get('email'):
            return {'message': '缺少必要字段 (username, email, password)'}, 400

        # 检查用户是否已存在
        if User.query.filter_by(username=json_data['username']).first():
            return {'message': '用户名已存在'}, 409

        # 密码加密存储 (不要存明文！)
        password_hash = generate_password_hash(json_data['password'])

        new_user = User(
            username=json_data['username'],
            email=json_data['email'],
            password_hash=password_hash
        )

        db.session.add(new_user)
        db.session.commit()

        return {'message': '用户创建成功', 'id': new_user.id}, 201

# --- 单个用户资源 (对应 /users/<id>) ---
class UserItem(Resource):
    # GET: 获取单个用户
    def get(self, user_id):
        user = User.query.get(user_id)
        if not user:
            return {'message': '用户不存在'}, 404
        return jsonify({'id': user.id, 'username': user.username, 'email': user.email})

    # PUT: 更新用户信息
    def put(self, user_id):
        user = User.query.get(user_id)
        if not user:
            return {'message': '用户不存在'}, 404

        json_data = request.get_json()

        # 更新字段
        if 'username' in json_data:
            user.username = json_data['username']
        if 'email' in json_data:
            user.email = json_data['email']
        if 'password' in json_data:
            user.password_hash = generate_password_hash(json_data['password'])

        db.session.commit()
        return {'message': '用户更新成功'}

    # DELETE: 删除用户
    def delete(self, user_id):
        user = User.query.get(user_id)
        if not user:
            return {'message': '用户不存在'}, 404

        db.session.delete(user)
        db.session.commit()
        return {'message': '用户删除成功'}

# 4. 注册路由
api.add_resource(UserList, '/users')
api.add_resource(UserItem, '/users/<int:user_id>')

# 5. 启动应用
if __name__ == '__main__':
    with app.app_context():
        db.create_all() # 首次运行时自动创建数据库表
    app.run(debug=True, host='0.0.0.0', port=5000)

🚀 第三步：运行与测试

  1. 启动服务：在终端运行 python app.py。
  2. 测试接口：你可以使用 Postman、curl 或浏览器插件来测试。

示例操作：

  • 创建用户 (POST)
    • URL: http://127.0.0.1:5000/users
    • Body (JSON):
      json { "username": "alice", "email": "alice@example.com", "password": "securepassword123" }
  • 获取所有用户 (GET)
    • URL: http://127.0.0.1:5000/users
  • 获取特定用户 (GET)
    • URL: http://127.0.0.1:5000/users/1

📌 核心要点总结

  1. ORM 的优势：通过 User.query... 这样的语法，你完全不需要写 SELECT * FROM users... 这样的 SQL 语句，代码更易读且不易出错。
  2. 安全性：
    • 密码加密：代码中使用了 werkzeug.security 进行密码哈希，这是生产环境必须的。
    • 防注入：SQLAlchemy 会自动处理参数转义，有效防止 SQL 注入攻击。
  3. RESTful 规范：
    • GET /users：查全部
    • POST /users：增
    • GET /users/<id>：查单个
    • PUT /users/<id>：改
    • DELETE /users/<id>：删

📦 部署到腾讯云

当你把这个 app.py 部署到腾讯云的服务器（如轻量应用服务器或 CVM）后：

  1. 确保腾讯云服务器的安全组开放了 5000 端口（或你配置的其他端口）。
  2. 阿里云的网站就可以通过 http://<腾讯云公网IP>:5000/users 来访问数据了。

发表评论