完整 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_factory:
sqlite3.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"]}})。
如何加分页? 列表接口加上 limit 和 offset 参数,再返回总数,就能支撑前端分页或无限滚动。
要不要加登录鉴权? 只要接口不是完全公开,就建议加。最简单的做法是签发 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 状态码