一、wtforms
pip3 install wtforms 0 要使用邮箱认证，还需要装一个模块：pip3 install email-validator 1 等同于django中的forms，做数据验证，模板的渲染 2 使用
1 基本使用
# wtforms from flask import Flask, render_template, request, redirect from wtforms import Form from wtforms.fields import simple from wtforms import validators from wtforms import widgets app = Flask(__name__, template_folder='templates') app.debug = True class LoginForm(Form): # 字段（内部包含正则表达式） name = simple.StringField( label='用户名', validators=[ validators.DataRequired(message='用户名不能为空.'), validators.Length(min=6, max=18, message='用户名长度必须大于6且小于18') ], widget=widgets.TextInput(), # 页面上显示的插件 render_kw={'class': 'form-control'} ) # 字段（内部包含正则表达式） pwd = simple.PasswordField( label='密码', validators=[ validators.DataRequired(message='密码不能为空.'), validators.Length(min=8, message='用户名长度必须大于%(min)d'), validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*d)(?=.*[$@$!%*?&])[A-Za-zd$@$!%*?&]{8,}", message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符') ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} ) @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'GET': form = LoginForm() return render_template('login.html', form=form) else: form = LoginForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证，提交的值为：', form.data) else: print(form.errors) return render_template('login.html', form=form) if __name__ == '__main__': app.run()
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <h1>登录</h1> <form method="post"> <p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p> <p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p> <input type="submit" value="提交"> </form> </body> </html>
from flask import Flask, render_template, request, redirect from wtforms import Form from wtforms.fields import core from wtforms.fields import html5 from wtforms.fields import simple from wtforms import validators from wtforms import widgets app = Flask(__name__, template_folder='templates') app.debug = True class RegisterForm(Form): name = simple.StringField( label='用户名', validators=[ validators.DataRequired() ], widget=widgets.TextInput(), render_kw={'class': 'form-control'}, default='xxx' ) pwd = simple.PasswordField( label='密码', validators=[ validators.DataRequired(message='密码不能为空.') ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} ) pwd_confirm = simple.PasswordField( label='重复密码', validators=[ validators.DataRequired(message='重复密码不能为空.'), validators.EqualTo('pwd', message="两次密码输入不一致") ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} ) email = html5.EmailField( label='邮箱', validators=[ validators.DataRequired(message='邮箱不能为空.'), validators.Email(message='邮箱格式错误'), ], widget=widgets.TextInput(input_type='email'), render_kw={'class': 'form-control'} ) gender = core.RadioField( label='性别', choices=( (1, '男'), (2, '女'), ), coerce=int # “1” “2” ) city = core.SelectField( label='城市', choices=( ('bj', '北京'), ('sh', '上海'), ) ) hobby = core.SelectMultipleField( label='爱好', choices=( (1, '篮球'), (2, '足球'), ), coerce=int ) favor = core.SelectMultipleField( label='喜好', choices=( (1, '篮球'), (2, '足球'), ), widget=widgets.ListWidget(prefix_label=False), option_widget=widgets.CheckboxInput(), coerce=int, default=[1, 2] ) def __init__(self, *args, **kwargs): super(RegisterForm, self).__init__(*args, **kwargs) self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球')) def validate_pwd_confirm(self, field): """ 自定义pwd_confirm字段规则，例：与pwd字段是否一致 :param field: :return: """ # 最开始初始化时，self.data中已经有所有的值 if field.data != self.data['pwd']: # raise validators.ValidationError("密码不一致") # 继续后续验证 raise validators.StopValidation("密码不一致") # 不再继续后续验证 @app.route('/register', methods=['GET', 'POST']) def register(): if request.method == 'GET': form = RegisterForm(data={'gender': 2,'hobby':[1,2]}) # initial return render_template('register.html', form=form) else: form = RegisterForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证，提交的值为：', form.data) else: print(form.errors) return render_template('register.html', form=form) if __name__ == '__main__': app.run()
<html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <h1>用户注册</h1> <form method="post" novalidate style="padding:0 50px"> {% for field in form %} <p>{{field.label}}: {{field}} {{field.errors[0] }}</p> {% endfor %} <input type="submit" value="提交"> </form> </body> </html>
二、信号
0 pip3 install blinker 1 让开发者可是在请求过程中定制一些用户行为 2 内置信号 内置信号： request_started = _signals.signal('request-started') # 请求到来前执行 request_finished = _signals.signal('request-finished') # 请求结束后执行 before_render_template = _signals.signal('before-render-template') # 模板渲染前执行 template_rendered = _signals.signal('template-rendered') # 模板渲染后执行 got_request_exception = _signals.signal('got-request-exception') # 请求执行出现异常时执行 request_tearing_down = _signals.signal('request-tearing-down') # 请求执行完毕后自动执行（无论成功与否） appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行（无论成功与否） appcontext_pushed = _signals.signal('appcontext-pushed') # 应用上下文push时执行 appcontext_popped = _signals.signal('appcontext-popped') # 应用上下文pop时执行 message_flashed = _signals.signal('message-flashed') # 调用flask在其中添加数据时，自动触发 3 信号可以干什么 1 记录日志（某个表新增了数据，记录日志，某个表删除了数据。。。） 2 更新缓存（假设给article表数据加了缓存，一旦该表插入数据，一旦触发article表新增数据的信号，删除缓存，重新生成缓存（同步的话速度稍微慢一些，结合celery，异步生成缓存）） 4 内置信号的使用 -第一步：定义要给函数 -第二步：跟信号绑定 信号.connect(函数内存地址) -第三步：触发信号（内置信号自动触发）信号.send(self) 5 自定义信号 -第0步：定义一个信号 -第一步：定义要给函数 -第二步：跟信号绑定 信号.connect(函数内存地址) -第三步：触发信号（内置信号自动触发）信号.send(self)
1. 内置信号
# 信号 from flask import Flask, signals,render_template app = Flask(__name__) # 内置信号使用 # 内置信号request_started = _signals.signal('request-started') # 请求到来前执行 # 给它绑定一个函数执行 # 第一步：写好函数 def before_request_signal(*args,**kwargs): # print('请求来了，触发了这个信号') print(args) print(kwargs) print('模板渲染了，触发了这个信号') # 第二步：跟信号绑定 signals.request_started.connect(before_request_signal) # signals.request_finished.connect(before_request_signal) # 模板渲染后 # signals.template_rendered.connect(before_request_signal) # 第三步：触发信号（不需要手动触发，请求走到这，就会触发） @app.route('/') def hello_world(): # return 'hello world' return render_template('index.html') if __name__ == '__main__': app.run() app.wsgi_app()
2 自定义信号
# 自定义信号 from flask import Flask, signals, render_template from flask.signals import _signals app = Flask(__name__) # 第0步：定义信号（定义了一个叫lqz的信号） lqz = _signals.signal('lqz') # 第一步：写好函数 def before_request_signal(*args, **kwargs): print(args) print(kwargs) print('触发了自定义的lqz信号') # 第二步：跟自定义的信号绑定 lqz.connect(before_request_signal) # 第三步：触发信号（自定义信号，要自己触发） @app.route('/') def hello_world(): res = render_template('index.html') # 在模板渲染之后，触发自定义的lqz信号 lqz.send('sss',name='lqz') return res if __name__ == '__main__': app.run()
三、多app应用(了解)
# 多app应用，在一个flask项目中，可以有多个app对象 # werkzeug：0.16.1 及之前版本 from werkzeug.wsgi import DispatcherMiddleware from werkzeug.serving import run_simple from flask import Flask, current_app app1 = Flask('app01') app2 = Flask('app02') @app1.route('/index') def index(): return "app01" @app2.route('/index2') def index2(): return "app2" dm = DispatcherMiddleware(app1, { '/pro': app2, }) if __name__ == "__main__": run_simple('localhost', 5000, dm) # 当请求来了：dm()--->DispatcherMiddleware的__call__
四、SQLAlchemy
1 orm框架，django-orm 2 第三方的orm框架，可以独立于web项目使用 3 pip3 install SQLAlchemy 4 组成部分 Engine，框架的引擎 Connection Pooling ，数据库连接池 Dialect，选择连接数据库的DB API种类 Schema/Types，架构和类型 SQL Exprression Language，SQL表达式语言 5 SQLAlchemy不能创建数据库，创建表，创建字段
# 执行原生sql ## 执行原生sql import time import threading import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.engine.base import Engine engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/book?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=2, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) def task(arg): conn = engine.raw_connection() cursor = conn.cursor() cursor.execute( "select * from app01_author" ) result = cursor.fetchall() print(result) cursor.close() conn.close() for i in range(20): t = threading.Thread(target=task, args=(i,)) t.start()
# 数据新增 ## orm的 单表操作 from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import Users,Person,Hobby engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) Connection = sessionmaker(bind=engine) # 每次执行数据库操作时，都需要创建一个Connection,是从连接池中拿链接 con = Connection() # ############# 执行ORM操作 ############# # 创建出对象，通过conn.add(对象)，把对象增加进去 # obj1 = Users(name="egon") # con.add(obj1) # 一对多的关系，新增数据 # hobby=Hobby(caption='橄榄球') # hobby2=Hobby(caption='乒乓球') # con.add(hobby) # person=Person(name='lqz',hobby_id=1) # con.add(person) # con.add_all([hobby,hobby2]) hobby=Hobby(caption='美女') person=Person(name='egon',hobby=hobby) # person=Person(name='egon',hobby_id=3) con.add_all([hobby,person]) # 提交事务 con.commit() # 关闭session，其实是将连接放回连接池 con.close()
# models.py # 通过类，创建表 import datetime from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index Base = declarative_base() class Users(Base): __tablename__ = 'users' # 数据库表名称 id = Column(Integer, primary_key=True) # id 主键 name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空 # email = Column(String(32), unique=True) # #datetime.datetime.now不能加括号，加了括号，以后永远是当前时间 # ctime = Column(DateTime, default=datetime.datetime.now) # extra = Column(Text, nullable=True) __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一 # Index('ix_id_name', 'name', 'email'), #索引 ) # 一对多，一个爱好，可以有多个人喜欢，关联字段写在多的一方，写在Person中 class Hobby(Base): __tablename__ = 'hobby' id = Column(Integer, primary_key=True) caption = Column(String(50), default='篮球') class Person(Base): __tablename__ = 'person' nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) # hobby指的是tablename而不是类名，uselist=False # 外键关联 # hobby_id跟hobby表的id字段关联 hobby_id = Column(Integer, ForeignKey("hobby.id")) # 跟数据库无关，不会新增字段，只用于快速链表操作 # 类名，backref用于反向查询 hobby = relationship('Hobby', backref='pers') # 把表同步到数据库 def init_db(): """ 根据类创建数据库表 :return: """ engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) # 把所有别Base管理的表都创建出来 Base.metadata.create_all(engine) def drop_db(): """ 根据类创建数据库表 :return: """ engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) # 把所有别Base管理的表都删除 Base.metadata.drop_all(engine) if __name__ == '__main__': init_db() # drop_db()