从零开始：利用 Flask 或 Django 构建 RESTful API

引言

在当今这个数据驱动的时代，构建高效、可扩展的后端服务变得尤为紧张。RESTful API 作为一种设计模式，已经成为今世 Web 开发的标准之一。无论是小型初创公司还是大型企业，都需要一个强大的后端来支持前端应用的快速迭代和用户需求的变革。Flask 和 Django 是 Python 生态中最受欢迎的两个 Web 框架，它们都可以用来构建 RESTful API。本文将具体介绍如何利用这两个框架来构建 RESTful API，并通过实例帮助你理解其核心概念和实际应用。
基础语法介绍

什么是 RESTful API？

REST（Representational State Transfer）是一种软件架构风格，它定义了一组束缚条件和架构原则。RESTful API 是基于这些原则设计的 Web 服务接口，通常利用 HTTP 协议进行通信。RESTful API 的核心特性包罗：

• 无状态性：每个请求都必须包罗所有必要的信息，服务器不会生存任何会话状态。
  
• 统一接口：利用标准的 HTTP 方法（GET、POST、PUT、DELETE 等）来操纵资源。
  
• 资源导向：API 的 URL 应该指向资源，而不是操纵。
  

Flask vs Django

Flask

Flask 是一个轻量级的 Web 框架，非常适合快速开发和小型项目。它的核心非常简单，但可以通过扩展来添加更多功能。Flask 的机动性使得它可以适应各种差异的应用场景。
Django

Django 是一个全栈框架，提供了丰富的功能和工具，适合大型项目和企业级应用。Django 内置了 ORM、认证体系、管理后台等，可以大大减少开发时间和工作量。
核心概念

• 路由：定义 URL 与视图函数之间的映射关系。
  
• 视图：处置惩罚请求并返回相应。
  
• 模子：表现数据库中的数据布局。
  
• 序列化器：将模子对象转换为 JSON 格式，以便通过 API 返回。
  

基础实例

利用 Flask 构建 RESTful API

问题描述

假设我们需要构建一个简单的 API 来管理用户的个人信息。用户可以创建、读取、更新和删除自己的信息。
代码示例

首先，安装 Flask 和 Flask-RESTful：

1. pip install Flask Flask-RESTful

复制代码

然后，创建一个简单的 Flask 应用：

1. from flask import Flask, request
2. from flask_restful import Resource, Api
4. app = Flask(__name__)
5. api = Api(app)
7. # 模拟数据库
8. users = {}
10. class UserResource(Resource):
11.     def get(self, user_id):
12.         if user_id in users:
13.             return users[user_id], 200
14.         else:
15.             return {"message": "User not found"}, 404
17.     def post(self, user_id):
18.         if user_id in users:
19.             return {"message": "User already exists"}, 400
20.         else:
21.             data = request.get_json()
22.             users[user_id] = data
23.             return data, 201
25.     def put(self, user_id):
26.         if user_id in users:
27.             data = request.get_json()
28.             users[user_id].update(data)
29.             return users[user_id], 200
30.         else:
31.             return {"message": "User not found"}, 404
33.     def delete(self, user_id):
34.         if user_id in users:
35.             del users[user_id]
36.             return {"message": "User deleted"}, 200
37.         else:
38.             return {"message": "User not found"}, 404
40. api.add_resource(UserResource, '/users/<string:user_id>')
42. if __name__ == '__main__':
43.     app.run(debug=True)

复制代码

利用 Django 构建 RESTful API

问题描述

同样的场景，我们利用 Django 来构建一个用户管理的 API。
代码示例

首先，安装 Django 和 Django REST framework：

1. pip install Django djangorestframework

复制代码

然后，创建一个新的 Django 项目和应用：

1. django-admin startproject myproject
2. cd myproject
3. python manage.py startapp myapp

复制代码

编辑 myapp/models.py 文件，定义用户模子：

1. from django.db import models
3. class User(models.Model):
4.     name = models.CharField(max_length=100)
5.     email = models.EmailField(unique=True)
7.     def __str__(self):
8.         return self.name

复制代码

编辑 myapp/serializers.py 文件，定义序列化器：

1. from rest_framework import serializers
2. from .models import User
4. class UserSerializer(serializers.ModelSerializer):
5.     class Meta:
6.         model = User
7.         fields = ['id', 'name', 'email']

复制代码

编辑 myapp/views.py 文件，定义视图：

