REST是一種軟件架構模式，它的核心概念包括：資源，資源在REST中代表的是URL，表示一個網絡實體。它的使用可以使操作關系變得更加有序，增強URL的可讀性，使資源描述與視圖松耦合，并且可以提供OpenAPI，便于第三方集成等。下面就來講講flask的restful代碼風格

首先，需要引入flask-restful資源包

from flask_restful import *

引入包后將進入正式的使用環節，包括以下三個基本步驟

# 在restful風格中想返回中文，需要取消ascii編碼的配置 app.config["RESTFUL_JSON"] = {"ensure_ascii": False}# 1、創建組件對象 組件對象 = Api(app)# 定義返回數據格式輸出 types_fields = {'id': fields.Integer,'name': fields.String(attribute='type_name') }# 創建解析對象 type_parser = reqparse.RequestParser() # 添加解析后的入參 type_parser.add_argument('typeName',type=str,required=True,help='必須添加新聞分類名',location='form')# 2、定義類視圖 class 自定義視圖類(Resource):# 給指定函數添加指定的裝飾器method_decorators = {"get": [mydecorator1, mydecorator2],"post": [mydecorator2]}# 使用marshal序列化返回字典類型1@marshal_with(types_fields)def get(self):types = NewsType.query.all()return typesdef post(self):# 獲取入參 args = type_parser.parse_args()typeName = args.get('typeName')# 操作數據newsType = NewsType()newsType.Type_name = typeNamedb.session.add(newsType)db.session.commit()# 使用marshal序列化返回字典類型2 [newsType:type_fields]return marshal(newsType,types_fields)# 3、組件添加類視圖 組件對象.add_resource(視圖類, URL資源段)

Example

項目架構（此處沒有編寫相應的html，可根據需求與前幾章內容自行補充）

app.py

from flask_migrate import Migrate, MigrateCommand from flask_script import Manager from app.models.news_model import * from app.models.user_model import *import flask_caching.backendsfrom app import create_app from exts import dbapp = create_app()manager = Manager(app=app)migrate = Migrate(app=app,db=db) manager.add_command('db',MigrateCommand)if __name__ == '__main__':manager.run()

settings.py

import osclass Config:DEBUG = TrueSQLALCHEMY_DATABASE_URI = 'mysql://root:123456@127.0.0.1:3306/newsdb'SQLALCHEMY_TRACK_MODIFICATIONS = TrueSQLALCHEMY_ECHO = True# secret_key加密碼SECRET_KEY = 'JIAMIMA'RECAPTCHA_PUBLIC_KEY = "SHAJDGBCYSDCUDUSAD"RECAPTCHA_PRIVATE_KEY = "AUERHDNSBDSDSADSLADDQ"RECAPTCHA_PARAMETERS = {'hl':'zh','render':'explicit'}RECAPTCHA_DATA_ATTRS = {'theme':'dark'}# 項目路徑BASE_DIR = os.path.dirname(os.path.abspath(__file__))# 靜態文件夾的路徑STATIC_DIR = os.path.join(BASE_DIR,'static')TEMPLATE_DIR = os.path.join(BASE_DIR,'templates')# 頭像的上傳目錄UPLOAD_ICON_DIR = os.path.join(STATIC_DIR,'upload/icon')# 相冊的上傳目錄UPLOAD_PHONE_DIR = os.path.join(STATIC_DIR,'upload/phone')class DevelopmentConfig(Config):ENV = 'development'DEBUG = Trueclass ProductionConfig(Config):ENV = 'production'DEBUG = True

exts下的init.py

from flask_caching import Cache from flask_cors import CORS from flask_restful import Api from flask_sqlalchemy import SQLAlchemy import pymysql pymysql.install_as_MySQLdb()db = SQLAlchemy() cors = CORS() cache = Cache()

app下的init.py

from flask import Flask from app.apis.news_api import news_bp from app.apis.user_api import user_bp from exts import db, cors, cache from settings import DevelopmentConfigconfig = {'CACHE_TYPE': 'redis','CACHE_REDIS_HOST': '127.0.0.1','CACHE_REDIS_PORT': 6379 }def create_app():app = Flask(__name__,static_folder='../static',template_folder='../templates')app.config.from_object(DevelopmentConfig)db.init_app(app=app)cors.init_app(app=app,supports_credentials=True)cache.init_app(app=app,config=config)app.register_blueprint(news_bp)app.register_blueprint(user_bp)return app

utils下的init.py

