Flaskz介绍6--使用手册
项目开发部署过程中，经常用到的知识点和业务场景集合，持续添加中……
项目初始化
- 创建venv环境
python -m venv venv-name # venv-name是制定的venv目录，一般使用venv - 激活&退出到当前venv环境
source venv/bin/activate # 激活 deactivate # 退出 - 依赖包文件
pip install -r requirements.txt # 安装文件中列出的依赖包 pip uninstall -y -r requirements.txt # 卸载已安装依赖包，没有-y会有是否卸载提示 pip freeze > requirements.txt # 将已安装依赖包输出到文件 - 其它常用pip命令
pip list # 列出已安装的包 pip list -o # 查看可升级的包 pip show flaskz # 查看安装包信息 pip install flaskz # 安装 pip install 'flaskz[ext]' # 安装&指定额外的扩展依赖项 pip uninstall flaskz # 卸载 pip install --upgrade flaskz # 升级 pip install --upgrade pip # 升级pip pip install --upgrade --force-reinstall flaskz # 强制重新安装，依赖项也会重新安装 pip install --upgrade --no-deps --force-reinstall flaskz # 强制重新安装，--no-deps避免重新安装依赖项 pip install --no-deps flaskz # 不安装包依赖项 - Flask命令
export FLASK_APP=admin_app.py # 设置Flask环境变量 flask run --host=0.0.0.0 --port=666 # 启动Flask应用 ps aux | grep 'python.*flask run' # 查看Flask进程 pkill -f 'python.*flask run' # 停止Flask应用
Alembic
- 创建Alembic迁移环境
alembic init mitrations # mitrations是存放迁移文件的目录 - 初始化Alembic
alembic revision -m "migration init" # 创建基础空迁移 alembic upgrade head # 初始化数据库alembic表 - 生成数据库版本文件
alembic revision --autogenerate -m "add template" # "add template"是指定的版本名称 - 更新数据库表结构
alembic upgrade head # head表示最新版本 - 数据库降级操作
alembic downgrade 5656baaceae2 # 5656baaceae2是指定要降到的版本 - 跳过指定版本
alembic stamp 5656baaceae2 - 查看版本信息
alembic current # 显示当前版本 alembic history # 查看版本历史记录 - 将数据库更新sql语句输出到文件(可用于多个数据库的更新)
alembic upgrade head --sql >[migration.sql] - 可自动监测到的数据模型变化
模式元素 更改 表 添加、删除 列 添加、删除、允许空值(nullable) 索引 索引的基本变化、显示命名的唯一性约束、支持自动生成索引和唯一性约束 键 基本的重命名 - 无法自动监测到的数据模型变化
模式元素 更改 表 重命名 列 重命名、类型改变(可开启)、长度变化(可开启) 约束 无明确名称的约束 类型 ENUM这类数据库后段不直接支持的类型 以上变化，需要手动创建迁移或修改默认迁移配置才能检测到
- 开启列类型变化/长度变化自动检测
# env.py def run_migrations_online(): # ... context.configure( connection=connection, target_metadata=target_metadata, compare_type=True # 开启列类型变化/列长度变化检测 )
数据库
- 切换/配置数据库
修改 config.py/FLASKY_DATABASE_URI 和 alembic.ini/sqlalchemy.url 相关配置，对于某些数据库需要安装驱动依赖
- sqlite数据库，默认情况下，级联外键删除不起作用，可通过以下代码启用
# https://docs.sqlalchemy.org/en/14/dialects/sqlite.html#foreign-key-support from sqlalchemy.engine import Engine from sqlalchemy import event @event.listens_for(Engine, "connect") def set_sqlite_pragma(dbapi_connection, connection_record): cursor = dbapi_connection.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close()
数据模型
- 如果数据库表的列名和模型列名称不一致时，如何处理
在模型列的info属性中添加field属性，示例如下
system_default = Column('default', Boolean, default=False, info={'field': 'system_default'}) - 指定模糊查询的列
可以在模型类定义中添加 like_columns 列表，模糊查询时只会查询列表中的列，示例如下
class TemplateModel(ModelBase, ModelMixin): name = Column(String(32), unique=True, nullable=False) description = Column(String(255)) #... like_columns = ['name', description] # 列表中可以是列名，也可以是列对象 - 禁止列被用户编辑
在模型类定义中添加 auto_columns 列表，编辑/添加时，列表中的列数据会被过滤掉，示例如下
class TemplateModel(ModelBase, ModelMixin): id = Column(Integer, primary_key=True, autoincrement=True) # primary key name = Column(String(32), unique=True, nullable=False) updated_at = Column(DateTime(), default=datetime.now, onupdate=datetime.now) auto_columns = ['id', updated_at] # 列表中可以是列名，也可以是列对象 - 如何减少关系级联查询的次数
对于确认需要的数据，可以将关系定义中的lazy参数设置为"join"，这样会通过连接查询的方式，将数据一次查询出来，示例如下
from_device = relationship('Device', uselist=False, primaryjoin='and_(Link.from_device_id == Device.id)', lazy='joined') - 如何禁止relationship更新关联数据
relationship默认会对关联数据进行增删操作，但有些relationship只是引用而不是父子关系，请参考一下实例
class ACLListEntry(ModelBase, ModelMixin): # ... static_route_id = Column(Integer, ForeignKey('route_statics.id')) # 静态路由ID static_route = relationship('StaticRoute', uselist=False, lazy='joined') # 这里的relationship只是引用StaticRoute，但是不应该对static_route有任何的增删改操作如果不想对关联数据进行任何的更新操作，可以将relationship中的cascade设置为None或""
static_route = relationship('StaticRoute', uselist=False, lazy='joined', cascade=None) # 设置cascade为None通过cascade设置，如果ACLListEntry的json中有static_route对象，也不会更新StaticRoute。
- 如何指定relationship的关联关系
SQLAlchemy会根据ForeignKey进行relationship的自动关联，但是如果一个模型类中，有指向同一个模型类的多个外键时，必须在relationship中指定通过哪个外键进行关联
class Link(ModelBase, ModelMixin): __tablename__ = 'tp_links' id = Column(Integer, primary_key=True, autoincrement=True) from_device_id = Column(Integer, ForeignKey('device_devices.id', ondelete='CASCADE'), nullable=False) # Link的from_device_id是Device.id的外键 to_device_id = Column(Integer, ForeignKey('device_devices.id', ondelete='CASCADE'), nullable=False) # Link的to_device_id是Device.id的外键 #... from_device = relationship('Device', uselist=False, primaryjoin='and_(Link.from_device_id == Device.id)', lazy='joined') # 通过from_device_id外键与Device进行关联 to_device = relationship('Device', uselist=False, primaryjoin='and_(Link.to_device_id == Device.id)', lazy='joined') # 通过to_device_id外键与Device进行关联 - 如何一次查询多个模型的所有数据
请参考 query_multiple_model 函数
- 如何连接多个数据库
请参考 多数据库
- 如何禁止Alembic对指定的数据库表更新
一般用于从第三方数据库查询数据的场景，第三方数据库的表结构是不需要我们维护更新的，实现代码如下
# https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.environment.EnvironmentContext.configure.params.include_object # alembic/evn.py/include_object函数 IGNORE_TABLES = [] def include_object(object, name, type_, reflected, compare_to): if type_ == 'table' and (name in IGNORE_TABLES or object.info.get("skip_autogenerate", False)): # 过滤表 return False elif type_ == "column" and object.info.get("skip_autogenerate", False): # 过滤列 return False return True # alembic/evn.py/run_migrations_online函数 context.configure( ... include_object=include_object # 添加配置 ) # 模型类定义 class LSNode(MDBBase, ModelMixin): __tablename__ = 'ls_nodes' __table_args__ = {'info': {'skip_autogenerate': True}} # 不会更新ls_nodes数据库表结构 hash_id = Column(String(255), nullable=False, primary_key=True) - ★数据模型类如何被Alembic自动检测
因为ModelBase被抽取到了flaskz中，导致Alembic不能通过flaskz.models.ModelBase检测到模型类的增删及修改，有两种处理方法
- 将flaskz.models.ModelBase导入到某个项目文件中(例如/app/modules/__init__.py)，
然后在/migrations/venv.py中导入项目文件中的ModelBase，示例如下
# app/modules/__init__.py from flaskz.models import ModelBase # 避免IDE自动删除未引用导入 if ModelBase: pass # migrations/venv.py from app.modules import ModelBase # 从app.modules导入ModelBase，而不是从flaskz.models导入 target_metadata = ModelBase.metadata #... - 将模型类都import到/migrations/venv.py中，示例如下
from app.sys_mgmt import model # 模型类所在的py文件 from app.modules import template #... from flaskz.models import ModelBase target_metadata = ModelBase.metadata #...
方法1相当于全局设置，比方法2简洁，可以避免因没有import导致的不自动更新问题。
而方法2相当于定制化，比方法1更灵活，可以减少类似于禁止Alembic对指定的数据库表更新的工作量(不import即可)。
可以根据具体场景选择对应的方案。
- 将flaskz.models.ModelBase导入到某个项目文件中(例如/app/modules/__init__.py)，
然后在/migrations/venv.py中导入项目文件中的ModelBase，示例如下
API封装
- 如何添加API
请参考 自定义API
- 如何指定init_model_rest_blueprint函数生成的模型类API
可以通过routers参数制定要创建的API列表，请参考 创建数据模型API
- 如何指定API只能被登录用户访问
将路由处理函数通过rest_login_required函数装饰器封装即可，请参考 权限控制
- 如何指定API只能被拥有模块权限和操作权限的用户访问
将路由处理函数通过rest_permission_required函数装饰器封装即可，请参考 权限控制
- 如何记录操作日志
请参考 API权限控制和操作日志
- 如何自定义权限控制和日志
请参考 API权限控制和操作日志
- 如何实现PUT模式的全量更新功能
init_model_rest_blueprint函数创建的更新API使用的是PATCH模式(局部更新)， 如果想要实现PUT模式的更新(全量更新)，现在可以通过补全json数据的方法来实现。(此功能暂时不考虑添加到flaskz中)
- 如何让服务支持跨域访问
- 安装flask-cors
pip install flask-cors - 通过CORS函数为app应用添加跨域功能
from flask_cors import CORS app = Flask(__name__) CORS(app)
- 安装flask-cors
Linux操作
- 服务进程查看
ps -ef | grep flask # 查看flask进程 ps -ef | grep gunicorn # 查看gunicorn进程 - 设置环境变量
export FLASK_APP=admin_app.py - 启动Flask测试服务器
nohup flask run --host=0.0.0.0 --port=666 & - 使用配置文件启动Gunicornweb服务(Windows不支持)
- 修改 /doc/server/gunicorn_config.py 配置信息
- 将gunicorn_config.py文件copy到应用根目录(admin-app.py所在目录)
- 通过 gunicorn -c gunicorn_config.py admin-app:app 命令启动gunicorn服务
- 设置网络代理
export proxy="http://10.20.30.60:8000" export http_proxy=proxy # 设置http代理 export https_proxy=proxy # 设置https代理 unset http_proxy # 取消http代理设置 unset https_proxy # 取消https代理设置
Issues & Solutions
以下为开发过程中遇到的各种问题及解决方法。
- flask做为静态文件服务器时，Chrome浏览器ERR_INVALID_HTTP_RESPONSE异常
- 如果是通过flask命令启动的服务，添加--without-threads启动参数
flask run --host=0.0.0.0 --port=666 --without-threads - 如果是通过python命令行启动的服务，添加threaded=False参数
if __name__ == '__main__': app.run(host='0.0.0.0', port=666, debug=True, threaded=False)
具体原因待查...猜测可能跟Chrome的多线程资源访问有关
- 如果是通过flask命令启动的服务，添加--without-threads启动参数
- 如果数据库的登录账号/密码包含特殊字符时(例如@符号)，需要先转义再使用
例如MySQL的账号/密码为:root/admin@123， 如果设置url = 'mysql+pymysql://root:admin@123@10.10.10.10:3306/flaskz-admin'会导致连接失败。
FLASKZ_DATABASE_URI = 'mysql+pymysql://root:admin%40123@10.10.10.10:3306/flaskz-admin'(-SQLAlchemy)
sqlalchemy.url = mysql+pymysql://root:admin%%40123@10.10.10.10:3306/flaskz-admin(-alembic/ini)
- SQLAlchemy删除数据时SAWarning: DELETE statement on table expected to delete 1 row(s); 0 were matched异常
设置Model的confirm_deleted_rows : False
class DeviceExtPrefixAddressSid(ModelBase, ModelMixin): __tablename__ = 'device_device_ext_prefix_address_sids' __mapper_args__ = { 'confirm_deleted_rows': False } id = Column(Integer, primary_key=True, autoincrement=True) //... - flask中，通过request.json获取请求中的json数据时，引发BadRequest异常 ( "Did not attempt to load JSON data because the request Content-Type was not 'application/json'.")
异常原因: 请求的Content-Type没有设置为application/json或application/*+json引起的
可以通过request.get_json(force=True, silent=True)方法强制解析JSON数据，如果没有对应的数据，返回None。
- 自定义正则匹配路由(常用于api转发)
class RegexConverter(PathConverter): # Werkzeug<2.2.3，BaseConverter/PathConverter都支持层级匹配(url中包含/)，但>=2.2.3必须只有PathConverter支持层级匹配 def __init__(self, url_map, regex): super(RegexConverter, self).__init__(url_map) self.regex = regex app.url_map.converters['regex'] = RegexConverter #注册converter @api_bp.route('/<regex(".*"):path>', methods=HTTP_METHODS) #api_bp所有的路由回调函数都是remote def remote(path): return forward_request(base_url + path) - 通过forward_request函数转发api请求，并直接返回前端时，如果请求的api返回的response的header中有Transfer-Encoding，可能导致前端ajax请求一直处于pending状态
具体原因待查...，用旧版本的pip包运行没问题，新版本才有这个问题，猜测可能是升级的pip包中，逻辑发生了变化。
api_bp = Blueprint('api', __name__) base_url = 'http://10.124.206.140/' HTTP_METHODS = ['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH'] @api_bp.route('/<regex(".*"):path>', methods=HTTP_METHODS) def remote(path): remote_res = forward_request(base_url+request.full_path) res = make_response(remote_res[0], remote_res[1]) for k, v in remote_res[2]: if k not in ['Transfer-Encoding']: res.headers[k] = v return res - 解决itsdangerous 2.0+版本废弃jws的问题
(Deprecated since version 2.0: ItsDangerous will no longer support JWS in version 2.1)
如果想在后续版本继续使用jws功能的话，可以使用flaskz.auth中的类代替(flaskz version>=0.9)。
from itsdangerous import TimedJSONWebSignatureSerializer # 2.1之前版本 from flaskz.auth import TimedJSONWebSignatureSerializer # 2.1及之后版本 - SQLite线程异常，ProgrammingError: SQLite objects created in a thread can only be used in that same thread
可以通过添加check_same_thread=False解决
FLASKZ_DATABASE_URI = 'sqlite:///./_sqlite/flaskz-admin.db?check_same_thread=False' - 为已有数据模型类(table)添加列外键或唯一性约束，alembic更新数据库时ValueError异常，ValueError: Constraint must have a name
有如下两个解决方案:
- [推荐]手动修改alembic生成的version文件，为约束设置name
batch_op.create_unique_constraint(None, ['new_name']) batch_op.create_foreign_key(None, 'sys_users', ['user_id'], ['id']) # 改为 batch_op.create_unique_constraint('new_name', ['new_name']) batch_op.create_foreign_key('user_id', 'sys_users', ['user_id'], ['id']) - 在添加列外键或唯一性约束时指定约束名称的name
class TemplateModel(ModelBase, ModelMixin): __tablename__ = 'templates' __table_args__ = ( UniqueConstraint('new_name', name="new_name"), # 添加唯一性约束名字 ) id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32), nullable=False, unique=True) new_name = Column(String(32)) # 请注意此处没有unique=True，否则生成的版本文件中会有两个new_name的唯一性约束 # ... user_id = Column(Integer, ForeignKey('sys_users.id', name='user_id')) # 指定外键的name
以上异常&解决方案只针对修改已经存在的表，对于新建的模型类(table)没有此问题。
- [推荐]手动修改alembic生成的version文件，为约束设置name
- SQLite更新NotImplementedError异常，NotImplementedError: No support for ALTER of constraints in SQLite dialect.
Please refer to the batch mode feature which allows for SQLite migrations using a copy-and-move strategy.
可以在alembic的env.py配置文件中添加render_as_batch=True参数
def run_migrations_online(): # ... with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, include_object=include_object, render_as_batch=True # 添加 ) # ...在生成的version文件中会添加batch相关的代码逻辑
def upgrade(): with op.batch_alter_table('templates', schema=None) as batch_op: # 如果不设置render_as_batch，这行是没有的 batch_op.create_unique_constraint('new_name', ['new_name']) batch_op.create_foreign_key('user_id', 'sys_users', ['user_id'], ['id']) - MySQL中不能设置Text类型的列为unique，否则会引发异常(MySQL error: key specification without a key length)
可以添加一个hash列作为unique列，然后通过hashlib生成对应的hash字符串
# 数据模型类，添加hash列 class AModel(ModelBase, ModelMixin): a_col = Column(Text(), nullable=False) # Text列不能作为unique列 a_col_hash = Column(String(255), nullable=False, unique=True) # 添加hash列作为unique列 # ... # 使用hashlib生成hash值 AModel.add({ 'a_col': a_col_value, 'a_col_hash': hashlib.sha256(a_col_value.encode('utf-8')).hexdigest() # 生成hash字符串 # ... }) - 版本<2.2.3的Flask生成的以'/'结尾的路由，如果有参数并设置path类型类型转换, Flask不会将结尾不带/的请求自动重定向到带/的路由(例如:@api_bp.route('abc/<path:did>/')
有如下三种解决方法
- 升级到>=2.2.3版本(推荐)
- 路由参数不使用path类型转换(如果参数中带'/'，需要转换成'%2F')
- 使用结尾带'/'的URL
- Windows和MacOS系统，Python3.8及以后版本使用multiprocessing.Process多进程，可能会引发RuntimeError异常(RuntimeError: An attempt has been made to start a new process...)
引发原因可能是3.8版本对于进程的启动方式做了更改(spawn/fork)
- 如果是单独启动服务，在各个OS平台都建议使用flask命令的方式启动，可以避免这个问题，但请注意，此种方式可能不共享主进程的内存变量，需要单独初始化
- 如果是通过gunicorn启动服务，可以正常运行，而且子进程可以共享主进程的内存变量
为了代码统一，建议在进程中初始化用到的系统变量或配置
- 在多线程中不能访问flask.current_app/flask.g，会引发RuntimeError异常(RuntimeError: Working outside of application context)
Flask使用线程局部变量(thread-local variables)来处理请求，这意味着current_app只在创建它的请求上下文中有效，在新线程的上下文中并不存在，所以在其他线程中尝试访问current_app，便会引发上述Error
如果要在多线程中使用current/g的话，可以将app/current_app作为线程target函数的参数传递到函数中，在函数中通过app.app_context()创建一个应用上下文来使用current_app/g
def create_app(config_name): app = Flask(__name__) # ... thread = Thread(target=thread_function,args=(app,)) # app作为函数的参数 thread.start() return app @sys_mgmt_bp.route('/auth/login/', methods=['POST']) def sys_auth_login(): """用户登录Session/Cookie""" app = current_app._get_current_object() # 获取当前应用对象的引用 thread = Thread(target=thread_function, args=(app,)) # current_app作为函数的参数 def thread_function(app): # app作为函数的参数 with app.app_context(): # 创建应用上下文 print(current_app) # 使用current_app - pss查询，如何传递属性值为空字符串("" / " ")的查询条件
为了更好的适配前端请求，系统默认会移除值为空字符串的查询条件 ex)"module": ""，可以通过以下两个方法实现空值查询
- 通过对象+操作符的方式传递空值查询参数 ex)"action": {"==": ""}
{ "search": { "module": "",//不起作用 "action": { "==": "" //起作用，查询action==""的数据 } } } - 设置ignore_null/_ignore_null参数，适用所有属性，如果有属性不需要空值查询，请不要传递查询参数
{ "search": { "ignore_null": false,//查询module=="" AND action==""的数据 "module": "", "action": { "==": "" } } }
- 通过对象+操作符的方式传递空值查询参数 ex)"action": {"==": ""}
- 高并发设置
- 修改数据库(MySQL)的最大连接数
- 临时 - SET GLOBAL max_connections = 10000
- 永久 - [mysqld] max_connections = 10000(my.cnf/my.ini)
- 设置应用数据库连接参数，FLASKZ_DATABASE_ENGINE_KWARGS = {'pool_timeout': 30, 'pool_size': 200, "poolclass": QueuePool, 'max_overflow': 1000}(config.py)
- 设置类Unix操作系统可以打开文件的最大数量(HTTPS请求/数据库连接)， ulimit -n 10000
以上参数根据实际情况(硬件/并发数)等调整
- 修改数据库(MySQL)的最大连接数