1. from rest_framework import viewsets
2. from .models import User
3. from .serializers import UserSerializer
5. class UserViewSet(viewsets.ModelViewSet):
6.     queryset = User.objects.all()
7.     serializer_class = UserSerializer

复制代码

编辑 myapp/urls.py 文件，定义路由：

1. from django.urls import path, include
2. from rest_framework.routers import DefaultRouter
3. from .views import UserViewSet
5. router = DefaultRouter()
6. router.register(r'users', UserViewSet)
8. urlpatterns = [
9.     path('', include(router.urls)),
10. ]

复制代码

最后，编辑 myproject/urls.py 文件，包罗应用的路由：

1. from django.contrib import admin
2. from django.urls import path, include
4. urlpatterns = [
5.     path('admin/', admin.site.urls),
6.     path('api/', include('myapp.urls')),
7. ]

复制代码

运行迁移下令以创建数据库表：

1. python manage.py migrate

复制代码

启动 Django 开发服务器：

1. python manage.py runserver

复制代码

现在，你可以通过访问 http://127.0.0.1:8000/api/users/ 来测试 API。
进阶实例

利用 Flask 构建复杂的 RESTful API

问题描述

假设我们需要构建一个更复杂的 API，支持分页、过滤和排序功能。
高级代码实例

首先，安装 Flask-SQLAlchemy 和 Flask-Marshmallow：

1. pip install Flask-SQLAlchemy Flask-Marshmallow marshmallow-sqlalchemy

复制代码

然后，创建一个 Flask 应用：

1. from flask import Flask, request, jsonify
2. from flask_sqlalchemy import SQLAlchemy
3. from flask_marshmallow import Marshmallow
5. app = Flask(__name__)
6. app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
7. db = SQLAlchemy(app)
8. ma = Marshmallow(app)
10. # 定义用户模型
11. class User(db.Model):
12.     id = db.Column(db.Integer, primary_key=True)
13.     name = db.Column(db.String(100))
14.     email = db.Column(db.String(100), unique=True)
16. # 定义序列化器
17. class UserSchema(ma.SQLAlchemyAutoSchema):
18.     class Meta:
19.         model = User
21. user_schema = UserSchema()
22. users_schema = UserSchema(many=True)
24. @app.route('/users', methods=['GET'])
25. def get_users():
26.     query = User.query
27.     name = request.args.get('name')
28.     if name:
29.         query = query.filter(User.name.contains(name))
30.    
31.     sort_by = request.args.get('sort_by', 'name')
32.     sort_order = request.args.get('sort_order', 'asc')
33.     if sort_order == 'desc':
34.         query = query.order_by(getattr(User, sort_by).desc())
35.     else:
36.         query = query.order_by(getattr(User, sort_by))
38.     page = request.args.get('page', 1, type=int)
39.     per_page = request.args.get('per_page', 10, type=int)
40.     paginated_users = query.paginate(page, per_page, error_out=False)
41.     return users_schema.jsonify(paginated_users.items)
43. @app.route('/users', methods=['POST'])
44. def create_user():
45.     data = request.get_json()
46.     new_user = User(name=data['name'], email=data['email'])
47.     db.session.add(new_user)
48.     db.session.commit()
49.     return user_schema.jsonify(new_user), 201
51. @app.route('/users/<int:user_id>', methods=['PUT'])
52. def update_user(user_id):
53.     user = User.query.get_or_404(user_id)
54.     data = request.get_json()
55.     user.name = data.get('name', user.name)
56.     user.email = data.get('email', user.email)
57.     db.session.commit()
58.     return user_schema.jsonify(user)
60. @app.route('/users/<int:user_id>', methods=['DELETE'])
61. def delete_user(user_id):
62.     user = User.query.get_or_404(user_id)
63.     db.session.delete(user)
64.     db.session.commit()
65.     return '', 204
67. if __name__ == '__main__':
68.     db.create_all()
69.     app.run(debug=True)

复制代码

利用 Django 构建复杂的 RESTful API

问题描述

同样的场景，我们利用 Django 来构建一个支持分页、过滤和排序的 API。
高级代码实例

首先，安装 Django-Filter：

1. pip install django-filter

复制代码

然后，编辑 myapp/filters.py 文件，定义过滤器：

1. import django_filters
2. from .models import User
4. class UserFilter(django_filters.FilterSet):
5.     name = django_filters.CharFilter(lookup_expr='icontains')
7.     class Meta:
8.         model = User
9.         fields = ['name', 'email']

