1. 搭建OTA升级后台基础框架

任务说明

本案例搭建一个最小可用的OTA升级后台服务，提供三个核心接口：固件版本上传/登记、设备注册、升级任务创建。后台使用Python Flask实现，数据存储采用SQLite，无需额外安装数据库服务。通过HTTP接口即可完成固件管理、设备管理和升级任务下发前的数据准备。

环境依赖

Python 3.8+
Flask 2.x
SQLite3（Python内置）

安装依赖：pip install flask

完整实现代码

import sqlite3
import os
from datetime import datetime
from flask import Flask, request, jsonify, g

app = Flask(__name__)
DB_PATH = os.path.join(os.path.dirname(__file__), 'ota.db')

# ---------- 数据库初始化 ----------
def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect(DB_PATH)
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(exc):
    db = g.pop('db', None)
    if db is not None:
        db.close()

def init_db():
    db = sqlite3.connect(DB_PATH)
    db.executescript('''
        CREATE TABLE IF NOT EXISTS firmware (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            version TEXT NOT NULL UNIQUE,
            file_path TEXT NOT NULL,
            md5 TEXT NOT NULL,
            created_at TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS device (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT NOT NULL UNIQUE,
            model TEXT NOT NULL,
            current_version TEXT NOT NULL,
            registered_at TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS upgrade_task (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            firmware_version TEXT NOT NULL,
            target_model TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            created_at TEXT NOT NULL
        );
    ''')
    db.commit()
    db.close()

# ---------- 固件版本管理 ----------
@app.route('/firmware', methods=['POST'])
def upload_firmware():
    data = request.get_json()
    if not data or not all(k in data for k in ('version', 'file_path', 'md5')):
        return jsonify({'error': '缺少version/file_path/md5字段'}), 400
    db = get_db()
    try:
        db.execute(
            'INSERT INTO firmware (version, file_path, md5, created_at) VALUES (?,?,?,?)',
            (data['version'], data['file_path'], data['md5'], datetime.now().isoformat())
        )
        db.commit()
    except sqlite3.IntegrityError:
        return jsonify({'error': '固件版本已存在'}), 409
    return jsonify({'message': '固件登记成功', 'version': data['version']}), 201

@app.route('/firmware', methods=['GET'])
def list_firmware():
    db = get_db()
    rows = db.execute('SELECT version, file_path, md5, created_at FROM firmware ORDER BY id DESC').fetchall()
    return jsonify([dict(r) for r in rows])

# ---------- 设备注册 ----------
@app.route('/device', methods=['POST'])
def register_device():
    data = request.get_json()
    if not data or not all(k in data for k in ('device_id', 'model', 'current_version')):
        return jsonify({'error': '缺少device_id/model/current_version字段'}), 400
    db = get_db()
    try:
        db.execute(
            'INSERT INTO device (device_id, model, current_version, registered_at) VALUES (?,?,?,?)',
            (data['device_id'], data['model'], data['current_version'], datetime.now().isoformat())
        )
        db.commit()
    except sqlite3.IntegrityError:
        return jsonify({'error': '设备ID已注册'}), 409
    return jsonify({'message': '设备注册成功', 'device_id': data['device_id']}), 201

@app.route('/device', methods=['GET'])
def list_devices():
    db = get_db()
    rows = db.execute('SELECT device_id, model, current_version, registered_at FROM device ORDER BY id DESC').fetchall()
    return jsonify([dict(r) for r in rows])

# ---------- 升级任务创建 ----------
@app.route('/task', methods=['POST'])
def create_task():
    data = request.get_json()
    if not data or not all(k in data for k in ('firmware_version', 'target_model')):
        return jsonify({'error': '缺少firmware_version/target_model字段'}), 400
    db = get_db()
    # 校验固件版本存在
    fw = db.execute('SELECT version FROM firmware WHERE version = ?', (data['firmware_version'],)).fetchone()
    if not fw:
        return jsonify({'error': '固件版本不存在'}), 404
    db.execute(
        'INSERT INTO upgrade_task (firmware_version, target_model, created_at) VALUES (?,?,?)',
        (data['firmware_version'], data['target_model'], datetime.now().isoformat())
    )
    db.commit()
    return jsonify({'message': '升级任务创建成功'}), 201

@app.route('/task', methods=['GET'])
def list_tasks():
    db = get_db()
    rows = db.execute('SELECT firmware_version, target_model, status, created_at FROM upgrade_task ORDER BY id DESC').fetchall()
    return jsonify([dict(r) for r in rows])

if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=5000, debug=True)

运行方式

将上述代码保存为 ota_server.py。
在终端执行 python ota_server.py，服务启动在5000端口。

接口调用与验证

使用curl或Postman按以下顺序调用接口，即可完成一次完整的后台数据准备流程：

# 1. 登记固件版本 v1.0
curl -X POST http://localhost:5000/firmware \
  -H "Content-Type: application/json" \
  -d '{"version":"1.0","file_path":"/firmware/esp32_v1.0.bin","md5":"abc123"}'

# 2. 注册设备
curl -X POST http://localhost:5000/device \
  -H "Content-Type: application/json" \
  -d '{"device_id":"dev_001","model":"esp32","current_version":"0.9"}'

# 3. 创建升级任务
curl -X POST http://localhost:5000/task \
  -H "Content-Type: application/json" \
  -d '{"firmware_version":"1.0","target_model":"esp32"}'

# 4. 查询所有任务，确认状态为 pending
curl http://localhost:5000/task

验证通过条件：

步骤1返回 201 且消息为“固件登记成功”；重复提交相同版本返回 409。
步骤2返回 201 且消息为“设备注册成功”。
步骤3返回 201 且消息为“升级任务创建成功”。
步骤4返回的JSON数组中包含一条记录，其 firmware_version 为 "1.0"、target_model 为 "esp32"、status 为 "pending"。
若步骤3中指定不存在的固件版本（如 "9.9"），返回 404 且错误信息为“固件版本不存在”。

数据持久化说明

所有数据保存在运行目录下的 ota.db 文件中。服务重启后数据不丢失。如需重置数据，删除该文件后重新启动服务即可。

📚 课程目录