完整 CRUD API(Flask + SQLite)

Flask 是 Python 生态里最轻量的 Web 框架之一,几分钟就能搭起一个可用的 REST API。下面用一个“文章管理”接口演示完整 CRUD:列表、详情、新增、更新、删除,配上 SQLite 零配置存储和 CORS 跨域支持。你可以先在本地跑通,再按同样的思路迁移到 MySQL/PostgreSQL,或换成 FastAPI、Express 实现。整篇示例可以直接当作你的后端脚手架:结构清晰、注释到位、改字段就能用。

1. 安装依赖

mkdir flask-api && cd flask-api
python3 -m venv venv
source venv/bin/activate
pip install flask flask-cors

2. 主应用

from flask import Flask, request, jsonify
from flask_cors import CORS
import sqlite3
from datetime import datetime

app = Flask(__name__)
CORS(app)  # 跨域支持

DATABASE = 'app.db'


def get_db():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    with get_db() as db:
        db.execute('''CREATE TABLE IF NOT EXISTS articles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            content TEXT,
            created_at TEXT DEFAULT (datetime('now'))
        )''')
        db.commit()


init_db()


@app.route('/api/articles', methods=['GET'])
def get_articles():
    with get_db() as db:
        articles = db.execute(
            'SELECT * FROM articles ORDER BY created_at DESC'
        ).fetchall()
    return jsonify([dict(a) for a in articles])


@app.route('/api/articles/<int:id>', methods=['GET'])
def get_article(id):
    with get_db() as db:
        article = db.execute(
            'SELECT * FROM articles WHERE id = ?', (id,)
        ).fetchone()
    if not article:
        return jsonify({'error': '文章不存在'}), 404
    return jsonify(dict(article))


@app.route('/api/articles', methods=['POST'])
def create_article():
    data = request.get_json()
    if not data or not data.get('title'):
        return jsonify({'error': '标题不能为空'}), 400

    with get_db() as db:
        cursor = db.execute(
            'INSERT INTO articles (title, content) VALUES (?, ?)',
            (data['title'], data.get('content', ''))
        )
        db.commit()
        article = db.execute(
            'SELECT * FROM articles WHERE id = ?', (cursor.lastrowid,)
        ).fetchone()
    return jsonify(dict(article)), 201


@app.route('/api/articles/<int:id>', methods=['PUT'])
def update_article(id):
    data = request.get_json()
    with get_db() as db:
        cursor = db.execute(
            'UPDATE articles SET title=?, content=? WHERE id=?',
            (data.get('title'), data.get('content'), id)
        )
        db.commit()
        if cursor.rowcount == 0:
            return jsonify({'error': '文章不存在'}), 404
        article = db.execute(
            'SELECT * FROM articles WHERE id = ?', (id,)
        ).fetchone()
    return jsonify(dict(article))


@app.route('/api/articles/<int:id>', methods=['DELETE'])
def delete_article(id):
    with get_db() as db:
        cursor = db.execute('DELETE FROM articles WHERE id=?', (id,))
        db.commit()
        if cursor.rowcount == 0:
            return jsonify({'error': '文章不存在'}), 404
    return jsonify({'message': '删除成功'})


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

3. 启动和测试

python3 app.py
# 服务运行在 http://localhost:5000

# 测试 API
curl http://localhost:5000/api/articles
curl -X POST -H "Content-Type: application/json" \
  -d '{"title":"Hello Flask","content":"API example"}' \
  http://localhost:5000/api/articles

4. 生产环境部署:别用内置服务器

Flask 自带的开发服务器(app.run)只适合本地调试,它单进程、无并发能力,直接暴露到公网很容易被慢请求拖垮。生产环境建议用 gunicorn 承载:

pip install gunicorn
gunicorn -w 4 -b 127.0.0.1:8000 app:app

4 个 worker 是中小站点常用的起步配置。再配合 Nginx 反向代理,把 80/443 流量转发到 8000,就是一台可用的生产 API。反向代理的配置可以直接套用后端对接文档里的 Nginx 示例。

5. 错误处理:状态码要表意

示例里 404 和 400 都返回了明确的状态码和 JSON 错误信息。生产环境的错误返回最好统一成固定结构,前端按 code 分支处理,而不是解析中文文案:

{
  "error": {
    "code": "ARTICLE_NOT_FOUND",
    "message": "文章不存在"
  }
}

另外记得在写接口里做参数校验:标题为空返回 400,而不是等数据库抛异常。