from random import randomfrom flask import request, g from flask_restful import abortfrom app.models.user_model import User from app.utils.smssend import SmsSendAPIDemo from exts import cachedef sendMessage(phone):SECRET_ID = "********" # 產品密鑰ID，產品標識SECRET_KEY = "************" # 產品私有密鑰，服務端生成簽名信息使用，請嚴格保管，避免泄露BUSINESS_ID = "***************" # 業務ID，易盾根據產品業務特點分配api = SmsSendAPIDemo(SECRET_ID,SECRET_KEY,BUSINESS_ID)# secret_pair = SecretPair(SECRET_ID,SECRET_KEY)# api = SmsSendAPIDemo(BUSINESS_ID,secret_pair)# 隨機產生驗證碼code = ""for i in range(4):ran = random.randint(0,9)code += str(ran)params = {"mobile": phone,"templateId": "10084","paramType": "json","params": {'code':code,'time':'20211224'}# 國際短信對應的國際編碼(非國際短信接入請注釋掉該行代碼)# "internationalCode": "對應的國家編碼"}ret = api.send(params)return ret,codedef check_user():auth = request.headers.get('Authorization')if not auth:abort(401, msg = '請先登錄')mobile = cache.get(auth)if not mobile:abort(402, mag = '無效的令牌')user = User.query.filter(User.phone==mobile).first()if not user:abort(403, msg = '此用戶已被管理員刪除！')g.user = userdef login_required(func):def wrapper(*args,**kwargs):check_user()return func(*args,**kwargs)return wrapper

smssend.py

import hashlib from hashlib import md5 import json import random import time import urllib import urllib.requestclass SmsSendAPIDemo(object):API_URL = "https://sms.dun.163.com/v2/sendsms"VERSION = "v2"def __init__(self, secret_id, secret_key, business_id):"""Args:secret_id (str) 產品密鑰ID，產品標識secret_key (str) 產品私有密鑰，服務端生成簽名信息使用business_id (str) 業務ID，易盾根據產品業務特點分配"""self.secret_id = secret_idself.secret_key = secret_keyself.business_id = business_iddef gen_signature(self, params=None):"""生成簽名信息Args:params (object) 請求參數Returns:參數簽名md5值"""buff = ""for k in sorted(params.keys()):buff += str(k) + str(params[k])buff += self.secret_keyreturn hashlib.md5(buff.encode("utf-8")).hexdigest()def send(self, params):"""請求易盾接口Args:params (object) 請求參數Returns:請求結果，json格式"""params["secretId"] = self.secret_idparams["businessId"] = self.business_idparams["version"] = self.VERSIONparams["timestamp"] = int(time.time() * 1000)params["nonce"] = int(random.random() * 100000000)params["signature"] = self.gen_signature(params)try:params = urllib.parse.urlencode(params)params = params.encode('utf-8')request = urllib.request.Request(self.API_URL, params)content = urllib.request.urlopen(request, timeout=5).read()return json.loads(content)# response = request.post(self.API_URL, data=params)# return response.json()except Exception as ex:print("調用API接口失敗:", str(ex))if __name__ == "__main__":"""示例代碼入口"""SECRET_ID = "***********" # 產品密鑰ID，產品標識SECRET_KEY = "***************************" # 產品私有密鑰，服務端生成簽名信息使用，請嚴格保管，避免泄露BUSINESS_ID = "*******************************" # 業務ID，易盾根據產品業務特點分配api = SmsSendAPIDemo(SECRET_ID, SECRET_KEY, BUSINESS_ID)params = {"mobile": "15010185644","templateId": "10084","paramType": "json","params": "{'code':200','time':'20211224'}"# 國際短信對應的國際編碼(非國際短信接入請注釋掉該行代碼)# "internationalCode": "對應的國家編碼"}ret = api.send(params)if ret is not None:if ret["code"] == 200:taskId = ret["data"]["taskId"]print("taskId = %s" % taskId)else:print ("ERROR: ret.code=%s,msg=%s" % (ret['code'], ret['msg']))

models下的user_model.py

from app.models import BaseModel from exts import dbclass User(BaseModel):username = db.Column(db.String(50),nullable=False)password = db.Column(db.String(128),nullable=False)phone = db.Column(db.String(11),unique=True,nullable=False)icon = db.Column(db.String(256))newsList = db.relationship('News',backref='author')comment = db.relationship('Comment',backref='user')replys = db.relationship('Reply',backref='user')def __str__(self):return self.username

