基于Python构建MySQL RESTful接口：Flask与Django实战对比

夏知夏

发布时间：2026-10-13 | 2人浏览过

原创

本文介绍将MySQL访问封装为独立无状态服务DBGate的设计思路与优劣分析。通过对比Flask-restless与Django REST framework两种Python框架的实现方案，详细讲解配置步骤、代码示例及PUT/PATCH语义差异，帮助开发者快速构建高性能、易扩展的数据库访问层。

在进行游戏服务分层架构设计时，作者计划将MySQL数据库访问剥离为独立服务DBGate，主要基于以下考量：

将数据库请求统一收敛至DBGate，可使该服务保持无状态特性，从而支持水平扩展。

随着业务数据量增长，MySQL需进行分库分表处理。通过DBGGate内部逻辑屏蔽底层细节，对外部调用者透明，无需修改客户端代码。

利用RESTful规范限制数据交互方式，仅开放GET、POST、PATCH、PUT等基础HTTP方法用于增删改查，不支持复杂查询。这一设计契合游戏业务场景，若涉及复杂查询的网站业务则不适用此方案。

DBGate采用多进程架构，便于精确控制与MySQL的连接数，从而实现对数据库访问量的阈值保护，防止过载。

在DBGate层面集中处理访问量统计、慢查询监控及权限控制等逻辑，提升系统可维护性。

当前使用Python实现，未来若需切换其他语言操作MySQL，只需遵循标准HTTP请求即可，避免了语言层面的兼容性问题。

然而，该架构也存在一定弊端：

首要问题是单次请求响应时间增加，因为增加了中间服务层且采用HTTP协议传输。

部署复杂度提升，原有直接操作MySQL的开发思维需转变，初期可能存在适应期。

综合权衡后，利大于弊，因此决定搭建DBGate服务。

手动为每个数据库表编写RESTful接口工作量巨大，庆幸的是Django和Flask均提供了相应解决方案，下文将逐一介绍。

Flask方案

参考资源：flask-restless

flask-restless使用简便，直接展示核心代码：

import datetime
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restless import APIManager
app = Flask(__name__)
db = SQLAlchemy(app)
restless = APIManager(app, flask_sqlalchemy_db=db)
class User(db.Model):
"""
user
"""
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(255), unique=True, nullable=False)
password = db.Column(db.String(255), nullable=False)
create_time = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
login_time = db.Column(db.DateTime)
restless.create_api(User, methods=['GET', 'POST', 'DELETE', 'PATCH', 'PUT'], results_per_page=100)
db.create_all()
if __name__ == '__main__':
app.run(port=25000)

其支持的RESTful操作映射如下：

获取用户列表：GET /user

创建用户：POST /user

获取单个用户：GET /user/1

全量更新单个用户：PUT /user/1

部分更新单个用户：PATCH /user/1

注意事项：

HTTP请求头中必须包含Content-Type: application/json。

在flask-restless中，PUT与PATCH行为一致，仅更新传入的字段，不会覆盖未提及的字段。

Django方案

参考资源：Django REST framework

Django实现相对复杂，但其自带可视化操作界面，示例如下：

QQ20140803 2

1. 在settings.py中添加配置：

REST_FRAMEWORK = {
# Use hyperlinked styles by default.
# Only used if the `serializer_class` attribute is not set on a view.
'DEFAULT_MODEL _SERIALIZER_CLASS':
'rest_framework.serializers .HyperlinkedModelSerializer',
# Use Django's standard `django.contrib.auth` permissions,
# or allow read-only access for unauthenticated users.
'DEFAULT _PERMISSION_CLASSES': [
#'rest_framework.permissions .DjangoModelPermissionsOrAnonReadOnly',
'rest_framework .permissions.IsAdminUser',
]
}

2. 通过startapp命令创建应用demo：

3. 修改demo应用的models.py：

class User(models.Model):
# key是保留字
password = models.IntegerField()
nick = models.CharField(max_length=255)
create_time = models.DateTimeField(default=datetime.datetime.now)

4. 在demo目录下新建serializers.py：

class User(models.Model):
# key是保留字
password = models.IntegerField()
nick = models.CharField(max_length=255)
create_time = models.DateTimeField(default=datetime.datetime.now)

5. 修改demo应用下的views.py：

from django.shortcuts import render
from rest_framework import viewsets
from serializers import UserSerializer
from models import User
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer

6. 在demo目录下新建urls.py：

import os.path
from django.conf.urls import patterns, include, url
from django.conf.urls.static import static
from django.conf import settings
import views
from rest_framework import routers
appname = os.path.basename (os.path.dirname(os.path.abspath(__file__)))
router = routers.DefaultRouter()
router.register('users', views.UserViewSet, appname)
urlpatterns = patterns('',
url(r'^', include(router.urls)),
)

7. 在主项目mysite/urls.py中include demo.urls及rest_framework.urls：

urlpatterns = patterns('',
url(r'^demo/', include('demo.urls')),
url(r'^admin/', include(admin.site.urls)),
url(r'^api-auth/', include('rest_framework.urls', namespace='rest_framework'))
)

8. 执行数据初始化命令：

python manage.py syncdb

随后访问http://127.0.0.1:8000/demo即可看到管理界面：

QQ20140803 1

对应的测试代码如下：

import json
import requests
from urlparse import urljoin
BASE_URL = 'http://127.0.0.1:16500/'
AUTH = ('admin', 'admin')
def test_get_user_list():
rsp = requests.get(urljoin(BASE_URL, '/demo/users/'), auth=AUTH, headers={
'Accept': 'application/json'
})
assert rsp.ok
def test_post_user_list():
json_data = dict(
password=0,
nick='oo',
create_time='2014-03-3T03:3:3'
)
rsp = requests.post(urljoin(BASE_URL, '/demo/users/'), auth=AUTH, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
}, data=json.dumps(json_data))
assert rsp.ok
def test_get_user():
rsp = requests.get(urljoin(BASE_URL, '/demo/users/1'), auth=AUTH, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
})
assert rsp.ok
def test_put_user():
json_data = dict(
password=100,
nick='xx',
create_time='2014-03-3T03:3:3'
)
# 注意最后的 /
rsp = requests.put(urljoin(BASE_URL, '/demo/users/1/'), auth=AUTH, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
}, data=json.dumps(json_data),
)
assert rsp.ok, rsp.status_code
def test_patch_user():
json_data = dict(
password=300,
)
rsp = requests.patch(urljoin(BASE_URL, '/demo/users/1/'), auth=AUTH, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
}, data=json.dumps(json_data),
)
assert rsp.ok, rsp.status_code

Django REST framework严格区分PUT与PATCH语义，这与flask-restless不同，使用时需注意。

总结

以上介绍了使用Python为MySQL实现RESTful接口的两种主流方案，希望有所帮助。如有疑问请留言，小编将及时回复。感谢支持！