我需要一些技巧来构建包含大约 35000 个静态（不变）JSON 数据的 REST API。
这是我第一次认真构建 REST API，所以我需要一些设计决策建议。
首先，我计划使用 Flask 来构建 API，因为我熟悉它并使用 MongoDB 来存储数据。但是我听说 MongoDB 对于不变的数据不是一个好的选择。
我想知道的是：
步骤：
1) Upload my data to DB
2) Create a REST API that helps the user fetch the data
3) Upload the REST API to some server
4) Test with Postman to see if it works
我的整体想法是否正确？
任何建议都会很棒。提前致谢。
【问题讨论】：
如果您不确定要使用什么数据库，我会选择 PostgreSQL。它是可扩展的，所以如果你需要在你的数据集上构建它，它就可以正常工作。就性能而言，它只取决于它收到多少请求，但我敢打赌它可以处理你扔给它的任何东西。
关于 API，如果你设置了 Flask，那么我推荐包 Flask-Restful。在名为 models.py 的文件中使用 ORM 概述您的数据库。在名为 resources 的文件夹中，制作用作 API 资源的文件。示例是 blogposts.py，它将对所有或单个帖子、帖子、放置和删除单个帖子有一个获取请求。这是我为一个非常轻量级的博客准备的东西。使用 peewee 作为 ORM 和另一个名为 Flask-HTTPAuth 的包进行身份验证。
# blogposts.py
import json
from flask import jsonify, Blueprint, abort, make_response
from flask_restful import (Resource, Api, reqparse, inputs, fields,
url_for, marshal, marshal_with)
from auth import auth
import models
blogpost_fields = {
'id': fields.Integer,
'title': fields.String,
'content': fields.String,
'created': fields.DateTime
}
def blogpost_or_404(id):
try:
blogpost = models.BlogPost.get(models.BlogPost.id==id)
except models.BlogPost.DoesNotExist:
abort(404)
else:
return blogpost
class BlogPostList(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument(
'title',
required=True,
help='No title provided',
location=['form', 'json']
)
self.reqparse.add_argument(
'content',
required=False,
nullable=True,
location=['form', 'json'],
default=''
)
super().__init__()
def get(self):
blogpost = [marshal(blogpost, blogpost_fields)
for blogpost in models.BlogPost.select()]
return {'BlogPosts': blogpost}
@marshal_with(blogpost_fields)
@auth.login_required
def post(self):
args = self.reqparse.parse_args()
blogpost = models.BlogPost.create(**args)
return (blogpost, 201, {
'Location': url_for('resources.blogposts.blogpost', id=blogpost.id)
})
class BlogPost(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument(
'title',
required=False,
help='No title provided',
location=['form', 'json']
)
self.reqparse.add_argument(
'content',
required=False,
nullable=True,
location=['form', 'json'],
default=''
)
super().__init__()
@marshal_with(blogpost_fields)
def get(self, id):
return (blogpost_or_404(id))
@marshal_with(blogpost_fields)
@auth.login_required
def put(self, id):
args = self.reqparse.parse_args()
try:
blogpost = models.BlogPost.select().where(
models.BlogPost.id==id).get()
except models.BlogPost.DoesNotExist:
return make_response(json.dumps(
{'error': 'That blogpost does not exist or is not editable'}
), 403)
else:
query = blogpost.update(**args).where(models.BlogPost.id==id)
query.execute()
blogpost = (blogpost_or_404(id))
return (blogpost, 200, {
'Location': url_for('resources.blogposts.blogpost', id=id)
})
@auth.login_required
def delete(self, id):
try:
blogpost = models.BlogPost.select().where(
models.BlogPost.id==id).get()
except models.BlogPost.DoesNotExist:
return make_response(json.dumps(
{'error': 'That blogpost does not exist or is not editable'}
), 403)
else:
query = blogpost.delete().where(models.BlogPost.id==id)
query.execute()
return '', 204, {'Location': url_for('resources.blogposts.blogposts')}
blogposts_api = Blueprint('resources.blogposts', __name__)
api = Api(blogposts_api)
api.add_resource(
BlogPostList,
'/blogposts',
endpoint='blogposts'
)
api.add_resource(
BlogPost,
'/blogposts/<int:id>',
endpoint='blogpost'
)
资源类有带有 http 方法名称的方法，这是设置允许的方法。例如，如果我尝试在没有 ID 的情况下删除到 /blogposts，它会以不允许的方法响应。删除仅针对单个帖子定义。编组确定响应中包含哪些信息，您可以使用顶部的 blogpost_fields 对其进行定义。在每个类的 init 中，我们定义了请求解析器，它决定了 API 需要的信息。在这个例子中，我们只需要一个标题和帖子内容。在用户资源中，您可以添加电子邮件、用户名、密码、验证密码、管理员状态等内容。
# models.py
import datetime
import jwt
from argon2 import PasswordHasher
from peewee import *
import config
DATABASE = PostgresqlDatabase('blogdb', user=config.DB['USER'], password=config.DB['PW'], host=config.DB['HOST'])
HASHER = PasswordHasher()
class User(Model):
username = CharField(unique=True)
email = CharField(unique=True)
password = CharField()
class Meta:
database = DATABASE
@classmethod
def create_user(cls, username, email, password, **kwargs):
email = email.lower()
try:
cls.select().where(
(cls.email==email)|(cls.username**username)
).get()
except cls.DoesNotExist:
user = cls(username=username, email=email)
user.password = user.set_password(password)
user.save()
return user
else:
raise Exception("User with that email or username already exists")
@staticmethod
def verify_auth_token(token):
try:
payload = jwt.decode(token, config.SECRET_KEY)
return payload['sub']
except jwt.ExpiredSignatureError:
return 'Signature expired. Please log in again.'
except jwt.InvalidTokenError:
return 'Invalid token. Please log in again.'
@staticmethod
def set_password(password):
return HASHER.hash(password)
def verify_password(self, password):
return HASHER.verify(self.password, password)
def generate_auth_token(self, id):
try:
payload = {
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=0, seconds=5),
'iat': datetime.datetime.utcnow(),
'sub': id
}
return jwt.encode(
payload,
config.SECRET_KEY,
algorithm='HS256'
)
except Exception as e:
return e
class BlogPost(Model):
title = CharField(default='', unique=True)
content = TextField(default='')
created = DateTimeField(default=datetime.datetime.now)
class Meta:
database = DATABASE
def initialize():
DATABASE.connect()
DATABASE.create_tables([User, BlogPost], safe=True)
DATABASE.close()
# auth.py
from flask import g
from flask_httpauth import HTTPTokenAuth
import models
auth = HTTPTokenAuth(scheme='Bearer')
@auth.verify_token
def verify_token(token):
user = models.User.verify_auth_token(token)
if user is not None:
g.user = user
return True
return False
如果您曾经使用过像 SQLAlchemy 这样的 ORM，那么 Models 就很容易解释了。我会推荐该软件包，因为您的数据集远远大于此示例中的数据集。 HTTPAuth 允许您使用所需的身份验证方法来装饰您的 API 资源方法。在我的示例中，登录将生成一个 JWT，该 JWT 需要作为 Bearer 令牌随每个请求一起发送。
所有这些都设置好后，您在 app.py 中注册您的 API 蓝图
# app.py
app = Flask(__name__)
app.register_blueprint(users_api, url_prefix='/api/v1')
app.register_blueprint(blogposts_api, url_prefix='/api/v1')
app.register_blueprint(login_api)
就是这样！
【讨论】：