6. 代码要点解读

  • 参数化查询:所有 SQL 都通过 ? 占位符传参,而不是字符串拼接。这是防 SQL 注入最基本也最有效的一招,任何把用户输入直接拼进 SQL 的写法都应该在 code review 里被拦下。
  • row_factorysqlite3.Row 让查询结果可以用列名取值,转成 dict 后直接 JSON 序列化,省去手写映射。
  • created_at 默认值:在数据库层用 datetime('now') 生成时间,避免应用实例之间时区不一致。
  • CORS:前后端分离时跨域是常态,flask-cors 一行启用,但线上要收敛到白名单。
  • 状态码:创建返回 201、删除返回 200、找不到返回 404、参数错误返回 400,语义清晰前端才好处理。

7. 用 pytest 验证核心逻辑

import pytest
from app import app

@pytest.fixture
def client():
    app.config['TESTING'] = True
    return app.test_client()

def test_create_and_get(client):
    r = client.post('/api/articles', json={'title': '测试'})
    assert r.status_code == 201
    rid = r.get_json()['id']
    r2 = client.get(f'/api/articles/{rid}')
    assert r2.status_code == 200

def test_missing_title(client):
    r = client.post('/api/articles', json={})
    assert r.status_code == 400

自动化测试的作用是让“改坏接口”这件事在合并前就被拦住,而不是等前端联调时才发现。

8. 常见问题

为什么用 SQLite? 零配置、单文件,适合原型和小流量场景。需要多进程写入或更高并发时,再换 MySQL/PostgreSQL,把数据访问层做薄薄一层适配即可。切换数据库时,记得把建表语句和日期函数一起迁移,避免踩语法差异的坑。

前端跨域报错怎么办? CORS(app) 默认放行所有来源,线上建议指定白名单:CORS(app, resources={r"/api/*": {"origins": ["https://your-site.com"]}})

如何加分页? 列表接口加上 limitoffset 参数,再返回总数,就能支撑前端分页或无限滚动。

要不要加登录鉴权? 只要接口不是完全公开,就建议加。最简单的做法是签发 JWT,在请求头带 Authorization: Bearer <token>,服务端校验。Flask 生态里有现成的 flask-jwt-extended 可以做。

一个实际接入场景:前端页面调用

假设你的静态站(比如用 AI 建站工具生成的展示页)需要展示“公司动态”列表,前端直接调用 /api/articles 即可:

fetch('https://api.example.com/api/articles')
  .then(r => r.json())
  .then(list => {
    document.querySelector('#news').innerHTML = list
      .slice(0, 5)
      .map(a => `<li><a href="/article/${a.id}">${a.title}</a></li>`)
      .join('');
  });

注意两点:把 API 地址放到配置项而不是写死在代码里;上线后关掉 debug=True,否则错误堆栈会直接暴露给访问者。

9. 什么时候选 Flask

Flask 适合中小项目、原型验证、内部工具和 AI 生成站点的后端。它的优点是轻、上手快、文档全;缺点是异步和并发不是强项。如果你预期高并发 IO(聊天、推送、长连接),可以考虑 FastAPI(原生 async)或 Node.js。

场景 推荐 理由
小团队后台、管理接口 Flask 轻量、简单、Python 生态
高并发 IO、流式输出 FastAPI 原生 async,性能更好
前后端同构、前端团队主导 Node.js/Express 同一语言
复杂业务、快速迭代 Django 自带 ORM、Admin、认证

对多数建站场景,Flask 搭配 gunicorn 已经足够;等真的出现瓶颈再迁移也不迟,技术选型不必一开始就上最重的方案。

总结

这篇示例演示的 CRUD 是一个通用骨架:参数化查询、JSON 输出、CORS、状态码、测试。换业务字段、换数据库、换语言,套路都差不多。建议你按这个结构搭自己的 API 模板,把错误结构、分页、鉴权一步步补进去,比每次从零写更省心。遇到不确定的细节,优先查官方文档和你所用版本的迁移说明,别照搬旧教程。

参考:Flask 官方文档 https://flask.palletsprojects.com/ ,gunicorn 文档 https://docs.gunicorn.org/

功能特性

  • ✅ 完整 CRUD 操作
  • ✅ SQLite 数据库(零配置)
  • ✅ 参数化查询(防 SQL 注入)
  • ✅ CORS 跨域支持
  • ✅ JSON 输入输出
  • ✅ 错误处理和 HTTP 状态码