# Flask+SQLite简易后端数据存储实现原理

## 核心观点

Flask框架配合SQLite数据库，通过请求级连接管理、参数化查询和数据校验三层机制，实现了轻量、安全、易维护的后端数据存储方案，非常适合小型项目原型开发、教学演示和单机应用场景。
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-9508030/e19cf6bae6bd3d4d7a87f44feca8345f.png)


**架构流程说明：**
![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/7c7f6622c1c25068f1e00cdf159d6451.png)


## 背景与现状

在Web应用开发领域，数据持久化是最基础也是最核心的功能之一。传统企业级方案往往涉及复杂的ORM配置、连接池管理、事务处理等概念，对初学者形成较高的学习门槛。

当前主流方案存在以下问题：

- MySQL/PostgreSQL需要独立安装和运维数据库服务
- ORM框架（如SQLAlchemy）学习曲线陡峭
- 过度工程化导致小型项目开发效率低下

Flask+SQLite方案的优势在于：

- 零配置启动，Python内置sqlite3模块
- 单文件存储，便于备份和迁移
- 轻量级架构，核心代码不超过200行
- 适合快速原型开发和教学演示

![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/8f7ecbf5b813adc50a82484335fe25ee.png)

## 关键要素

### 要素 1：SQLite数据库特性

SQLite是一个嵌入式关系数据库，将整个数据库存储在单个磁盘文件中。与MySQL/PostgreSQL需要独立服务进程不同，SQLite通过文件直接读写，无需网络连接。

核心特性：

- 零安装：Python标准库内置sqlite3模块
- 单文件：所有数据存储在goods.db一个文件中
- ACID事务：支持完整的事务特性
- 跨平台：同一数据库文件可在不同操作系统使用
- 读并发：多个读取操作可并发执行
- 写串行：写入操作自动加锁排队

**数据库初始化代码：**

```python
import sqlite3
import os

DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "goods.db")

def init_db():
    """初始化数据库表结构"""
    db = sqlite3.connect(DB_PATH)
  
    # 创建商品表
    db.execute("""
        CREATE TABLE IF NOT EXISTS goods (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            price REAL NOT NULL,
            stock INTEGER NOT NULL DEFAULT 0,
            category TEXT NOT NULL DEFAULT '',
            description TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL
        )
    """)
  
    # 动态检测并添加新字段（支持数据库迁移）
    cur = db.execute("PRAGMA table_info(goods)")
    columns = {row[1] for row in cur.fetchall()}
  
    if "stock" not in columns:
        db.execute("ALTER TABLE goods ADD COLUMN stock INTEGER NOT NULL DEFAULT 0")
  
    if "created_at" not in columns:
        db.execute("ALTER TABLE goods ADD COLUMN created_at TEXT NOT NULL DEFAULT ''")
  
    db.commit()
    db.close()

# 应用启动时自动初始化
init_db()
```



### 要素 2：Flask的g对象机制

Flask的`g`对象是请求上下文全局变量，每个请求拥有独立的命名空间，生命周期从请求开始到请求结束。

工作原理：

1. 请求到达时，Flask创建新的请求上下文
2. 首次调用`get_db()`时，创建数据库连接并存入`g.db`
3. 同一请求内后续调用直接返回已有连接，实现复用
4. 请求结束时，`teardown_appcontext`钩子自动关闭连接

**连接管理代码：**

```python
from flask import g
import sqlite3

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()
```

![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/94242a4c4ed62fda6a597f68a76eb30d.png)

这种设计实现了请求级连接复用，既避免了频繁创建连接的开销，又保证了线程安全。
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-9508030/2a53718fb2d8cff3cfb9bf51b7d6f855.png)

### 要素 3：三层架构设计

本系统采用经典的三层架构模式：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-9508030/d6c8a44e50b83776c2c5b9f1ae9e706e.png)

**架构分层详解：**
![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/b86efd27c9d07cb5b3edab073f705941.png)

**路由层**：接收HTTP请求，解析参数，返回JSON响应

- 定义RESTful API端点
- 处理URL参数和请求体
- 调用业务逻辑层处理

**业务逻辑层**：执行数据校验和业务规则

- validate_good()函数校验数据完整性
- 类型转换和范围检查
- 错误处理和异常响应

**数据访问层**：管理数据库连接和SQL操作

- get_db()获取连接
- init_db()初始化表结构
- close_db()自动清理资源

**三层架构代码示例：**

