赞
踩
Flask是一个轻量级的Python Web应用程序框架,它被广泛用于快速开发简单而灵活的Web应用程序。它以简洁、易用和可扩展为设计原则,并且具有丰富的生态系统和活跃的社区支持。(轻量级,拓展性强)
Werkzeug 是一个 Python 的 WSGI(Web Server Gateway Interface)工具库,它是 Flask 框架的核心组件之一。WSGI 是一个 Python Web 应用程序和 Web 服务器之间的标准接口,它定义了 Web 应用程序如何与服务器进行通信。
# 使用pip安装
pip install flask==1.1
# 升级命令
pip install --upgrade flask
from flask import Flask
app = Flask(__name__)
@app.route("/index", methods=["GET"])
def hello():
return "hello world"
app.run()
Gunicorn(Green Unicorn)是一个用于运行 Python Web 应用程序的 HTTP 服务器。它是一个 WSGI(Web Server Gateway Interface)HTTP 服务器,可以通过与其他 Web 框架和应用程序配合使用来提供高性能的 Web 服务。
特别注意:gunicorn只是一个wsgi服务器,这种服务器是一种中间层用来链接web应用程序和web服务器程序(比如nginx)的。
yum install gunicorn
# 写一个测试py
from flask import Flask
app = Flask(__name__)
@app.route("/index")
def hello():
return "hello world"
# 使用gunicorn运行flaskweb代码,-w 4代表同时开启四个进程
gunicorn -w 4 filename:app
from flask import Flask # Flask实例化生成核心对象app # 后续所有的东西都要和app绑定 # 参数传递,当前模块名 app = Flask(__name__) # 路由和视图函数绑定 # 也可不使用装饰器,使用app的app_url_rule()方法 @app.route("/index") def hello(): return "hello world" # app.add_url_rule("/index", view_func=hello) # 启动内置web服务 app.run()
除此之外,flask可以开启debug模式,开启debug模式之后,当代码发生修改,则会自动重新启动代码,不需要手动重启(热更新)
# 添加代码即可开启
app.debug = True
# 指定端口
app.run(host='0.0.0.0', port='9000')
通过debug可以发现,app对象类有两个属性,这两个属性用于管理 URL 路由和视图函数之间的映射关系。
url_map
属性:
url_map
是一个 Map
对象,用于存储应用程序中定义的路由规则。它包含了所有已注册的 URL 路由和与之相关联的视图函数。url_map
提供了一些有用的方法,如 add()
、update()
和 bind()
,用于添加、更新和绑定路由规则。view_function
属性:
view_function
是一个字典,用于存储应用程序中定义的视图函数及其名称。endpoint
作为键,对应的视图函数对象作为值。app.view_functions
访问 view_function
属性。在url_map中使用url做索引,使用endpoint为值(endpoint不明显定义一般为函数名称,endpoint全局唯一,函数名可以重复)
通过使用url_for
方法可以使用endpoint找到相关的路由信息,传递参数为endpoint
from flask import Flask, url_for
# 访问127.0.0.1/会在控制台打印endpoint为hello的url
@app.route("/")
def root():
print(url_for("hello"))
return "This is /"
通过构造动态url,可以使用不固定的url来传递相应参数,使用动态url时,需在方法中定义相同的形参
@app.route("/sctl/<name>/<passwd>")
def sctl(name, passwd):
return f"my name is {name}, my password is {passwd}"
# 可以指定参数类型
@app.route("/sctl/<name>/<int:passwd>")
def sctl(name, passwd):
return f"my name is {name}, my password is {passwd}"
# 常见的请求方法:get,post,delete,update,head,option
@app.route("/index", methods=['POST', 'GET'])
def hello():
return "hello world"
前文使用动态url的方法能够处理一些url的用户信息,但是十分低效,并且只能处理url的用户传递信息,一般用的的更多的方法是使用request对象,将用户请求的所有信息全部封装在一个对象中。
使用下列代码查看request包的方法
from flask import Flask, request
app = Flask(__name__)
@app.route("/login")
def login():
print(dir(request))
return "This is login"
app.run(debug=True, host='0.0.0.0', port=9000)
request常用方法:
request.url
:返回值为当前访问的完整url
request.method
:返回值为当前url的请求方法
request.args
:常用于get请求,返回值为ImmutableMultiDict
(其实就是字典,type方法的返回值显示也为dict
),内容为请求参数键值对
举例:当name为root,密码为123456时才能登录成功
from flask import Flask, request app = Flask(__name__) @app.route("/login") def login(): print(request.method) print(request.args) context = request.args if context["name"] == "root" and context["passwd"] == "123456": return "login successful" return "login fail" app.run(debug=True, host='0.0.0.0', port=9000)
request.form
:接受form表单的数据,数据类型与request.args
相同
request.json
:接受json格式的数据,数据类型也是字典
将所有的项目代码放入一个文件是可行的,但是这是不可取的,这样会导致代码很乱,不好维护,通过将代码拆分成多个模块,可以加强代码的可阅读性和可维护性
project
|- config --> 存放相关配置参数
|- |- settings.py --> 存放参数设置
|- route
|- |- entity(directory) --> 存放实例对象的路由和视图
|- |- __init__.py --> 设置方法注册路由蓝图
|- server.py --> 项目运行入口
|- app.py --> 加载相关配置,以及生成app实例
一般app.py
的内容
""" @Author: Thousand @file: app.py @time: 2023/7/20 14:49 """ # 导入路由 import router import os from flask import Flask # 加载配置 def create_app(config=None): app = Flask(__name__) app.config.from_object('config.settings') if 'FLASK_CONF' in os.environ: app.config.from_envvar('FLASK_COND') if config is not None: if isinstance(config, dict): app.config.update(config) elif config.endswith(".py"): app.config.form_pyfile(config) router.init_app(app) return app
# 使用Blueprint模块 from flask import Blueprint # 使用blueprint对象生成蓝图 # Blueprint第一个参数为name:这个参数将会被当作endpoint # 第二个参数import_name:用来方方便找到模块的根路径,通常为__name__ # url_prefix:区分url路线,给每个访问的url添加一个前缀,可以用来区分版本 var = blueprint("Student", __name__, url_prefix='/v1') # 添加路由 @var.route(url) def index(): return "" # 上述只是构建了蓝图,必须使用app的register_blueprint()方法来注册蓝图 # 得益于python的函数定义不需要指定类型,所以这里只要app就行 from route.directory.object import var def init_app(app): app.register_blueprint(var)
pymysql库是python用来连接MySQL数据库并用来操作MySQL数据库的。
先建立connect连接对象,然后使用connect连接对象建立sql语句,最后执行sql语句。
在 MySQL 数据库中,游标(Cursor)用于在 SQL 查询结果上进行遍历和访问。通过游标,你可以逐行获取查询结果,并对每行数据进行处理。
import pymysql # 这里必须指定传递的参数,因为方法中有*,如果单独出现星号 *,则星号 * 后的参数必须用关键字传入 db = pymysql.connect(localhost=, user=, password=,database=) # 生成游标,可以理解为用一个指针指着位置 cursor = db.cursor() # 定义sql语句 query = "select * from table;" # 执行sql语句 cursor.excute(query) # 获取数据 data = cursor.fetchall() # 打印结果 for i in data: print(data) # 关闭连接 db.close() # 也可以定义变量 query = "SELECT * FROM users WHERE username = %s" # 定义变量 username = 'John' # 执行查询 cursor.execute(query, (username,))
ORM(Object-Relational mapping,对象关系映射)是一种编程技术,用于将对象模型映射到关系数据库模型。它提供了一种方便的方式来操作和管理数据库,同时减少了开发人员在关系数据库和对象之间进行转换的工作量。ORM 的主要思想是通过定义对象和关系数据库之间的映射关系,自动将数据从数据库中读取并映射为对象,或者将对象持久化到数据库中。通过 ORM,开发人员可以使用面向对象的方式操作数据,而无需直接编写 SQL 语句。
通俗的讲:将数据库中的一个表映射成类,字段映射成属性。
ORM缺点进行解决的一些方法:使用继承父亲类,将一些公共的属性写在一个类中,或者可以使用一些自动化的工具,去自动生成映射类。
flask使用ORM:先下载sql操作模块——flask_sqlalchemy
在config/settings
中设置连接参数
# 该参数使用了pymysql库,使用之前先安装pymysql
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://数据库用户名:密码@数据库地址:端口/连接数据库名称"
一般orm对象放在项目的models
包中
# 该包的__init__.py内容如下
# 导入flask_sqlalchemy包
from flask_sqlalchemy import SQLAlchemy
# 创建SQLAlchemy实例
db = SQLAlchemy()
# 初始化SQLAlchemy实例,并将其与app.py文件挂钩
def init_app_db(app):
db.init_app(app)
# 导入包下面的模块,保证与项目挂钩
from . import
创建orm映射对象
# 导入db
from . import db
# 创建orm映射对象
# SQLAlchemy().Model对象中有对数据库基本操作属性和方法,通过继承该类,使其称为orm映射对象
class OrmClassName(db.Model):
__tablename = "数据库中的表名"
# orm映射类的字段可以和数据库中的属性名相同,也可以不相同,不相同的时候应当传入该字段映射的数据库属性
property = db.Column(["name"], db.Type, [CONSTRAINTS])
SQLAlchemy中Type对应数据库类型举例:
SQLAlchemy类型 数据库类型 String char, varchar Integer 各种int类型 DateTime datetime TIMESTAMP timestamp Enum enum Float float Decimal decimal/numeric
- 时间输入可以使用datetime模块
SQLAlchemy中的数据库约束
- nullable
- default
- primary_key
- atuoincrement
- unique
一般中间表或只用定义表结构的schema
推荐使用db.Table()
创建实例的方式
table_name = db.Table("表名", db.Column("字段名", 类型,约束))
当需要处理外键的时候需要使用到SQLAlchemy提供的relationship
属性
在引入外键的类中的属性如下写法:
字段名 = db.Column(db.ForeignKey("表名.字段名"))
在被引入外键的类中如下写法:
# 新添加一个表中没有的属性
# 定义了secondary参数指定中间表就可以让另一个主要表不使用relationship字段
属性 = db.relationship("外键名称", secondary=指定中间表, backref="设置反向引用属性")
migrate是flask提供的项目数据库版本迭代管理工具(用于开发环境),通过使用这个工具就可以将代码中的orm映射类添加到数据库中。总的来说,就是可以通过更改orm映射类,从而更改项目数据库。
# 安装
pip install flask_migrate
# 导入
import flask_migrate import Migrate
# 使用Migrate连接数据库
migrate = Migrate(newApp, db)
migrate是在命令行中使用
# init:初始化,同步migrate和数据库的版本
# migrate:类似git的add命令,上传但未提交
# upgrade:提交版本更新
flask --app server:newApp db init
flask --app server:newApp db upgrade
flask --app server:newApp db migrate -m comment
在完成映射之后,就可以使用该映射类对数据库中的表进行操作。
orm查询都是用的orm类的query带的属性方法
flask的查询有三个:get()
, filter_by()
, filter()
其中get()
用于主键查询返回单个值
OrmClassName.query.get(primary_key)
db.session.query(OrmClassName).get(primary_key)
filter_by()
适用于简单的条件查询
OrmClassName.query.filter_by(conditions).all()/.first()
filter()
用于复杂的条件查询
# 这个查询不仅支持单个条件查询,还支持多个条件查询
# 该方法的默认条件拼接为and,要使用or需进行导入和指明
result = ProductInfo.query.filter(or_(ProductInfo.product_name.like(f'%{my_key}%'),
ProductInfo.product_address.like(f'%{my_key}%'))).all()
# # 查询product_id > 5 的所有记录
# ProductInfo.query.filter(ProductInfo.product_id > 5).all()
# # 查询记录添加时间在1个小时以前的所有记录
# ProductInfo.query.filter(ProductInfo.add_time < (datetime.datetime.now() - datetime.timedelta(hours=1))).all()
# # 查询种类为fruits的前十条记录
# ProductInfo.query.filter(ProductInfo.product_kind == "fruit").limit(10).all()
# # 查询产地为hunan的第一条记录
# ProductInfo.query.filter(ProductInfo.product_address == "hunan").first()
# # 查询价格大于10的所有记录,并且倒叙排序
# ProductInfo.query.filter(ProductInfo.product_price > 10).order_by(ProductInfo.product_id.desc()).all()
limit() count() get() paginate()
offset()
order_by()
group_by()
增删改都是建立在查询操作之上的,前提都是拿到对应对象,然后修改对应对象的字段值,或者删除,只是最后要进行提交
# 增加/修改改动
db.add()
# 删除
db.delete()
# 提交改动
db.session()
flask提供了命令行的上下文操作环境,该操作环境可以让你验证你的代码可行性。当你的代码在该上下文环境中测试成功之后,再写入项目中,可以减少代码的bug。
# 进入上下文环境
# 进入上文环境之后就可以对代码进行测试
flask --app 项目入口文件:项目中app实例名称 shell
现在通用的数据交换是json,程序的返回格式也应该是json(是一种轻量级的数据交换格式,它以易于理解和生成的方式存储和传输数据。JSON 数据格式是基于 JavaScript 的子集,但已成为许多编程语言中通用的数据格式,包括 Python、Java、C++ 等。)格式,一般返回三个部分code
,message
,data
{
"code":
"message":
"data":[
]
}
一般在libs
公用功能模块中编写:
创建response.py
模块:
def generate_response(data=None, message="success", code=0):
# 约定返回的数据格式
if data is None:
data = []
return {
"code": code,
"message": message,
"data": data
}
约定了标准的api接口返回,但是数据库的查询结果有时候是对象,在传递的时候无法进行传输,这就要用到数据库序列化,flask中也有自带的数据库序列化包,不过自己写更方便。
常用的思路就是让对象实现dict()要求的方法转换成一个字典,这样就可以自动转换为json。
# 要让dict()将对象转换成字典,必须让该类实现keys和__getitem__方法
def keys(self):
return 转换的字段名
def __getitem__(self, item):
return getattr(self, item)
restful是一种接口的编写风格,它将资源视为一种对象,使用http的请求方法实现状态转移。
post --> 添加, get --> 查询, delete–> 删除, put --> 修改
flask中也有restful相关的模块–flask_restful
# 安装 pip install flask_restful # flask的restful路由由Api对象进行管理 # 将某一路由蓝图添加到Api对象上 api = Api(Blueprint) # 创建Restful风格接口对象 # 类内部创建相应请求方法的类方法 class ClassName(Resource): def get(self): def post(self): def put(self): def delete(self): # 使用Api的add_resource将具体路由和资源绑定 api.add_resource(ClassName, "/routerName", [endpoint=]) # 还可以使用动态路由传递一些参数 api.add_resource(ClassName, "/routerName[/<DynamicRouter>]", [endpoint=])
后端编写了api接口之后,应该规定能够访问的用户,要对访问的用户进行限制。
当然flask也提供了相关的模块,但是自己写更方便。
想不修改对象的方法然后给方法添加新的功能使用装饰器
首先在数据库中创建相应的权限和用户表,创建关系中间biao
from . import db app_permission = db.Table("app_permission", db.Column("api_id"), db.ForeignKey("api_token.id"), db.Column("permission_id", db.ForeignKey("api_permission.id"))) class ApiToken(db.Model): __tablename__ = "api_token" id = db.Column(db.Integer, primary_key=True, autoincrement=True) appid = db.Column(db.String(128), nullable=False) secretkey = db.Column(db.String(128), nullable=False) manage = db.relationship("ApiPermission", secondary=app_permission, backref="token") class ApiPermission(db.Model): __tablename__ = "api_permission" id = db.Column(db.Integet, primary_key=True, autoincrement=True) url = db.Column(db.String(128), nullable=False) method_type = db.Column(db.String(128), nullalbel=False)
编写一个认证功能模块放在libs/auth
中
from models.user import ApiToken # 实现装饰器 def auth_required(func): def inner(*args, **kwargs): if api_auth(): return func(*args, **kwargs) else: return "认证失败" def api_auth(): params = request.args appid = params.get("appid") salt = params.get("salt") sign = params.get("sign") timestamp = params.get("timestamp") if time.time - float(timestamp) > 600: return False api_token = ApiToken.query.filter_by(appid=appid).first() if not api_token: return False if not has_permission(api_token, request.path, request.metho.lower()) return false secretkey = api_token.secretkey user_sign = appid + salt + secretkey + timestamp m1 = md5() m1.update(user_sign.encode(encoding="utf-8")) if sign != m1.hexdigest(): return False else: return True def has_permission(api_token, url, method): # 客户端请求的方法和url mypermission = method + url # 获取此api_token对象的所有url权限 all_permission = [permission.method_type + permission.url for permission in api_token.manage] if mypermission in all_permission: return True else: return False
session-cookie认证由于session会消耗服务器资源等缺点被舍弃,所以使用token认证代替,token保存在客户端,客户端每次请求都带上token从而进行认证。
token认证使用模块:jwt
import jwt def create_token(uid): # 生成token # 在settings中设置了过期时间 expires_in = current_app.config.get("EXPIRES_IN") # 定义token生成参数 payload = {"uid": uid, "exp":time.time() + expires_in} # 获取token的生成密钥 key = current_app.config["SECRET_KEY"] # 生成token token = jwt.encode(payload, key) return token # 验证token def token_auth(): token = request.headers.get("token") if token: try: jwt.decode(token, current_app.config.get("SECRET_KEY"), algorithms=["HS256"]) except InvalidSignatureError as e: print("token不合法", e) # return False raise TokenFailException except ExpiredSignatureError as e: print("token过期", e) # return False raise TokenFailException else: return true return false
前面的代码中所有的数据验证都是直接写在代码中使用if-else进行验证测试,但是这种参数的验证可以单独写成一个模块。
模块包:wtforms
wtforms是python的一个表单验证包,可以用来验证表单数据是否合法
常用属性:
DataRequired
:验证字段是否为空。示例:validators=[DataRequired()]
Length
:验证字段的长度是否在指定的范围内。示例:validators=[Length(min=2, max=50)]
EqualTo
:验证字段的值是否等于其他字段的值。示例:validators=[EqualTo('password', message='密码不一致')]
Email
:验证字段是否为有效的电子邮件地址。示例:validators=[Email()]
Regexp
:使用正则表达式验证字段的值是否符合指定的模式。示例:validators=[Regexp(r'^[A-Za-z0-9_-]*$', message='只允许包含字母、数字、下划线和破折号')]
NumberRange
:验证字段的值是否在指定的数值范围内。示例:validators=[NumberRange(min=0, max=100)]
URL
:验证字段是否为有效的 URL 地址。示例:validators=[URL()]
from wtforms import Form, StringField, validators # 创建表单类 # 通过继承表单基类Form来创建 class UserForm(Form): # 定义字段,并定义验证规则 username = StringField(validators=[DataRequired()]) password = StringField(validators=[DataRequired(), Regexp(r'\w{6,18}', message="密码不符合要求")]) # 还可以自定义验证方法,方法名称必须定义为:validate_验证字段名 def validate_username(self, value): if User.query.filter_by(username=value.data).first(): return false # 还可以直接重写Form中的validate方法添加自定以规则 class LoginForm(Form): username = StringField(validators=[DataRequired()]) password = StringField(validators=[DataRequired(), Regexp(r'\w{6,18}', message="密码不符合要求")]) def validate(self): super().validate() if self.errors: return False user = User.query.filter_by(username=self.username.data).first() if user and check_password_hash(user.password, self.password.data): return user else: raise ValidationError("验证失败!")
在整个项目中异常处理是一个十分必要的工作,但是前期开发过程不建议进行异常处理。需要通过异常信息来判断代码错误。
""" @Author: Thousand @file: error_code.py @time: 2023/8/2 16:44 """ from werkzeug.exceptions import HTTPException class APIException(HTTPException): code = 500 # http状态码 message = "fail!" # 状态描述信息 status_code = 9999 # 程序状态 def __init__(self, message=None, code=None, status_code=None): if message: self.message = message if code: self.code = code if status_code: self.status_code = status_code super(APIException, self).__init__() def get_body(self, environ=None, scope=None) -> str: body = dict( message=self.message, code=self.status_code ) import json content = json.dumps(body) return content def get_headers(self, environ=None, scope=None, ): return [('content-Type', 'application/json')] # 自定义异常类 class APIAuthorizedException(APIException): message = "API授权认证失败" status_code = 10002 code = 401 class FormValidateException(APIException): message = "表单验证失败" status_code = 10003 class TokenFailException(APIException): message = "token不合法,验证失败" status_code = 10005 code = 401 class AuthorizedFailException(APIException): message = "认证失败" status_code = 10002 code = 401
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。