复制代码

编辑 myapp/views.py 文件，定义视图：

1. from rest_framework import filters
2. from .models import User
3. from .serializers import UserSerializer
4. from .filters import UserFilter
6. class UserViewSet(viewsets.ModelViewSet):
7.     queryset = User.objects.all()
8.     serializer_class = UserSerializer
9.     filter_backends = [filters.SearchFilter, filters.OrderingFilter, django_filters.rest_framework.DjangoFilterBackend]
10.     filterset_class = UserFilter
11.     search_fields = ['name', 'email']
12.     ordering_fields = ['name', 'email']

复制代码

现在，你可以通过访问 http://127.0.0.1:8000/api/users/?name=John&sort_by=name&sort_order=desc&page=1&per_page=10 来测试 API。
实战案例

项目配景

假设你在一个电商平台上工作，需要构建一个 API 来管理商品信息。商品信息包罗名称、描述、代价、库存等。你需要支持商品的增删改查操纵，而且提供分页、过滤和排序功能。
问题描述

• 商品信息需要存储在数据库中。
  
• API 需要支持分页、过滤和排序功能。
  
• 需要处置惩罚并发请求，包管数据的一致性。
  

解决方案

利用 Flask 构建 API

• 模子定义：定义商品模子，利用 SQLAlchemy 作为 ORM。
  
• 序列化器：利用 Flask-Marshmallow 将模子对象转换为 JSON 格式。
  
• 视图函数：定义处置惩罚请求的视图函数，支持分页、过滤和排序。
  
• 并发处置惩罚：利用 Flask-SQLAlchemy 的事件管理功能，包管数据的一致性。
  

代码实现

1. from flask import Flask, request, jsonify
2. from flask_sqlalchemy import SQLAlchemy
3. from flask_marshmallow import Marshmallow
5. app = Flask(__name__)
6. app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///products.db'
7. db = SQLAlchemy(app)
8. ma = Marshmallow(app)
10. # 定义商品模型
11. class Product(db.Model):
12.     id = db.Column(db.Integer, primary_key=True)
13.     name = db.Column(db.String(100))
14.     description = db.Column(db.String(500))
15.     price = db.Column(db.Float)
16.     stock = db.Column(db.Integer)
18. # 定义序列化器
19. class ProductSchema(ma.SQLAlchemyAutoSchema):
20.     class Meta:
21.         model = Product
23. product_schema = ProductSchema()
24. products_schema = ProductSchema(many=True)
26. @app.route('/products', methods=['GET'])
27. def get_products():
28.     query = Product.query
29.     name = request.args.get('name')
30.     if name:
31.         query = query.filter(Product.name.contains(name))
32.    
33.     sort_by = request.args.get('sort_by', 'name')
34.     sort_order = request.args.get('sort_order', 'asc')
35.     if sort_order == 'desc':
36.         query = query.order_by(getattr(Product, sort_by).desc())
37.     else:
38.         query = query.order_by(getattr(Product, sort_by))
40.     page = request.args.get('page', 1, type=int)
41.     per_page = request.args.get('per_page', 10, type=int)
42.     paginated_products = query.paginate(page, per_page, error_out=False)
43.     return products_schema.jsonify(paginated_products.items)
45. @app.route('/products', methods=['POST'])
46. def create_product():
47.     data = request.get_json()
48.     new_product = Product(name=data['name'], description=data['description'], price=data['price'], stock=data['stock'])
49.     db.session.add(new_product)
50.     db.session.commit()
51.     return product_schema.jsonify(new_product), 201
53. @app.route('/products/<int:product_id>', methods=['PUT'])
54. def update_product(product_id):
55.     product = Product.query.get_or_404(product_id)
56.     data = request.get_json()
57.     product.name = data.get('name', product.name)
58.     product.description = data.get('description', product.description)
59.     product.price = data.get('price', product.price)
60.     product.stock = data.get('stock', product.stock)
61.     db.session.commit()
62.     return product_schema.jsonify(product)
64. @app.route('/products/<int:product_id>', methods=['DELETE'])
65. def delete_product(product_id):
66.     product = Product.query.get_or_404(product_id)
67.     db.session.delete(product)
68.     db.session.commit()
69.     return '', 204
71. if __name__ == '__main__':
72.     db.create_all()
73.     app.run(debug=True)