```python
from flask import Flask, request, jsonify
from datetime import datetime

app = Flask(__name__)

# ==================== 路由层 ====================
@app.route("/api/goods", methods=["POST"])
def add_good():
    """添加商品接口"""
    # 解析请求体
    data = request.get_json()
  
    # 调用业务逻辑层
    fields, err_resp, err_code = validate_good(data)
    if err_resp:
        return err_resp, err_code
  
    # 调用数据访问层
    db = get_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cursor = db.execute(
        "INSERT INTO goods (name, price, stock, category, description, created_at) VALUES (?, ?, ?, ?, ?, ?)",
        (fields["name"], fields["price"], fields["stock"], fields["category"], fields["description"], now),
    )
    db.commit()
  
    # 返回响应
    good = {"id": cursor.lastrowid, **fields, "created_at": now}
    return jsonify(good), 201

# ==================== 业务逻辑层 ====================
def validate_good(data):
    """数据校验函数"""
    name = data.get("name", "").strip()
    price = data.get("price", 0)
    stock = data.get("stock", 0)
  
    if not name:
        return None, jsonify({"error": "商品名称不能为空"}), 400
  
    try:
        price = float(price)
    except (ValueError, TypeError):
        return None, jsonify({"error": "价格格式无效"}), 400
  
    if price < 0:
        return None, jsonify({"error": "价格不能为负数"}), 400
  
    try:
        stock = int(stock)
    except (ValueError, TypeError):
        return None, jsonify({"error": "库存格式无效"}), 400
  
    if stock < 0:
        return None, jsonify({"error": "库存不能为负数"}), 400
  
    return {"name": name, "price": price, "stock": stock, 
            "category": data.get("category", "").strip(),
            "description": data.get("description", "").strip()}, None, None

# ==================== 数据访问层 ====================
def get_db():
    """获取数据库连接"""
    if "db" not in g:
        g.db = sqlite3.connect(DB_PATH)
        g.db.row_factory = sqlite3.Row
    return g.db
```

### 要素 4：数据校验机制

数据校验采用前后端双重保障策略：

![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/e0902cb0a6ae215fa3a00ae8bac58d19.png)

**前端校验**：提升用户体验

- HTML5 required属性确保必填项
- input type="number"限制数值格式
- 实时反馈减少无效请求

**后端校验**：确保数据安全

- strip()去除首尾空格
- float()/int()类型转换捕获异常
- 业务规则检查（非负数、非空等）
- 白名单校验排序字段防止注入

**前端校验代码示例：**

```html
<form id="goodForm" onsubmit="submitForm(event)">
    <div class="form-group">
        <label>商品名称</label>
        
        <input type="text" id="goodName" required placeholder="请输入商品名称">
    </div>
  
    <div class="form-group">
        <label>价格 (元)</label>
        
        <input type="number" id="goodPrice" required step="0.01" min="0" placeholder="请输入价格">
    </div>
  
    <div class="form-group">
        <label>库存</label>
        
        <input type="number" id="goodStock" required min="0" step="1" placeholder="库存数量">
    </div>
  
    <button type="submit" class="btn-save">保存</button>
</form>

<script>
async function submitForm(e) {
    e.preventDefault();
  
    const body = {
        name: document.getElementById("goodName").value,
        price: document.getElementById("goodPrice").value,
        stock: document.getElementById("goodStock").value,
    };
  
    const res = await fetch("/api/goods", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(body),
    });
  
    const data = await res.json();
    if (res.ok) {
        showToast("添加成功");
    } else {
        showToast(data.error || "操作失败");
    }
}
</script>
```

**后端校验代码示例：**

```python
def validate_good(data):
    """后端数据校验"""
    # 1. 提取并清理数据
    name = data.get("name", "").strip()  # 去除首尾空格
    price = data.get("price", 0)
    stock = data.get("stock", 0)
  
    # 2. 必填项检查
    if not name:
        return None, jsonify({"error": "商品名称不能为空"}), 400
  
    # 3. 类型转换（捕获异常）
    try:
        price = float(price)
    except (ValueError, TypeError):
        return None, jsonify({"error": "价格格式无效"}), 400
  
    # 4. 业务规则检查
    if price < 0:
        return None, jsonify({"error": "价格不能为负数"}), 400
  
    try:
        stock = int(stock)
    except (ValueError, TypeError):
        return None, jsonify({"error": "库存格式无效"}), 400
  
    if stock < 0:
        return None, jsonify({"error": "库存不能为负数"}), 400
  
    # 5. 返回清理后的数据
    return {
        "name": name,
        "price": price,
        "stock": stock,
        "category": data.get("category", "").strip(),
        "description": data.get("description", "").strip()
    }, None, None
```

