完整 CRUD API(Flask + SQLite)

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

功能特性

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