复制代码

利用 Django 构建 API

• 模子定义：定义商品模子，利用 Django 的 ORM。
  
• 序列化器：利用 Django REST framework 将模子对象转换为 JSON 格式。
  
• 视图集：定义处置惩罚请求的视图集，支持分页、过滤和排序。
  
• 并发处置惩罚：利用 Django 的事件管理功能，包管数据的一致性。
  

代码实现

编辑 myapp/models.py 文件，定义商品模子：

1. from django.db import models
3. class Product(models.Model):
4.     name = models.CharField(max_length=100)
5.     description = models.TextField()
6.     price = models.DecimalField(max_digits=10, decimal_places=2)
7.     stock = models.IntegerField()
9.     def __str__(self):
10.         return self.name

复制代码

编辑 myapp/serializers.py 文件，定义序列化器：

1. from rest_framework import serializers
2. from .models import Product
4. class ProductSerializer(serializers.ModelSerializer):
5.     class Meta:
6.         model = Product
7.         fields = ['id', 'name', 'description', 'price', 'stock']

复制代码

编辑 myapp/filters.py 文件，定义过滤器：

1. import django_filters
2. from .models import Product
4. class ProductFilter(django_filters.FilterSet):
5.     name = django_filters.CharFilter(lookup_expr='icontains')
7.     class Meta:
8.         model = Product
9.         fields = ['name', 'price', 'stock']

复制代码

编辑 myapp/views.py 文件，定义视图集：

1. from rest_framework import filters
2. from .models import Product
3. from .serializers import ProductSerializer
4. from .filters import ProductFilter
6. class ProductViewSet(viewsets.ModelViewSet):
7.     queryset = Product.objects.all()
8.     serializer_class = ProductSerializer
9.     filter_backends = [filters.SearchFilter, filters.OrderingFilter, django_filters.rest_framework.DjangoFilterBackend]
10.     filterset_class = ProductFilter
11.     search_fields = ['name', 'description']
12.     ordering_fields = ['name', 'price', 'stock']

复制代码

编辑 myapp/urls.py 文件，定义路由：

1. from django.urls import path, include
2. from rest_framework.routers import DefaultRouter
3. from .views import ProductViewSet
5. router = DefaultRouter()
6. router.register(r'products', ProductViewSet)
8. urlpatterns = [
9.     path('', include(router.urls)),
10. ]

复制代码

编辑 myproject/urls.py 文件，包罗应用的路由：

1. from django.contrib import admin
2. from django.urls import path, include
4. urlpatterns = [
5.     path('admin/', admin.site.urls),
6.     path('api/', include('myapp.urls')),
7. ]

复制代码

运行迁移下令以创建数据库表：

1. python manage.py migrate

复制代码

启动 Django 开发服务器：

1. python manage.py runserver

复制代码

现在，你可以通过访问 http://127.0.0.1:8000/api/products/?name=Apple&sort_by=price&sort_order=desc&page=1&per_page=10 来测试 API。
扩展讨论

性能优化

• 缓存：利用缓存技术（如 Redis）来减少数据库查询次数，提高相应速度。
  
• 异步处置惩罚：利用异步任务队列（如 Celery）来处置惩罚耗时的任务，如发送邮件、天生报表等。
  
• 负载均衡：利用负载均衡器（如 Nginx）来分散请求，提高体系的可用性和性能。
  

安全性

• 身份验证：利用 JWT 或 OAuth2 进行身份验证，确保只有授权用户才能访问敏感数据。
  
• 输入验证：对用户输入进行严格的验证，防止 SQL 注入、XSS 攻击等安全漏洞。
  
• 日志记载：记载关键操纵的日志，便于问题排查和审计。
  

部署与运维

• 容器化：利用 Docker 容器化应用，简化部署和运维。
  
• 持续集成/持续部署：利用 CI/CD 工具（如 Jenkins、GitHub Actions）自动化测试和部署流程。
  
• 监控与报警：利用监控工具（如 Prometheus、Grafana）实时监控应用的状态，实时发现和解决问题。
  

社区与资源

• 官方文档：Flask 和 Django 的官方文档黑白常好的学习资源，发起深入阅读。
  
• 社区论坛：Stack Overflow、Reddit 等社区有很多开发者分享履历和解决问题的方法。
  
• 开源项目：GitHub 上有很多良好的开源项目，可以参考和学习。
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。