### 要素 5：安全性防护

系统实现了多重安全防护：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-9508030/21c1f1982309e95962af1ff73803284b.png)

**SQL注入防护对比：**


| 方式          | 代码示例                                           | 安全性         |
| ------------- | -------------------------------------------------- | -------------- |
| ❌ 字符串拼接 | `sql = f"SELECT * FROM goods WHERE name='{name}'"` | 危险！易受攻击 |
| ✅ 参数化查询 | `sql = "SELECT * FROM goods WHERE name=?"`         | 安全！自动转义 |

**攻击场景演示：**

```python
# ❌ 错误示例（易受SQL注入攻击）
name = request.get_json()["name"]
sql = f"SELECT * FROM goods WHERE name='{name}'"  # 危险！
db.execute(sql)

# 攻击场景：
# 用户输入: name = "'; DROP TABLE goods; --"
# 最终SQL: SELECT * FROM goods WHERE name=''; DROP TABLE goods; --'
# 结果: goods表被删除！

# ✅ 正确示例（参数化查询，安全）
name = request.get_json()["name"]
sql = "SELECT * FROM goods WHERE name=?"
cursor = db.execute(sql, (name,))  # 安全！SQLite自动转义
```

**白名单校验代码：**

```python
@app.route("/api/goods", methods=["GET"])
def get_goods():
    """查询商品列表"""
    db = get_db()
  
    # 获取排序参数
    sort = request.args.get("sort", "id")
    order = request.args.get("order", "asc")
  
    # ✅ 白名单校验（防止注入）
    allowed_sorts = {"id", "name", "price", "stock", "category", "created_at"}
    if sort not in allowed_sorts:
        sort = "id"  # 拒绝非法字段
  
    if order not in ("asc", "desc"):
        order = "asc"
  
    # 构建SQL（排序字段已通过白名单校验，可安全拼接）
    sql = f"SELECT * FROM goods ORDER BY {sort} {order.upper()}"
    rows = db.execute(sql).fetchall()
  
    return jsonify([dict(r) for r in rows])
```

**安全性设计要点：**


| 安全措施   | 实现方式                       | 防护目标     |
| ---------- | ------------------------------ | ------------ |
| 参数化查询 | `db.execute(sql, params)`      | SQL注入      |
| 白名单校验 | `if sort not in allowed_sorts` | 非法字段注入 |
| 数据清理   | `.strip()`, `float()`, `int()` | 非法数据     |
| 类型检查   | `try-except`捕获异常           | 类型错误     |
| 范围检查   | `if price < 0`                 | 异常值       |

### 要素 6：RESTful API设计

接口遵循RESTful规范：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-9508030/29aa7496005c4a99a76528d4bef4d971.png)


**API方法详解：**

![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/e0e52eac9cb67940e111c3ce89ed223b.png)

**RESTful API完整代码实现：**

```python
# 查询列表
@app.route("/api/goods", methods=["GET"])
def get_goods():
    db = get_db()
    category = request.args.get("category", "")
    sort = request.args.get("sort", "id")
    order = request.args.get("order", "asc")
  
    # 白名单校验
    allowed_sorts = {"id", "name", "price", "stock", "category", "created_at"}
    if sort not in allowed_sorts:
        sort = "id"
    if order not in ("asc", "desc"):
        order = "asc"
  
    sql = "SELECT * FROM goods"
    params = []
    if category:
        sql += " WHERE category=?"
        params.append(category)
    sql += f" ORDER BY {sort} {order.upper()}"
  
    rows = db.execute(sql, params).fetchall()
    return jsonify([dict(r) for r in rows])

# 添加商品
@app.route("/api/goods", methods=["POST"])
def add_good():
    data = request.get_json()
    fields, err_resp, err_code = validate_good(data)
    if err_resp:
        return err_resp, err_code
  
    db = get_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cursor = db.execute(
        "INSERT INTO goods (name, price, stock, category, description, created_at) VALUES (?, ?, ?, ?, ?, ?)",
        (fields["name"], fields["price"], fields["stock"], fields["category"], fields["description"], now),
    )
    db.commit()
    good = {"id": cursor.lastrowid, **fields, "created_at": now}
    return jsonify(good), 201  # 201 Created

# 更新商品
@app.route("/api/goods/<int:good_id>", methods=["PUT"])
def update_good(good_id):
    data = request.get_json()
    fields, err_resp, err_code = validate_good(data)
    if err_resp:
        return err_resp, err_code
  
    db = get_db()
    cursor = db.execute(
        "UPDATE goods SET name=?, price=?, stock=?, category=?, description=? WHERE id=?",
        (fields["name"], fields["price"], fields["stock"], fields["category"], fields["description"], good_id),
    )
    db.commit()
  
    if cursor.rowcount == 0:
        return jsonify({"error": "商品不存在"}), 404
  
    return jsonify({"id": good_id, **fields})

# 删除商品
@app.route("/api/goods/<int:good_id>", methods=["DELETE"])
def delete_good(good_id):
    db = get_db()
    cursor = db.execute("DELETE FROM goods WHERE id=?", (good_id,))
    db.commit()
  
    if cursor.rowcount == 0:
        return jsonify({"error": "商品不存在"}), 404
  
    return jsonify({"message": "删除成功"})
```

