Python----Flask(快速上手，前后端分离模式)

随顺性德

Flask官网地址Flask中文网准备1.Python3.4及更高版本的Python3或Python2.7，推荐使用最新版Python3。2.安装FlaskpipinstallFlask 3.选择一款api测试工具 postmanPostmanAPIPlatformhttps://www.postman.com/apifox  Apifox-API文档、调试、Mock、测试一体化协作平台。拥有接口文档管理、接口调试、Mock、自动化测试等功能，接口开发、测试、联调效率，提升10倍。最好用的接口文档管理工具，接口自动化测试工具。Apifox=Postman+Swagger+Mock+JMeter。集接口文档工具、接口Mock工具、接口自动化测试工具、接口调试工具于一体，提升10倍研发效率。是最好用的API文档工具，API自动化测试工具，APIMock工具，接口文档管理工具，接口文档生成工具。https://apifox.com/?utm_source=baidu_pinzhuan&utm_medium=sem&utm_campaign=pinzhuan&utm_content=pinzhuan&utm_term=apifoxyapiYApi-高效、易用、功能强大的可视化接口管理平台YApi是高效、易用、功能强大的api管理平台，旨在为开发、产品、测试人员提供更优雅的接口管理服务。可以帮助开发者轻松创建、发布、维护API，YApi还为用户提供了优秀的交互体验，开发人员只需利用平台提供的接口数据写入工具以及简单的点击操作就可以实现接口的管理。http://yapi.mglicai.com/简单示例一个最小的Flask应用示例fromflaskimportFlask#初始化appapp=Flask(__name__)@app.route('/')defhello_world():return'Hello,World!'if__name__=="__main__":#host指能够访问该后端服务的ip，一般填写0.0.0.0，表示任意ip都可访问#port指该后端服务部署的端口，不指定时默认为5000#debug指是否开启调试模式，当开启调试模式，修改后端文件会自行重启后端服务app.run(host="0.0.0.0",port=5600,debug=False)执行以上代码后在控制台有以下输出：在浏览器中访问以上两个中的任意地址。或使用接口工具请求。一个包含大部分请求方式的示例fromflaskimportFlaskfromflaskimportrequestfromflaskimportjsonifyimportosapp=Flask(__name__)#设置静态数据us_data={"piter":{"sex":"man","year":"24"}}#一个接口由以下部分组成：路由、请求方法、接口函数。注意接口函数名不可重复。#路由即url除ip和端口部分，使用route()装饰器来把函数绑定到URL。#请求方法即HTTP请求方式，这些请求方式是由HTTP协议定义的，用于描述客户端（如浏览器）如何与服务器进行交互。#Flask支持所有标准的HTTP请求方式，并允许你通过装饰器和视图函数来灵活地处理它们。#GET请求，请求指定的页面信息，并返回实体主体。这是最常见的请求方式，通常用于获取数据或资源。@app.route('/get_data_by_us_name',methods=['GET'])defget_data_by_us_name():#在flask中客户端发送的请求参数都可以使用request来获取。us_name=request.args.get("name")#获取用户数据us_data_get=us_data.get(us_name,{})ifus_data_get=={}:status_code=400result={"msg":"Queryfailed!Pleasecheckthequeryusername.","us_data":us_data_get,"code":status_code}else:status_code=200result={"msg":"Querysuccessful.","us_data":us_data_get,"code":status_code}#jsonify是flask中一个用于生成JSON响应的便捷函数。returnjsonify(result),status_code#POST请求，向指定资源提交数据进行处理请求，数据被包含在请求体中。#提交json@app.route('/add_us_data_by_json',methods=['POST'])defadd_us_data_by_json():globalus_dataadd_us_data=request.get_json()ifadd_us_data["name"]inus_data:status_code=400result={"msg":"Addfailed!Pleasecheckthequerydata.","code":status_code}else:ifadd_us_data["name"]notinus_data:us_data[add_us_data["name"]]={"sex":add_us_data["sex"],"year":add_us_data["year"]}status_code=200result={"msg":"Addsuccessful.","code":status_code}returnjsonify(result),status_code#例如提交表单@app.route('/add_us_data_by_form',methods=['POST'])defadd_us_data_by_form():globalus_dataus_name=request.form.get("name",None)us_sex=request.form.get("sex",None)us_year=request.form.get("year",None)ifus_nameandus_sexandus_year:ifus_namenotinus_data:us_data[us_name]={"sex":us_sex,"year":us_year}status_code=200result={"msg":"Addsuccessful.","code":status_code}else:status_code=400result={"msg":"Addfailed!Pleasecheckthequerydata.","code":status_code}#jsonify是flask中一个用于生成JSON响应的便捷函数。returnjsonify(result),status_code#上传文件@app.route('/add_us_data_by_file',methods=['POST'])defadd_us_data_by_file():us_pic_list=request.files.getlist("pic_list",None)us_pic=request.files.get("us_pic",None)pic_save_name_title=request.form.get("pic_save_name_title",None)ifus_pic_listandus_picandpic_save_name_title:ifnotos.path.isdir("./static")s.mkdir("./static")forget_fileinus_pic_list:get_file.save("./static/{}_{}".format(pic_save_name_title,get_file.filename))us_pic.save("./static/{}_{}".format(pic_save_name_title,us_pic.filename))status_code=200result={"msg":"Addsuccessful.","code":status_code}else:status_code=400result={"msg":"Addfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#PUT请求，从客户端向服务器传送的数据取代指定的资源的内容。@app.route('/up_us_data/',methods=['PUT'])defup_us_data(name):globalus_dataifnameinus_data:us_year=request.form.get("year",None)ifus_year:us_data[name]["year"]=us_yearstatus_code=200result={"msg":"Updatedsuccessfully.","code":status_code}else:result={"msg":"Updatedfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#DELETE请求，请求服务器删除指定的资源或页面。@app.route('/delete_us_data/',methods=['DELETE'])defdelete_us_data(name):globalus_dataifnameinus_data:us_data.pop(name)status_code=200result={"msg":"Deletesuccessfully.","code":status_code}else:result={"msg":"Deletefailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#此外在Flask中，你可以通过检查request.method来确定收到的请求类型，并在一个视图函数中处理多种请求方式。@app.route('/handle_request',methods=['GET','POST'])defhandle_request():ifrequest.method=='GET':#处理GET请求...return"HandlingGETrequest."elifrequest.method=='POST':#处理POST请求...return"HandlingPOSTrequest."if__name__=="__main__":app.run(host="0.0.0.0",port=5600,debug=True)运行代码后即可通过使用接口请求工具来测试接口。一个连接数据库实现增删改查的应用示例fromflaskimportFlaskfromflaskimportrequestfromflaskimportjsonifyfromflask_sqlalchemyimportSQLAlchemy#数据配置mysql_host="xxx.xxx.xxx.xxx"mysql_port=3360mysql_user="xxx"mysql_ps="xxx"mysql_database="xxx"app=Flask(__name__)app.config["SQLALCHEMY_DATABASE_URI"]="mysql+pymysql://{}:{}@{}:{}/{}".format(mysql_user,mysql_ps,mysql_host,mysql_port,mysql_database)#禁用对象修改的跟踪app.config["SQLALCHEMY_TRACK_MODIFICATIONS"]=False#建立连接db=SQLAlchemy(app)#定义用户表classUser(db.Model):id=db.Column(db.Integer,primary_key=True)username=db.Column(db.String(80),unique=True,nullable=False)email=db.Column(db.String(120),unique=True,nullable=False)def__repr__(self):returnf''#创建表withapp.app_context():db.create_all()#增@app.route('/add_us_data_by_json',methods=['POST'])defadd_us_data_by_json():add_us_data=request.get_json()try:new_user=User(username=add_us_data['username'],email=add_us_data['email'])db.session.add(new_user)db.session.commit()#提交事务以保存到数据库status_code=200result={"msg":"Addsuccessful.","code":status_code}except:db.session.rollback()status_code=400result={"msg":"Addfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#删@app.route('/delete_us_data/',methods=['DELETE'])defdelete_us_data(id):user=User.query.get(int(id))#通过id查询ifuser:db.session.delete(user)db.session.commit()#提交事务以删除记录status_code=200result={"msg":"Deletesuccessfully.","code":status_code}else:db.session.rollback()status_code=400result={"msg":"Deletefailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#改@app.route('/up_us_data/',methods=['PUT'])defup_us_data(id):add_us_data=request.get_json()user=User.query.get(int(id))ifuser:user.email=add_us_data['email']user.username=add_us_data['username']db.session.commit()#提交事务以保存更新status_code=200result={"msg":"Updatedsuccessfully.","code":status_code}else:db.session.rollback()result={"msg":"Updatedfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#查@app.route('/get_data_by_us_name',methods=['GET'])defget_data_by_us_name():us_id=request.args.get("id",type=int)us_data=User.query.get(us_id)ifnotus_data:status_code=400result={"msg":"Queryfailed!Pleasecheckthequeryusername.","us_data":{},"code":status_code}else:us_data_get={"id":us_data.id,"username":us_data.username,"email":us_data.email}status_code=200result={"msg":"Querysuccessful.","us_data":us_data_get,"code":status_code}returnjsonify(result),status_codeif__name__=="__main__":app.run(host="0.0.0.0",port=5600,debug=True)注意代码中数据库配置需要修改为实际的数据库配置，运行代码后即可通过使用接口请求工具来测试接口。一个完整的flask后端项目结构如上图所示，model文件通常用于定义数据表；static为静态文件夹用于存放一些用户上传的文件等；web用于存放蓝图，一般一个功能模块一个蓝图；app文件为程序入口，用于启动整个后端；congfig文件用于存放配置信息，如数据库配置等；requirements为该项目的依赖。将上个例子拆解为这种结构后，对应代码如下：app.pyfromflask_corsimportCORSfromwebimportcreate_appapp=create_app()#允许跨域请求CORS(app,supports_credentials=True)if__name__=='__main__':app.run(debug=False,host="0.0.0.0",port=5600)config.pymysql_host="xxx.xxx.xxx.xxx"mysql_port=3360mysql_user="xxx"mysql_ps="xxx"mysql_database="xxx"classConfig():"""工程配置信息"""#数据库的配置信息SQLALCHEMY_DATABASE_URI="mysql+pymysql://{}:{}@{}:{}/{}".format(mysql_user,mysql_ps,mysql_host,mysql_port,mysql_database)SQLALCHEMY_TRACK_MODIFICATIONS=False'运行运行models.pyfromflask_sqlalchemyimportSQLAlchemydb=SQLAlchemy()classUser(db.Model):id=db.Column(db.Integer,primary_key=True)username=db.Column(db.String(80),unique=True,nullable=False)email=db.Column(db.String(120),unique=True,nullable=False)def__repr__(self):returnf''web----__init__.pyfromflaskimportFlaskfrommodel.modelsimportdbfromconfigimportConfigdefcreate_app():app=Flask(__name__)app.config.from_object(Config)app.config['JSON_AS_ASCII']=False#查询时会显示原始SQL语句app.config['SQLALCHEMY_ECHO']=False#注册蓝图fromweb.userimportuserapp.register_blueprint(user)db.init_app(app)withapp.app_context():db.create_all()returnappweb---user---views.pyfrom.importuserfromflaskimportrequest,jsonifyfrommodel.modelsimportdb,User#增@user.route('/add_us_data_by_json',methods=['POST'])defadd_us_data_by_json():add_us_data=request.get_json()try:new_user=User(username=add_us_data['username'],email=add_us_data['email'])db.session.add(new_user)db.session.commit()#提交事务以保存到数据库status_code=200result={"msg":"Addsuccessful.","code":status_code}except:db.session.rollback()status_code=400result={"msg":"Addfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#删@user.route('/delete_us_data/',methods=['DELETE'])defdelete_us_data(id):user=User.query.get(int(id))#通过id查询ifuser:db.session.delete(user)db.session.commit()#提交事务以删除记录status_code=200result={"msg":"Deletesuccessfully.","code":status_code}else:db.session.rollback()status_code=400result={"msg":"Deletefailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#改@user.route('/up_us_data/',methods=['PUT'])defup_us_data(id):add_us_data=request.get_json()user=User.query.get(int(id))ifuser:user.email=add_us_data['email']user.username=add_us_data['username']db.session.commit()#提交事务以保存更新status_code=200result={"msg":"Updatedsuccessfully.","code":status_code}else:db.session.rollback()result={"msg":"Updatedfailed!Pleasecheckthequerydata.","code":status_code}returnjsonify(result),status_code#查@user.route('/get_data_by_us_name',methods=['GET'])defget_data_by_us_name():us_id=request.args.get("id",type=int)us_data=User.query.get(us_id)ifnotus_data:status_code=400result={"msg":"Queryfailed!Pleasecheckthequeryusername.","us_data":{},"code":status_code}else:us_data_get={"id":us_data.id,"username":us_data.username,"email":us_data.email}status_code=200result={"msg":"Querysuccessful.","us_data":us_data_get,"code":status_code}returnjsonify(result),status_code web---user---__init__.pyfromflaskimportBlueprintuser=Blueprint('user',__name__)from.importviews结束语        Flask是一个用Python编写的轻量级Web应用框架。它旨在提供一个简单、可扩展的基础，用于构建Web服务和Web应用程序。以上仅是一些简单使用的例子，能帮助你快上手并跑通。在工作实践还会遇到更多复杂的场景，比如token认证、复杂的sql查询等。如需了解更多关于flask的知识请阅读官方文档。