Flask 构建 Web API 入门指南
Flask API 入门实战:SQLAlchemy 连数据库、Marshmallow 做序列化校验,实现博客 API 的完整 CRUD(POST/GET/PUT/DELETE),掌握轻量框架的组装式开发。
直接回答:Flask 用"微框架 + 自选扩展"的方式构建 API:SQLAlchemy 管数据、Marshmallow 管序列化校验、路由装饰器管端点——每个部件都透明可控,是理解 Web API 本质的最佳路径。
Flask 的组装哲学
Flask 核心只有路由与请求响应,其他全靠扩展。这份"什么都要自己选"的自由,恰恰让你理解每个环节在干什么。本文的技术选型:SQLite + Flask-SQLAlchemy(ORM)+ Flask-Marshmallow(序列化)。
搭建项目
pip install flask flask-sqlalchemy flask-marshmallow marshmallow-sqlalchemy
# app.py 骨架
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///blog.db"
db = SQLAlchemy(app)
ma = Marshmallow(app)
定义模型
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
body = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, server_default=db.func.now())
with app.app_context():
db.create_all()
Schema:序列化与校验
from marshmallow import fields, validate, ValidationError
class PostSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Post
load_instance = True
sqla_session = db.session
fields = ("id", "title", "body", "created_at")
dump_only = ("id", "created_at")
title = fields.String(required=True, validate=validate.Length(min=1, max=200))
body = fields.String(required=True, validate=validate.Length(min=1))
post_schema = PostSchema()
posts_schema = PostSchema(many=True)
Schema 决定"哪些字段进出 API"以及"什么数据算合法"——它是模型的对外投影层。
POST:创建文章
from flask import request, jsonify
@app.post("/posts")
def create_post():
try:
post = post_schema.load(request.get_json())
except ValidationError as e:
return jsonify(errors=e.messages), 400
db.session.add(post)
db.session.commit()
return post_schema.jsonify(post), 201
GET:查询文章
@app.get("/posts")
def list_posts():
posts = Post.query.order_by(Post.created_at.desc()).all()
return posts_schema.jsonify(posts)
@app.get("/posts/<int:pid>")
def get_post(pid):
post = db.get_or_404(Post, pid)
return post_schema.jsonify(post)
PUT:更新文章
@app.put("/posts/<int:pid>")
def update_post(pid):
post = db.get_or_404(Post, pid)
try:
post = post_schema.load(request.get_json(), instance=post)
except ValidationError as e:
return jsonify(errors=e.messages), 400
db.session.commit()
return post_schema.jsonify(post)
DELETE:删除文章
@app.delete("/posts/<int:pid>")
def delete_post(pid):
post = db.get_or_404(Post, pid)
db.session.delete(post)
db.session.commit()
return "", 204
本地使用 flask run。示例未实现认证,勿直接公开;上线前添加身份和对象权限、分页,数据库失败时回滚事务。create_all 不迁移已有表结构。
常见问题(FAQ)
Q:Flask 做 API 和 FastAPI 比差在哪?
A:主要差在自动校验与文档:Flask 要手动组装 Marshmallow + 文档工具,FastAPI 类型注解一步到位。Flask 的优势是生态沉淀与心智简单。
Q:项目大了怎么组织?
A:Blueprint 按业务域拆分路由,应用工厂模式(create_app)管理实例,扩展用 db.init_app 延迟绑定——Flask 大项目的标准三件套。
Q:db.session 要手动关吗?
A:Flask-SQLAlchemy 在请求结束时自动处理。后台工作需建立独立 app.app_context();退出上下文时扩展清理会话,不要把请求 session 跨线程复用。
官方参考
本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。