overview
REST (Representational State Transfer) is adesign风格, 用于构建distributedsystem. REST API is 遵循RESTprinciples API, 它usingHTTPmethod (such asGET, POST, PUT, DELETE) 来operationresource, 并usingJSON or XMLetc.格式来传输data. Flask is a 轻量级 Webframework, 非常适合构建REST API. 本章节将介绍such as何usingFlaskDevelopmentREST API, includingroutingdesign, requestprocessing, responseformat, authenticationauthorizationetc. in 容.
1. RESTfuldesignprinciples
in DevelopmentREST API之 before , 我们需要Understand一些RESTfuldesignprinciples:
1.1 resource and URI
- 每个resource都应该 has 一个唯一 URI (统一resource标识符)
- URI应该using名词而不 is 动词, 例such as
/users而不 is/get_users - using复数形式表示resourcecollection, 例such as
/users表示usercollection,/users/1表示ID for 1 user - using嵌套URI表示resource之间 relationships, 例such as
/users/1/posts表示ID for 1 user 所 has 文章
1.2 HTTPmethod
GET- 获取resourcePOST- creationresourcePUT- updateresource (全量update)PATCH- updateresource (部分update)DELETE- deleteresource
1.3 status码
200 OK- request成功201 Created- resourcecreation成功204 No Content- request成功, 但没 has response in 容400 Bad Request- requestparametererror401 Unauthorized- 未authorization403 Forbidden- 禁止访问404 Not Found- resource不存 in500 Internal Server Error- server in 部error
1.4 data格式
- usingJSONserving as主要data交换格式
- 保持data格式 consistency
- using适当 HTTP头 (such as
Content-Type: application/json)
2. basicREST APIimplementation
让我们 from 一个 simple REST API开始, implementationuserresource CRUD (creation, 读取, update, delete) operation.
2.1 creationFlaskapplication
from flask import Flask, jsonify, request
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
2.2 mockdata
# mockuserdata
users = [
{'id': 1, 'name': '张三', 'email': 'zhangsan@example.com', 'age': 25},
{'id': 2, 'name': '李四', 'email': 'lisi@example.com', 'age': 30},
{'id': 3, 'name': '王五', 'email': 'wangwu@example.com', 'age': 35}
]
2.3 implementationrouting
获取所 has user
@app.route('/api/users', methods=['GET'])
def get_users():
return jsonify({
'status': 'success',
'data': users,
'message': '获取userlist成功'
}), 200
获取单个user
@app.route('/api/users/', methods=['GET'])
def get_user(user_id):
user = next((user for user in users if user['id'] == user_id), None)
if user:
return jsonify({
'status': 'success',
'data': user,
'message': '获取user成功'
}), 200
else:
return jsonify({
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}), 404
creationuser
@app.route('/api/users', methods=['POST'])
def create_user():
# 获取requestdata
data = request.get_json()
# verificationdata
if not data or not 'name' in data or not 'email' in data:
return jsonify({
'status': 'error',
'data': None,
'message': '缺 few 必要 字段: name and email'
}), 400
# creation new user
new_user = {
'id': users[-1]['id'] + 1 if users else 1,
'name': data['name'],
'email': data['email'],
'age': data.get('age', 0)
}
users.append(new_user)
return jsonify({
'status': 'success',
'data': new_user,
'message': 'creationuser成功'
}), 201
updateuser
@app.route('/api/users/', methods=['PUT'])
def update_user(user_id):
# 获取requestdata
data = request.get_json()
# finduser
user = next((user for user in users if user['id'] == user_id), None)
if not user:
return jsonify({
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}), 404
# updateuser
user['name'] = data.get('name', user['name'])
user['email'] = data.get('email', user['email'])
user['age'] = data.get('age', user['age'])
return jsonify({
'status': 'success',
'data': user,
'message': 'updateuser成功'
}), 200
deleteuser
@app.route('/api/users/', methods=['DELETE'])
def delete_user(user_id):
# finduser
global users
user = next((user for user in users if user['id'] == user_id), None)
if not user:
return jsonify({
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}), 404
# deleteuser
users = [user for user in users if user['id'] != user_id]
return jsonify({
'status': 'success',
'data': None,
'message': 'deleteuser成功'
}), 204
2.4 runapplication
if __name__ == '__main__':
app.run(debug=True)
3. usingFlask-RESTfulscale
Flask-RESTful is Flask 一个scale, 它providing了更简洁 方式来构建REST API. 它允许我们usingclass来定义resource, 并自动processingrouting and requestprocessing.
3.1 installationFlask-RESTful
pip install flask-restful
3.2 basicusing
from flask import Flask, jsonify, request
from flask_restful import Api, Resource
app = Flask(__name__)
api = Api(app)
# mockdata
users = [
{'id': 1, 'name': '张三', 'email': 'zhangsan@example.com', 'age': 25},
{'id': 2, 'name': '李四', 'email': 'lisi@example.com', 'age': 30},
{'id': 3, 'name': '王五', 'email': 'wangwu@example.com', 'age': 35}
]
# 定义userresourceclass
class UserResource(Resource):
def get(self, user_id=None):
if user_id:
# 获取单个user
user = next((user for user in users if user['id'] == user_id), None)
if user:
return {
'status': 'success',
'data': user,
'message': '获取user成功'
}, 200
else:
return {
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}, 404
else:
# 获取所 has user
return {
'status': 'success',
'data': users,
'message': '获取userlist成功'
}, 200
def post(self):
# creationuser
data = request.get_json()
if not data or not 'name' in data or not 'email' in data:
return {
'status': 'error',
'data': None,
'message': '缺 few 必要 字段: name and email'
}, 400
new_user = {
'id': users[-1]['id'] + 1 if users else 1,
'name': data['name'],
'email': data['email'],
'age': data.get('age', 0)
}
users.append(new_user)
return {
'status': 'success',
'data': new_user,
'message': 'creationuser成功'
}, 201
def put(self, user_id):
# updateuser
data = request.get_json()
user = next((user for user in users if user['id'] == user_id), None)
if not user:
return {
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}, 404
user['name'] = data.get('name', user['name'])
user['email'] = data.get('email', user['email'])
user['age'] = data.get('age', user['age'])
return {
'status': 'success',
'data': user,
'message': 'updateuser成功'
}, 200
def delete(self, user_id):
# deleteuser
global users
user = next((user for user in users if user['id'] == user_id), None)
if not user:
return {
'status': 'error',
'data': None,
'message': f'userID {user_id} 不存 in '
}, 404
users = [user for user in users if user['id'] != user_id]
return {
'status': 'success',
'data': None,
'message': 'deleteuser成功'
}, 204
# 添加resourcerouting
api.add_resource(UserResource, '/api/users', '/api/users/')
if __name__ == '__main__':
app.run(debug=True)
3.3 request解析
Flask-RESTfulproviding了RequestParserclass来processingrequestparameter 解析 and verification:
from flask_restful import reqparse
# creationrequest解析器
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, required=True, help='user名不能 for 空')
parser.add_argument('email', type=str, required=True, help='邮箱不能 for 空')
parser.add_argument('age', type=int, default=0, help='年龄必须 is 整数')
# in resourcemethodinusing
class UserResource(Resource):
def post(self):
# 解析requestparameter
args = parser.parse_args()
# creation new user
new_user = {
'id': users[-1]['id'] + 1 if users else 1,
'name': args['name'],
'email': args['email'],
'age': args['age']
}
users.append(new_user)
return {
'status': 'success',
'data': new_user,
'message': 'creationuser成功'
}, 201
4. dataverification and 序列化
in DevelopmentREST API时, dataverification and 序列化 is 非常 important . 我们可以usingMarshmallowlibrary来简化这些工作.
4.1 installationMarshmallow
pip install marshmallow
4.2 定义模式 (Schema)
from marshmallow import Schema, fields, validate
# 定义user模式
class UserSchema(Schema):
id = fields.Int(dump_only=True)
name = fields.Str(required=True, validate=validate.Length(min=2, max=50))
email = fields.Email(required=True)
age = fields.Int(validate=validate.Range(min=0, max=120))
# creation模式instance
user_schema = UserSchema()
users_schema = UserSchema(many=True)
4.3 using模式fordataverification and 序列化
@app.route('/api/users', methods=['POST'])
def create_user():
# 获取requestdata
data = request.get_json()
# verificationdata
errors = user_schema.validate(data)
if errors:
return jsonify({
'status': 'error',
'data': None,
'message': 'dataverification失败',
'errors': errors
}), 400
# creation new user
new_user = {
'id': users[-1]['id'] + 1 if users else 1,
'name': data['name'],
'email': data['email'],
'age': data.get('age', 0)
}
users.append(new_user)
# 序列化responsedata
result = user_schema.dump(new_user)
return jsonify({
'status': 'success',
'data': result,
'message': 'creationuser成功'
}), 201
4.4 usingFlask-Marshmallow
Flask-Marshmallow is Marshmallow Flaskscale, 它providing了 and Flask 集成:
from flask_marshmallow import Marshmallow
app = Flask(__name__)
ma = Marshmallow(app)
# 定义模式
class UserSchema(ma.Schema):
class Meta:
fields = ('id', 'name', 'email', 'age')
user_schema = UserSchema()
users_schema = UserSchema(many=True)
5. datalibrary集成
in practicalapplicationin, 我们通常会将datastore in datalibraryin. Flask-SQLAlchemy is Flask 一个scale, 它providing了 and SQLAlchemy 集成, 方便我们 in Flaskapplicationinusingdatalibrary.
5.1 installationFlask-SQLAlchemy
pip install flask-sqlalchemy
5.2 configurationdatalibrary
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
5.3 定义model
from datetime import datetime
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
age = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
def __repr__(self):
return f"<User {self.name}>"
5.4 creationdatalibrary表
with app.app_context():
db.create_all()
5.5 implementationREST API
@app.route('/api/users', methods=['GET'])
def get_users():
users = User.query.all()
result = users_schema.dump(users)
return jsonify({
'status': 'success',
'data': result,
'message': '获取userlist成功'
}), 200
@app.route('/api/users/', methods=['GET'])
def get_user(user_id):
user = User.query.get_or_404(user_id)
result = user_schema.dump(user)
return jsonify({
'status': 'success',
'data': result,
'message': '获取user成功'
}), 200
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
errors = user_schema.validate(data)
if errors:
return jsonify({
'status': 'error',
'data': None,
'message': 'dataverification失败',
'errors': errors
}), 400
new_user = User(name=data['name'], email=data['email'], age=data.get('age', 0))
db.session.add(new_user)
db.session.submitting()
result = user_schema.dump(new_user)
return jsonify({
'status': 'success',
'data': result,
'message': 'creationuser成功'
}), 201
@app.route('/api/users/', methods=['PUT'])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()
errors = user_schema.validate(data, partial=True)
if errors:
return jsonify({
'status': 'error',
'data': None,
'message': 'dataverification失败',
'errors': errors
}), 400
if 'name' in data:
user.name = data['name']
if 'email' in data:
user.email = data['email']
if 'age' in data:
user.age = data['age']
db.session.submitting()
result = user_schema.dump(user)
return jsonify({
'status': 'success',
'data': result,
'message': 'updateuser成功'
}), 200
@app.route('/api/users/', methods=['DELETE'])
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.submitting()
return jsonify({
'status': 'success',
'data': None,
'message': 'deleteuser成功'
}), 204
6. authentication and authorization
in DevelopmentREST API时, authentication and authorization is 非常 important . 我们可以usingJWT (JSON Web Token) 来implementationAPI authentication and authorization.
6.1 installationPyJWT
pip install pyjwt
6.2 JWTauthenticationimplementation
import jwt
from datetime import datetime, timedelta
from functools import wraps
# JWTconfiguration
app.config['JWT_SECRET_KEY'] = 'your-jwt-secret-key'
app.config['JWT_EXPIRATION_DELTA'] = timedelta(days=1)
# 生成JWTtoken
def generate_token(user_id):
payload = {
'exp': datetime.utcnow() + app.config['JWT_EXPIRATION_DELTA'],
'iat': datetime.utcnow(),
'sub': user_id
}
return jwt.encode(payload, app.config['JWT_SECRET_KEY'], algorithms='HS256')
# verificationJWTtoken
def verify_token(token):
try:
payload = jwt.decode(token, app.config['JWT_SECRET_KEY'], algorithmss=['HS256'])
return payload['sub']
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
# authentication装饰器
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = None
# from request头in获取token
if 'Authorization' in request.headers:
auth_header = request.headers['Authorization']
if auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
if not token:
return jsonify({
'status': 'error',
'data': None,
'message': '缺 few authenticationtoken'
}), 401
# verificationtoken
user_id = verify_token(token)
if not user_id:
return jsonify({
'status': 'error',
'data': None,
'message': '无效 authenticationtoken'
}), 401
# 将userID添加 to request on under 文
g.user_id = user_id
return f(*args, **kwargs)
return decorated
# loginrouting, 用于获取JWTtoken
@app.route('/api/login', methods=['POST'])
def login():
data = request.get_json()
if not data or not 'email' in data or not 'password' in data:
return jsonify({
'status': 'error',
'data': None,
'message': '缺 few 必要 字段: email and password'
}), 400
# simple loginverification (practicalapplicationin应该querydatalibrary并verificationpassword)
if data['email'] == 'admin@example.com' and data['password'] == 'password':
# 生成JWTtoken
token = generate_token(1) # fake设userID for 1
return jsonify({
'status': 'success',
'data': {
'token': token,
'token_type': 'Bearer',
'expires_in': app.config['JWT_EXPIRATION_DELTA'].total_seconds()
},
'message': 'login成功'
}), 200
else:
return jsonify({
'status': 'error',
'data': None,
'message': '邮箱 or passworderror'
}), 401
# usingauthentication装饰器保护routing
@app.route('/api/protected', methods=['GET'])
@token_required
def protected():
return jsonify({
'status': 'success',
'data': {
'user_id': g.user_id,
'message': '这 is a 受保护 routing'
},
'message': '访问受保护routing成功'
}), 200
6.3 usingFlask-JWT-Extended
Flask-JWT-Extended is a 更完整 JWTscale, 它providing了更 many functions:
pip install flask-jwt-extended
from flask_jwt_extended import JWTmanagementr, create_access_token, jwt_required, get_jwt_identity
# configurationJWT
app.config['JWT_SECRET_KEY'] = 'your-jwt-secret-key'
jwt = JWTmanagementr(app)
# loginrouting
@app.route('/api/login', methods=['POST'])
def login():
data = request.get_json()
if data['email'] == 'admin@example.com' and data['password'] == 'password':
# creation访问token
access_token = create_access_token(identity=1) # fake设userID for 1
return jsonify({
'status': 'success',
'data': {
'access_token': access_token,
'token_type': 'Bearer'
},
'message': 'login成功'
}), 200
else:
return jsonify({
'status': 'error',
'data': None,
'message': '邮箱 or passworderror'
}), 401
# usingJWT装饰器保护routing
@app.route('/api/protected', methods=['GET'])
@jwt_required()
def protected():
current_user_id = get_jwt_identity()
return jsonify({
'status': 'success',
'data': {
'user_id': current_user_id,
'message': '这 is a 受保护 routing'
},
'message': '访问受保护routing成功'
}), 200
7. APIdocumentation
for APIwritingdocumentation is 非常 important , 它可以helpingDevelopment者Understandsuch as何usingAPI. Swagger is a 流行 APIdocumentationtool, 我们可以usingFlask-RESTPlus or Flask-RESTX来生成Swaggerdocumentation.
7.1 installationFlask-RESTX
pip install flask-restx
7.2 usingFlask-RESTX生成APIdocumentation
from flask_restx import Api, Resource, fields
app = Flask(__name__)
api = Api(app, version='1.0', title='userAPI',
description='一个 simple usermanagementAPI',
doc='/api/docs') # documentation访问path
# 定义namespace
ns = api.namespace('users', description='useroperation')
# 定义model
user_model = api.model('User', {
'id': fields.Integer(readOnly=True, description='userID'),
'name': fields.String(required=True, description='user名', min_length=2, max_length=50),
'email': fields.String(required=True, description='邮箱'),
'age': fields.Integer(description='年龄', min=0, max=120),
'created_at': fields.DateTime(readOnly=True, description='creation时间')
})
# mockdata
users = [
{'id': 1, 'name': '张三', 'email': 'zhangsan@example.com', 'age': 25, 'created_at': '2024-01-01T00:00:00'},
{'id': 2, 'name': '李四', 'email': 'lisi@example.com', 'age': 30, 'created_at': '2024-01-02T00:00:00'}
]
# 定义resource
@ns.route('/')
class UserList(Resource):
@ns.marshal_list_with(user_model)
def get(self):
"""获取所 has user"""
return users
@ns.expect(user_model)
@ns.marshal_with(user_model, code=201)
def post(self):
"""creation new user"""
new_user = api.payload
new_user['id'] = users[-1]['id'] + 1 if users else 1
new_user['created_at'] = '2024-01-03T00:00:00'
users.append(new_user)
return new_user, 201
@ns.route('/')
class User(Resource):
@ns.marshal_with(user_model)
def get(self, user_id):
"""获取单个user"""
user = next((user for user in users if user['id'] == user_id), None)
if user:
return user
api.abort(404, f'userID {user_id} 不存 in ')
@ns.expect(user_model)
@ns.marshal_with(user_model)
def put(self, user_id):
"""updateuser"""
user = next((user for user in users if user['id'] == user_id), None)
if user:
user.update(api.payload)
return user
api.abort(404, f'userID {user_id} 不存 in ')
def delete(self, user_id):
"""deleteuser"""
global users
users = [user for user in users if user['id'] != user_id]
return '', 204
if __name__ == '__main__':
app.run(debug=True)
runapplication after , 我们可以through访问http://localhost:5000/api/docs来查看 and testAPIdocumentation.
8. errorprocessing
in DevelopmentREST API时, 我们需要processing各种errorcircumstances, 并返回适当 errorresponse.
8.1 全局errorprocessing
@app.errorhandler(404)
def not_found_error(error):
return jsonify({
'status': 'error',
'data': None,
'message': f'request resource不存 in : {request.url}'
}), 404
@app.errorhandler(500)
def internal_server_error(error):
app.logger.error(f'server in 部error: {error}')
return jsonify({
'status': 'error',
'data': None,
'message': 'server in 部error, 请稍 after 重试'
}), 500
# processingSQLAlchemyerror
from sqlalchemy.exc import SQLAlchemyError
@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
db.session.rollback()
app.logger.error(f'datalibraryerror: {error}')
return jsonify({
'status': 'error',
'data': None,
'message': 'datalibraryoperation失败'
}), 500
8.2 自定义exception
class APIError(Exception):
def __init__(self, message, status_code=400):
self.message = message
self.status_code = status_code
@app.errorhandler(APIError)
def handle_api_error(error):
return jsonify({
'status': 'error',
'data': None,
'message': error.message
}), error.status_code
# in 视graphfunctioninusing
@app.route('/api/users/', methods=['GET'])
def get_user(user_id):
user = User.query.get(user_id)
if not user:
raise APIError(f'userID {user_id} 不存 in ', 404)
result = user_schema.dump(user)
return jsonify({
'status': 'success',
'data': result,
'message': '获取user成功'
}), 200
9. testREST API
test is 确保APIquality important 手段. 我们可以usingunittest or pytest来testREST API.
9.1 usingpytesttestAPI
pip install pytest pytest-flask
import pytest
from app import app, db
@pytest.fixture
def client():
# configurationtestenvironment
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
with app.test_client() as client:
with app.app_context():
db.create_all()
yield client
with app.app_context():
db.drop_all()
def test_get_users(client):
"""test获取userlist"""
response = client.get('/api/users')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'success'
assert isinstance(data['data'], list)
def test_create_user(client):
"""testcreationuser"""
user_data = {
'name': 'testuser',
'email': 'test@example.com',
'age': 25
}
response = client.post('/api/users', json=user_data)
assert response.status_code == 201
data = response.get_json()
assert data['status'] == 'success'
assert data['data']['name'] == 'testuser'
assert data['data']['email'] == 'test@example.com'
9.2 usingPostmantestAPI
Postman is a 流行 APItesttool, 它允许我们:
- 发送各种HTTPrequest
- test不同 requestparameter and request体
- verificationresponsestatus码 and responsedata
- creationtestcollection and automationtest
- 生成APIdocumentation
10. deploymentREST API
当我们DevelopmentcompletionREST API after , 需要将其deployment to produceenvironment.
10.1 usingGunicorn
Gunicorn is a Python WSGI HTTPserver, 适合deploymentFlaskapplication:
pip install gunicorn
# runapplication
gunicorn -w 4 -b 0.0.0.0:5000 app:app
10.2 usingNginxserving as反向proxy
我们可以usingNginxserving as反向proxy, 将request转发 to Gunicornserver:
# Nginxconfigurationexample
server {
listen 80;
server_name example.com;
location / {
proxy_pass http://127.0.0.1:5000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# 静态fileconfiguration
location /static {
alias /path/to/your/app/static;
expires 30d;
}
}
10.3 usingDockerdeployment
Docker is a containerization平台, 它可以简化application deployment:
# Dockerfileexample
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt
COPY . .
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]
# 构建 and runDocker镜像
docker build -t flask-api .
docker run -p 5000:5000 flask-api
summarized
本章节介绍了usingFlaskDevelopmentREST API method, including:
- RESTfuldesignprinciples
- usingFlask原生functionsDevelopmentREST API
- usingFlask-RESTfulscale简化APIDevelopment
- usingMarshmallowfordataverification and 序列化
- and datalibrary集成
- usingJWTimplementationauthentication and authorization
- usingFlask-RESTX生成APIdocumentation
- errorprocessing and test
- APIdeployment
Flask is a 轻量级 Webframework, 非常适合构建REST API. throughusing适当 scale and 遵循best practices, 我们可以Development出 high quality, 可maintenance REST API.