models下的news_model.py

from app.models import BaseModel from exts import dbclass NewsType(BaseModel):__tablename__='news_type'type_name = db.Column(db.String(50),nullable=False)newsList = db.relationship('News',backref='newstype')class News(BaseModel):__tablename__ = 'news'title = db.Column(db.String(100),nullable=False)content = db.Column(db.Text,nullable=False)desc = db.Column(db.String(255),nullable=False)news_type_id = db.Column(db.Integer,db.ForeignKey('news_type.id'))user_id = db.Column(db.Integer,db.ForeignKey('user.id'))comments = db.relationship('Comment',backref='news')def __str__(self):return self.titleclass Comment(BaseModel):__tablename__ = 'comment'content = db.Column(db.String(255),nullable=False)love_num = db.Column(db.Integer,default=0)user_id = db.Column(db.Integer,db.ForeignKey('user.id'))news_id = db.Column(db.Integer,db.ForeignKey('news.id'))replys = db.relationship('Reply',backref='replys')def __str__(self):return self.contentclass Reply(BaseModel):__tablename__ = 'reply'content = db.Column(db.String(255), nullable=False)love_num = db.Column(db.Integer, default=0)user_id = db.Column(db.Integer, db.ForeignKey('user.id'))comment_id = db.Column(db.Integer,db.ForeignKey('comment.id'))def __str__(self):return self.content

apis下的user_api.py

