Flask REST APIDevelopment

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 - 获取resource
  • POST - creationresource
  • PUT - 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 - requestparametererror
  • 401 Unauthorized - 未authorization
  • 403 Forbidden - 禁止访问
  • 404 Not Found - resource不存 in
  • 500 Internal Server Error - server in 部error

1.4 data格式

  • usingJSONserving as主要data交换格式
  • 保持data格式 consistency
  • using适当 HTTP头 (such asContent-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.

codeexample

以 under is a 完整 Flask REST APIexample, package含usermanagementfunctions:

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_jwt_extended import JWTmanagementr, create_access_token, jwt_required, get_jwt_identity
from datetime import datetime

# 初始化application
app = Flask(__name__)

# configuration
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'your-jwt-secret-key'

# 初始化scale
db = SQLAlchemy(app)
ma = Marshmallow(app)
jwt = JWTmanagementr(app)

# 定义datamodel
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)
    password = db.Column(db.String(60), 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}>"

# 定义模式
class UserSchema(ma.Schema):
    class Meta:
        fields = ('id', 'name', 'email', 'age', 'created_at')

user_schema = UserSchema()
users_schema = UserSchema(many=True)

# creationdatalibrary表
with app.app_context():
    db.create_all()

# routing
@app.route('/api/register', methods=['POST'])
def register():
    """userregister"""
    data = request.get_json()
    
    # check邮箱 is 否已存 in 
    existing_user = User.query.filter_by(email=data['email']).first()
    if existing_user:
        return jsonify({
            'status': 'error',
            'data': None,
            'message': '邮箱已被register'
        }), 400
    
    # creation new user
    from werkzeug.security import generate_password_hash
    hashed_password = generate_password_hash(data['password'], method='sha256')
    
    new_user = User(
        name=data['name'],
        email=data['email'],
        password=hashed_password,
        age=data.get('age', 0)
    )
    
    db.session.add(new_user)
    db.session.submitting()
    
    return jsonify({
        'status': 'success',
        'data': user_schema.dump(new_user),
        'message': 'register成功'
    }), 201

@app.route('/api/login', methods=['POST'])
def login():
    """userlogin"""
    data = request.get_json()
    
    # verificationuser
    user = User.query.filter_by(email=data['email']).first()
    from werkzeug.security import check_password_hash
    
    if user and check_password_hash(user.password, data['password']):
        # 生成JWTtoken
        access_token = create_access_token(identity=user.id)
        return jsonify({
            'status': 'success',
            'data': {
                'access_token': access_token,
                'token_type': 'Bearer',
                'user': user_schema.dump(user)
            },
            'message': 'login成功'
        }), 200
    else:
        return jsonify({
            'status': 'error',
            'data': None,
            'message': '邮箱 or passworderror'
        }), 401

@app.route('/api/users', methods=['GET'])
@jwt_required()
def get_users():
    """获取所 has user"""
    users = User.query.all()
    return jsonify({
        'status': 'success',
        'data': users_schema.dump(users),
        'message': '获取userlist成功'
    }), 200

@app.route('/api/users/', methods=['GET'])
@jwt_required()
def get_user(user_id):
    """获取单个user"""
    user = User.query.get_or_404(user_id)
    return jsonify({
        'status': 'success',
        'data': user_schema.dump(user),
        'message': '获取user成功'
    }), 200

@app.route('/api/users/', methods=['PUT'])
@jwt_required()
def update_user(user_id):
    """updateuser"""
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    # 只允许update当 before loginuser information
    current_user_id = get_jwt_identity()
    if user.id != current_user_id:
        return jsonify({
            'status': 'error',
            'data': None,
            'message': '没 has permissionupdateotheruser information'
        }), 403
    
    # updateuserinformation
    if 'name' in data:
        user.name = data['name']
    if 'age' in data:
        user.age = data['age']
    
    db.session.submitting()
    
    return jsonify({
        'status': 'success',
        'data': user_schema.dump(user),
        'message': 'updateuser成功'
    }), 200

@app.route('/api/users/', methods=['DELETE'])
@jwt_required()
def delete_user(user_id):
    """deleteuser"""
    user = User.query.get_or_404(user_id)
    
    # 只允许delete当 before loginuser information
    current_user_id = get_jwt_identity()
    if user.id != current_user_id:
        return jsonify({
            'status': 'error',
            'data': None,
            'message': '没 has permissiondeleteotheruser information'
        }), 403
    
    db.session.delete(user)
    db.session.submitting()
    
    return jsonify({
        'status': 'success',
        'data': None,
        'message': 'deleteuser成功'
    }), 204

if __name__ == '__main__':
    app.run(debug=True)

练习题

  1. creation一个Flask REST API, implementationgraph书managementfunctions, includinggraph书 增删改查
  2. usingFlask-RESTfulscale简化APIDevelopment
  3. usingMarshmallowfordataverification and 序列化
  4. 集成SQLAlchemy, 将datastore to datalibraryin
  5. implementationJWTauthentication, 保护APIrouting
  6. usingFlask-RESTX生成APIdocumentation
  7. writingtest用例, testAPI 各种functions