**API接口文档：**


| 方法   | 路径           | 说明     | 请求体 | 响应码  |
| ------ | -------------- | -------- | ------ | ------- |
| GET    | /api/goods     | 查询列表 | -      | 200     |
| POST   | /api/goods     | 添加商品 | JSON   | 201     |
| PUT    | /api/goods/:id | 更新商品 | JSON   | 200/404 |
| DELETE | /api/goods/:id | 删除商品 | -      | 200/404 |
| GET    | /api/stats     | 统计信息 | -      | 200     |

## 实施路径
![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/7ed7c054213f32a6b5e4a0a237b6ad76.png)



**第一阶段：数据库初始化**

1. 定义数据库文件路径（goods.db）
2. 创建表结构，设置字段类型和约束
3. 使用IF NOT EXISTS确保幂等性
4. 通过PRAGMA table_info检测已有字段
5. 动态ALTER TABLE支持字段扩展

**第二阶段：连接管理实现**

1. 定义get_db()函数获取连接
2. 使用g对象存储连接实例
3. 设置row_factory为sqlite3.Row
4. 注册teardown_appcontext钩子
5. 实现自动关闭和资源清理

**第三阶段：路由与业务逻辑**

1. 定义路由装饰器和HTTP方法
2. 解析请求参数和JSON数据
3. 调用validate_good()校验数据
4. 执行SQL操作并提交事务
5. 构造JSON响应返回结果

**第四阶段：前端集成**

1. 使用fetch API发送请求
2. 处理响应数据和错误信息
3. 实现页面动态渲染
4. 添加交互反馈（Toast提示）
5. 处理加载状态和空数据展示

## 总结YXX 

Flask+SQLite简易后端数据存储方案以其零配置、单文件、轻量级的特性，为小型项目开发提供了优雅的解决方案。通过g对象实现请求级连接管理，既保证了性能又确保了线程安全；通过参数化查询和白名单校验有效防范SQL注入；通过三层架构实现职责分离和代码复用。

这套方案虽然简单，但包含了后端开发的核心理念：数据持久化、请求处理、安全防护、架构设计。对于初学者而言，是理解Web后端原理的最佳起点；对于有经验的开发者，是快速原型验证的利器。

当项目规模增长时，可以平滑升级到MySQL/PostgreSQL，引入SQLAlchemy ORM，采用Gunicorn+Nginx部署，但核心原理和设计思想始终不变。这正体现了"先学会走，再学会跑"的渐进式学习路径。
![图片.png](https://developer.qcloudimg.com/http-save/yehe-9508030/214b185e3a1c88f606cae2c702322b50.png)

Flask框架配合SQLite数据库，通过请求级连接管理、参数化查询和数据校验三层机制，实现了轻量、安全、易维护的后端数据存储方案，非常适合小型项目原型开发、教学演示和单机应用场景。

Flask+SQLite简易后端数据存储实现原理

架构设计

Flask+SQLite实现轻量级后端数据存储，结合单文件数据库和请求级连接管理，通过三层架构和参数化查询确保安全高效。该方案零配置启动，适合小型项目原型、教学演示和单机应用，包含数据校验、RESTful API设计及SQL注入防护等核心功能。

SQL注入攻击

数据完整性

原型开发

数据安全

文件存储

对象存储

数据存储

PostgreSQL

数据库

路由

SQLite

Python

HTML5

MySQL

JSON

2026采购季 | AI焕新·智启新局

tencentdb-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Flask+SQLite简易后端数据存储实现原理-腾讯云开发者社区-腾讯云

Flask+SQLite简易后端数据存储实现原理

Flask+SQLite简易后端数据存储实现原理

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