import uuid from random import randomfrom flask import Blueprint, jsonify, session, render_template from flask_restful import Api, Resource, reqparse, inputs, fields, marshal from werkzeug.security import generate_password_hash, check_password_hashfrom app.models.user_model import User from app.utils import sendMessage from exts import cache, dbuser_bp = Blueprint('user',__name__)api = Api(user_bp) sms_parser = reqparse.RequestParser() sms_parser.add_argument('mobile',type=inputs.regex(r'^1[356789]\d{9}$'),required=True,help='手機號碼格式錯誤',location=['form','args'])class toIndexApi(Resource):def get(self):return render_template('test.html')# 發送手機驗證碼 class SendMessageApi(Resource):def post(self):args = sms_parser.parse_args()mobile = args.get()ret,code = sendMessage(mobile)if ret is not None:if ret["code"] == 200:cache.set(mobile+"_code_l", code, timeout=180)return jsonify(cood=200,msg='短信發送成功！')else:print ("ERROR: ret.code=%s,msg=%s" % (ret['code'], ret['msg']))return jsonify(cood=200, msg='短信發送失敗！')return {'status':400,'msg':'驗證碼f發送有誤'}# 輸入 loginRegister_parser = sms_parser.copy() loginRegister_parser.add_argument('code',type=inputs.regex(r'^\d{4}$'),help='必須輸入四位數字驗證碼',required=True,location=['form','args']) # 輸出 user_fields = {'id': fields.Integer,'username': fields.String } # 用戶的登錄和注冊 class LoginAndRegisterApi(Resource):def post(self):args = loginRegister_parser.parse_args()mobile = args.get('mobile')code = args.get('code')cache_code = cache.get(mobile+'_code_l')if cache and code == cache_code:# 數據庫查詢是否存在相應的mobileusers = User.query.filter(User.phone==mobile).all().first()if not users:# 注冊處理user = User()user.phone = mobiles = ''for i in range(13):ran = random.randint(0.9)s += str(ran)user.username = '用戶' + sdb.session.add(user)db.session.commit()# 登陸處理 記住登錄狀態：session,cookie,cache(redis)# 說明用戶是登陸成功的token = str(uuid.uuid4()).replace('-', '') + str(random.randint(100, 999))# 存儲用戶的登錄信息cache.set(token, mobile)# cache.set(mobile+'_status',1)data = {'status': 200,'msg': '用戶登錄成功！','token': token,'users': marshal(users,user_fields) # 輸出的內容定制}return dataelse:return {'errormsg':'驗證碼錯誤','status': 400}# 忘記密碼 class ForgetPasswordApi(Resource):def get(self):s = 'qwertyuiopasdfghjklzxcvbnmMNBVCXZLKJHGFDSAPOIUYTREWQ1234567890'code = ' 'for i in range(4):ran = random.choice(s)code += ran# 保存codesession['code'] = codereturn {'code': code}# 申請重置密碼的輸入 reset_parser = reqparse.RequestParser() reset_parser.add_argument('imageCode',type=inputs.regex(r'^[a-zA-Z0-9]{4}$'),help='必須輸入正確格式的驗證碼',location=['form','args']) # 申請重置密碼 class ResetPasswordApi(Resource):def get(self):args = reset_parser.parse_args()mobile = args.get('mobile')imageCode = args.get('imageCode')code = session.get('code')if code and imageCode.lower() == code.lower():# 判斷手機號碼user = User.query.filter(User.phone == mobile).first()if user:# 發送手機驗證碼ret, smscode = sendMessage(mobile)if ret is not None:if ret["code"] == 200:cache.set(mobile + "_code_r", smscode, timeout=180)return jsonify(status=200, msg='短信發送成功！')else:print("ERROR: ret.code=%s,msg=%s" % (ret['code'], ret['msg']))return jsonify(status=200, msg='短信發送失敗！')else:return {'status':400,'msg':'驗證碼發送有誤'}else:return {'status':400,'msg':'此用戶未注冊，請先注冊！'}else:return {'status':400,'msg':'驗證碼輸入有誤或者超時'}# 更新密碼 # 客戶端要傳入的信息 update_parser = loginRegister_parser.copy() update_parser.add_argument('password',type=inputs.regex(r'^(?=.*\d)(?=.*[a-z])(?=.*[A-Z])[a-zA-Z0-9]{8,10}$'),help='必須輸入包含大小寫字母和數字組合，不能使用特殊字符',location='form') update_parser.add_argument('repassword',type=inputs.regex(r'^(?=.*\d)(?=.*[a-z])(?=.*[A-Z])[a-zA-Z0-9]{8,10}$'),help='必須輸入包含大小寫字母和數字組合，不能使用特殊字符',location='form')class UpdatePasswordApi(Resource):def post(self):args = update_parser.parse_args()code = args.get('code')mobile = args.get('mobile')cache_code = cache.get(mobile+'_code_r')# 判斷驗證碼是否正確if code and cache_code.low() == code.lower():user = User.query.filter(User.phone == mobile).first()password = args.get('password')repassword = args.get('repassword')# 判斷兩次密碼是否一樣if password == repassword:user.password = generate_password_hash(password)db.session.commit()return {'status':200,'msg':'設置密碼成功！'}else:return {'status':400,'msg':'兩次密碼不一致！'}else:return {'status':400,'msg':'驗證碼有誤！'}# 賬號密碼登錄 # 設置需要前端傳入的內容 password_login_parser = sms_parser.copy() password_login_parser.add_argument('password',type=str,help='必須輸入密碼',required=True,location='form') class PasswordLoginApi(Resource):def post(self):args = password_login_parser.parse_args()mobile = args.get('mobile')password = args.get('password')# 判斷用戶user = User.query,filter(User.phone == mobile).first()if user:if check_password_hash(user.password,password):# 說明用戶是登陸成功的token = str(uuid.uuid4()).replace('-','')+str(random.randint(100,999))# 存儲用戶的登錄信息cache.set(token,mobile)# cache.set(mobile+'_status',1)return {'status':200,'msg':'用戶登錄成功！','token':token}return {'status':400,'msg':'賬戶名或密碼有誤！'}# 訪問首頁 api.add_resource(toIndexApi,'/index') # 發送手機驗證碼 api.add_resource(SendMessageApi,'/sms') # 用戶的登錄和注冊 api.add_resource(LoginAndRegisterApi,'/code_login') # 忘記密碼 api.add_resource(ForgetPasswordApi,'/forget_password') # 申請重置密碼 api.add_resource(ResetPasswordApi,'/reset_password') # 重置密碼 api.add_resource(UpdatePasswordApi,'/update_password') # 賬號密碼登錄 api.add_resource(PasswordLoginApi,'/pwdlogin')

apis下的news_api.py

from flask import Blueprint, app, g from flask_restful import Api, Resource, fields, marshal_with, reqparse, marshalfrom app.models.news_model import NewsType, News from app.utils import login_required from exts import dbnews_bp = Blueprint('news',__name__,url_prefix='/news') api = Api(news_bp)# 新聞類型輸出格式 types_fields = {'id': fields.Integer,'name': fields.String(attribute='type_name') } # 新聞類型添加傳入 type_parser = reqparse.RequestParser() type_parser.add_argument('typeName',type=str,required=True,help='必須添加新聞分類名',location='form') # 新聞類型修改傳入 update_type_parser = type_parser.copy() update_type_parser.add_argument('id',type=int,required=True,help='必須要添加修改的新聞分類id') # 新聞類型刪除傳入 delete_type_parser = reqparse.RequestParser() delete_type_parser.add_argument('id',type=int,required=True,help='必須要添加修改的新聞分類id') # 新聞類型api class NewsTypeApi(Resource):@marshal_with(types_fields)def get(self):types = NewsType.query.all()# print(app.url_map)return types# 使用post添加新聞類型def post(self):args = type_parser.parse_args()typeName = args.get('typeName')# 數據庫添加newsType = NewsType()newsType.Type_name = typeNamedb.session.add(newsType)db.session.commit()return marshal(newsType,types_fields)# 修改分類名稱def put(self):args = update_type_parser.parse_args()typeId = args.get('id')new_type_name = args.get('typeName')type_obj = NewsType.query.get(typeId)if type_obj:type_obj.type_name = new_type_namedb.session.commit()data = {'status': 200,'msg': '修改成功','type': marshal(type_obj,types_fields)}else:data={'status': 400,'msg': '類型查找失敗！',}return data# 刪除分類名稱def delete(self):args = delete_type_parser.parse_args()typeId = args.get('id')type_obj = NewsType.query.get(typeId)if type_obj:db.session.delete(type_obj)db.session.commit()data = {'status': 200,'msg': '類型刪除成功！',}else:data = {'status': 400,'msg': '類型刪除失敗！',}return datanews_parser = reqparse.RequestParser() news_parser.add_argument('typeid',type=int,help='必須添加新聞類型id',required=True) news_parser.add_argument('page',type=int)# 自定義fields class AuthorName(fields.Raw):def format(self,value):return value.username# 每條新聞的格式 news_fields = {'id': fields.Integer,'title': fields.String,'desc': fields.String,'datetime': fields.DateTime(attribute='date_time'),'author': AuthorName(attribute='author'),'url': fields.Url('newdetail',absolute=True) # absolute=True代表絕對路徑 }# 新聞api class NewsListApi(Resource):# 獲取某個新聞分類下的新聞def get(self):args = news_parser.parse_args()typeid = args.get('typeid')page = args.get('page',1)# newsType = NewsType.query.get(typeid)pagination = NewsType.query.filter(NewsType.id==typeid).paginate(page=page,per_page=8)data = {'has_more': pagination.has_next,'data': marshal(pagination.items,news_fields),'return_count': len(pagination.items),'html': 'null'}return data# 回復的格式 reply_fields = {'user': AuthorName(attribute='user'),'content': fields.String,'datetime': fields.DateTime(attribute='date_time'),'lovenum': fields.Integer(attribute='love_num') } # 評價的格式 comment_fields = {'user': AuthorName(attribute='user'),'content': fields.String,'datetime': fields.DateTime(attribute='date_time'),'lovenum': fields.Integer(attribute='love_num'),'replys': fields.List(fields.Nested(reply_fields)) } news_detail_fields = {'id': fields.Integer,'title': fields.String,'content': fields.String,'datetime': fields.DateTime(attribute='date_time'),'author': AuthorName(attribute='author'),'comment': fields.List(fields.Nested(comment_fields)) }class NewsDetailApi(Resource):@marshal_with(news_detail_fields)def get(self,id):news = News.query.get(id)return newsdef post(self):pass# 定義新聞發布的傳入 add_news_parser = reqparse.RequestParser() add_news_parser.add_argument('title',type=str,required=True,help='必須填寫新聞標題') add_news_parser.add_argument('content',type=str,required=True,help='必須填寫新聞主體內容') add_news_parser.add_argument('typeid',type=int,required=True,help='必須填寫新聞類型id')class NewsApi(Resource):@login_requireddef post(self):args = add_news_parser.parse_args()title = args.get('title')content = args.get('content')typeId = args.get('typeid')news = News()news.title = titlenews.content = contentnews.news_type_id = typeIdnews.desc = content[:100]+'......'news.user_id = g.user.iddb.session.add(news)db.session.commit()data = {'status': 200,'msg': '新聞發布成功！','news': marshal(news,news_detail_fields)}return datadef patch(self):passdef put(self):passdef delete(self):passapi.add_resource(NewsTypeApi,'/types') api.add_resource(NewsListApi,'/newslist') api.add_resource(NewsDetailApi,'/newsdetail/<int:id>','newsdetail') api.add_resource(NewsApi,'/news')

總結

以上是生活随笔為你收集整理的flask框架初学-10-restful代码风格的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。