flask养吾剑总结

启动

from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Welcome to My Watchlist!'
......
app.run(debug=True)

注意，@app装饰器必须在app.run(debug=True)之前执行，这就表示路由定义函数必须在这个启动文件中，如果在其他位置，显然会发生循环引用问题。因为从定义上，flask区别于django，是为小型应用服务的。但实际上，它也有模块化的可能，如果需要布置模块化的大型应用，需要使用蓝图 blueprint，我们后面再学习。

路由

@app.route('/post/<int:post_id>', methods=['GET', 'POST'])
def show_post(post_id):
# show the post with the given id, the id is an integer
return 'Post %d' % post_id

1、可以使用path变量，变量可以自动转型。
类型如下：

methods=['GET', 'POST']

除了这种方法外，还可以使用类似django的url/path方法来指定路由，只是不能转发而已。

 app.add_url_rule('/login', 'login', route.login, methods=['GET', 'POST'])
app.add_url_rule('/logout', 'logout', route.logout)
app.add_url_rule('/index', 'index', methods=['GET', 'POST'])

第二个参数指定的是url的名称，方便之后通过逆路由方法进行url寻址。

逆路由

根据一个url映射的名称反转得到url的网址，个人称其为逆路由。逆路由由url_for函数实现。
需要注意的是，url_for可以接受任意个关键字参数，每个关键字参数对应 URL 中的变量：

@app.route('/user/<username>')
def profile(username):
return '{}'s profile'.format(escape(username))
url_for('profile', username='John Doe')

url_for还可以用来处理静态文件：

url_for('static', filename='style.css')

注意一定要加上filename参数。

模板

模板渲染使用render_template函数：

from flask import render_template
@app.route('/hello/')
@app.route('/hello/<name>')
def hello(name=None):
return render_template('hello.html', name=name)

模板函数也可以接收多个参数，每个参数都会被注入模板context之中。使用{{}} 表达式可以找到。
flask使用的是jinjia2模板，语法大体如下：

<title>Hello from Flask</title>
{% if std %}
<h1>Hello {{ std.name }}!</h1>
{% else %}
<h1>Hello, World!</h1>
{% endif %}
//循环字典
{% for item in dict_name %}
{{item}}={{dict_name[item]}}
{% endfor %}

注意 .表达式十分强大，类似el表达式，它不仅能够查找字段属性，还能查找字典属性、方法。当然，方法应该这样写：std.methodname()，甚至可以传值：std.methodname(i)

模板中如果要直接输出html语言，都会涉及到自动转义问题：

>>> from flask import Markup
>>> Markup('<strong>Hello %s!</strong>') % '<blink>hacker</blink>'
Markup(u'<strong>Hello &lt;blink&gt;hacker&lt;/blink&gt;!</strong>')
>>> Markup.escape('<blink>hacker</blink>')
Markup(u'&lt;blink&gt;hacker&lt;/blink&gt;')
>>> Markup('<em>Marked up</em> &raquo; HTML').striptags()
u'Marked up xbb HTML'

在python中，可以通过markup进行安全输出。markup.escape进行转义输出。unescape将转义的标记转换回文本字符串。
在模板中，请使用使用 |safe 过滤器过滤器。

request数据接收，重定向，404，公共上下文

和django一样，flask也通过request进行数据接收。所不同的是，request是一个线程全局变量，在各个地方都可以引用，而不只是视图函数中。事实上，这样做是比较方便的，笔者在django中也手工手工进行布置过，很简单，拦截器在每次请求时都讲request对象绑定到线程即可。

from flask import request,abort, redirect, url_for
request.form.get('key', '')//处理表单post数据，多组使用.getlist
request.args.get('key', '')//处理geturl数据
request.files['the_file']//处理文件流
request.cookies.get('username')//处理cookies
redirect(url_for('login'))//重定向
abort(401)//抛出一个401错误，使当前视图函数返回已注册的错误页面

错误界面的注册如下：

from flask import render_template
@app.errorhandler(404)
def page_not_found(error):
return render_template('page_not_found.html'), 404

对于多个模板内都需要使用的变量，我们可以使用 app.context_processor 装饰器注册一个模板上下文处理函数，如下所示：
app.py：模板上下文处理函数

@app.context_processor
def inject_user():
# 函数名可以随意修改
user = User.query.first()
return dict(user=user)
# 需要返回字典，等同于return {'user': user}

关于response

如果返回的是一个元组，那么元组中的项目可以提供额外的信息。元组中必须至少 包含一个项目，且项目应当由

1. 如果视图返回的是一个响应对象，那么就直接返回它。

2. 如果返回的是一个字符串，那么根据这个字符串和缺省参数生成一个用于返回的 响应对象。

3. 如果返回的是一个字典，那么调用 jsonify 创建一个响应对象。

4. 如果返回的是一个元组，那么元组中的项目可以提供额外的信息。元组中必须至少 包含一个项目，且项目应当由 (response, status) 、 (response, headers) 或者 (response, status, headers) 组成。 status 的值会重载状态代码， headers 是一个由额外头部值组成的列表 或字典。

5. 如果以上都不是，那么 Flask 会假定返回值是一个有效的 WSGI 应用并把它转换为 一个响应对象。

以上总结
1、返回字符串直接打印在前端。
2、返回字典自动转成json对象。当然，也可以手动转化：

jsonify([user.name for user in users])

3、返回元祖必须遵从固定格式。

可以使用 make_response() 包裹返回表达式，获得响应对象，并对该对象 进行修改，然后再返回:

@app.errorhandler(404)
def not_found(error):
resp = make_response(render_template('error.html'), 404)
resp.headers['X-Something'] = 'A value'
return resp

消息闪现flash和日志

和django一样，在且只在下一个请求中访问上一个请求结束时记录的消息。

flash(u'Invalid password provided', 'error')
——————————————————————
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
<ul class=flashes>
{% for category, message in messages %}
<li class="{{ category }}">{{ message }}</li>
{% endfor %}
</ul>
{% endif %}
{% endwith %}
——————————————————————
//消息可以在模板中先筛选一遍
{% with errors = get_flashed_messages(category_filter=["error"]) %}

日志：

app.logger.debug('A value for debugging')
app.logger.warning('A warning occurred (%d apples)', 42)
app.logger.error('An error occurred')

蓝图blueprint

simple_page = Blueprint('simple_page', __name__,
template_folder='templates')
@simple_page.route('/', defaults={'page': 'index'})
@simple_page.route('/<page>')
def show(page):
pass
————————————————————
app.register_blueprint(simple_page)

使用蓝图时，先用蓝图注册函数，然后再去app页面注册蓝图即可。注册成功后，可以使用bfname.methodname作为视图名称，使用url_for进行逆路由。比如上面的函数：simple_page.show

注册蓝图时，可以给其加上前缀：app.register_blueprint(simple_page, url_prefix=‘/pages’)
这就实现了模块化分层url开发。

表单

wtf表单和django使用模式基本一致

class OutageForm(FlaskForm):
class Meta:
csrf = False
country = SelectField("Country ", choices={})
plant = StringField("Plant")
c1 = BooleanField("定検情報有")
c2 = BooleanField('My Project',validators=[
DataRequired(message= u'邮箱不能为空'), Length(1, 64),)
c3 = BooleanField('Mechanical', default=True)
c4 = BooleanField('Electrical', default=True)
search = SubmitField("Search")
date_start = SelectField("Period", choices=[(str(i), str(i)) for i in range(2019, 2041)],
default=lambda: str(datetime.now().year - 1))
date_end = SelectField("", choices=[(str(i), str(i)) for i in range(2019, 2041)],
default=lambda: str(datetime.now().year + 4))
#自定义字段级验证，验证失败后的错误信息自动加入到form.errors和form.filed.errors
def validate_date_start(self, field):
if field.data > self.date_end.data:
raise ValidationError('')

字段和验证方法都比较一致。
详细的自定义验证方法见官方文档：
官方文档
也可以在表单外定义验证函数或工厂函数，然后在validators写函数名称(不能直接调用)即可。
表单数据取值如下：
模板中：

{{ form.country.label(class="control-label") }}
{{ form.country(class="form-control"，size="20") }}
{% for message in form.username.errors %}
<small class="error">{{ message }}</small><br>
{% endfor %}

显然，可以通过()来自定义写一些自己需要的html标签属性,也可以回显错误信息。
视图中：

form = OutageForm()
#"POST" == request.method and form.validate()可以更换为form.validate_on_submit()。它内部使用的return self.is_submitted() and self.validate()显然也是相同的算法。
if "POST" == request.method and form.validate():
country_cd = form.country.data
else:
errors=form.errors

需要注意的是，flask不向django一样，需要显式绑定request，因为flask的request是线程全局的，所以form初始化的过程中可以自动去寻找reqeust中的数据进行绑定。

表单对象的可用属性：

验证前：
form.country.choices 表单select的选项，如果想要动态设置，可以在表单传给模板之前这样设置。
验证后:
form.password.data 表单value值。
form.data 得到一个所有字段的字典，格式如下：
{'turbine_id': '', 'teiken_id': '', 'country': '30567', 'plant': '', 'c1': False, 'c2': False, 'c3': True, 'c4': True, 'search': True, 'date_start': '2019', 'date_end': '2024'}
form.errors 得到一个表单所有错误的字典，格式如下：
{'date_start': ['must bigger than date_end!']}
当然，也可以单独取该字段的errors
form.date_start.errors 得到的是list
['must bigger than date_end!']

基本字段

注意：RadioField和SelectField在不设置default的情况下，空值时为字符串“None”，而StringField的默认值为空字符串""，此时无论是做空值判断，还是保存到数据库，都需要进行进一步处理。

验证器：

• DataRequired/data_required:验证数据是否真实存在，即不能为空，必须是非空白字符串，否则触发StopValidation错误。
• InputRequired/input_required：和DataRequired的区别在于可以是空白字符串；
• Required/required:data_required的别名 Email/email：验证符合邮件的格式，只有最基本的验证；
• EqualTo/equal_to:比较两个字段的值，比如密码和确认密码，如果不相等就触发错误，equal_to(field,message),需要输入另一个字段的名字。
• IPAddress/ip_address:验证是否是ip地址，默认验证IPV4地址。
  MacAddress/mac_address:验证是否符合mac格式； UUID：是否是uuid格式；
• URL/url:验证是否符合url格式； Regexp/regexp:用提供的正则表达式验证字段；Regexp(r"")
• Length/length:设置字段值的长度，Length(min,max);
• NumberRange/number_range:设置一个数字字段的取值范围，可以针对浮点数和小数；NumberRange(min,max)
• Optional/optional:允许字段为空并停止验证；
  NoneOf/none_of:将传入的数据和无效的比较，是抛出异常Noneof(values).
• Anyof/any_of:将传入的数据和预设的数据比较，不是异常。Anyof(values)

也可以安装bootstrap3，使用它进行快速渲染：

 {% import "bootstrap/wtf.html" as wtf %}
{{ wtf.quick_form(form) }}

sqlalchemy

基本配置

db = SQLAlchemy()
app = Flask(__name__)
db.init_app(app)
#——————————————以下是config中的配置
# SQLAlchemy
SQLALCHEMY_DATABASE_URI = 'oracle+cx_oracle://{user}:{password}@{host}'.format(**{
'user': os.getenv('DB_USER', 'svcdb'),
'password': os.getenv('DB_PASSWORD', 'svcadmin'),
'host': os.getenv('DB_HOST', 'AWSXE'),
})
# 设置每次请求结束后会自动提交数据库的改动
SQLALCHEMY_COMMIT_ON_TEARDOWN = True
SQLALCHEMY_TRACK_MODIFICATIONS = True
# 打印sql语句
SQLALCHEMY_ECHO = True

定义模型

class OutageSchedule(db.Model):
__tablename__ = 'OUTAGE_SCHEDULE'
turbine_id = db.Column(db.String())
id_seq = Sequence('OBJECT_ID_SEQUENCE')
# oracle 序列作为主键自增
teiken_id = db.Column(db.Integer, id_seq,
server_default=id_seq.next_value(), primary_key=True)
description = db.Column(db.String())
outage_start = db.Column(db.Date)
outage_end = db.Column(db.Date)
outage_duration = db.Column(db.Integer)
outage_type_t = db.Column(db.String())
outage_type_g = db.Column(db.String())
execution = db.Column(db.Integer)
pr_date_m = db.Column(db.DateTime)
pr_date_e = db.Column(db.DateTime)
representive_id = db.Column(db.String())
representive_name = db.Column(db.String())
created_at = db.Column(db.DateTime, default=datetime.now)
created_by = db.Column(db.String(), default=lambda: current_user.get_id())
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
updated_by = db.Column(db.String(), default=lambda: current_user.get_id())
delete_flg = db.Column(db.Integer(), default=0)
deleted_at = db.Column(db.DateTime)
deleted_by = db.Column(db.String)

常用的sqlalchemy字段类型：

常用的sqlalchemy列表项

常用的sqlalchemy关系选项

表间关系实例

class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64))
# relationship 可以定义在任意一端，框架会自动查询自己有没有外键连接对方，或者对方有无外键连接自己。这里实际上是省略了foreign_keys=[User.role_id], 也可以使用字符串，字符串情况下如果只有一个外键关联，也可以不用数组。
users = db.relationship('User', backref='role')
# association_proxy 可以创造一个关联关系的代理，代理不仅可以用于查询，也可以增加删除
user_names= association_proxy('users ', 'name ')
# declared_attr将类方法映射为类属性，__mapper_args__属性配置传递给sqlalchemy.orm.Mapper的参数，比如默认排序
@declared_attr
def __mapper_args__(cls):
return {"order_by": cls.id.asc()}
def __repr__(self):
return '<Role %r>' % self.name
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True, index=True)
email = db.Column(db.String(64), unique=True)
pswd = db.Column(db.String(64))
# 相当于定义role_id到roles.id的外键。但数据库不存在改外键也可以关联
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
plant_id = db.Column(db.Integer())
unit_id = db.Column(db.Integer())
# 也可以不用ForeignKey，而是通过primaryjoin指定连接条件
# lazy表示如何加载关系，默认值为 select，使用时单条查询懒加载，joined，eagerly加载，查询时使用外连接一起获取。selectin，也是急切加载，额外使用一条select in 子查询获取，noload 不加载，只写属性。raise 访问时为null立即报错，通过其他手段进行赋值。
# 如果不是项目里所有的关联对象都需要急切加载，可以默认设置懒加载，然后在需要查询时使用.options(contains_eager(Unit.plant))进行调整
unit = db.relationship(Unit, primaryjoin=and_(
cast(foreign(plant_id), db.String) == Unit.plant_id,
foreign(unit_id) == Unit.unit_id), lazy='selectin', backref='ttil_targets')
def __repr__(self):
return 'User:%s' % self.name
###################
#options: joinedload,lazyload,load_only,defer
Unit.query.join(Plant).join(Customer).options(contains_eager(Unit.plant).contains_eager(Plant.customer),contains_eager(Unit.ttil_targets)).all()

定义基类和maxin

sqlalchemy.orm.declarative_base
sqlalchemy.ext.declarative.AbstractConcreteBase

class Base(AbstractConcreteBase, db.Model):
pass
########
class Base:
pass
Base = declarative_base(cls=Base)

sqlalchemy.orm.declarative_mixin

@declarative_mixin
class MyMixin:
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()

三态变化

sqlalchemy.orm.make_transient() 瞬时态
sqlalchemy.orm.make_transient_to_detached() 瞬时转离线，将从缓存重置所有属性。
sqlalchemy.orm.util.was_deleted() 查询是否已被删除

删增改查

# add
form = OutageEditForm()
form.create_duration()
outage_model = OutageSchedule(turbine_id=form.turbine_id.data or None,
description=form.description.data or None,
outage_start=form.outage_start.data or None,
outage_end=form.outage_end.data or None,
outage_type_g=form.outage_type_g.data or None,
outage_type_t=form.outage_type_t.data or None,
execution=form.execution.data or None,
outage_duration=form.outage_duration.data or None
)
db.session.add(outage_model)
db.session.commit()
# delete
db.session.delete(User)
db.session.commit()
# update
outage = OutageSchedule.query.filter_by(teiken_id=teiken_id).first()
outage.outage_start = form.outage_start.data or None
outage.outage_end = form.outage_end.data or None
outage.outage_type_t = form.outage_type_t.data or None
outage.outage_type_g = form.outage_type_g.data or None
outage.execution = form.execution.data or None
outage.description = form.description.data or None
outage.outage_duration = form.outage_duration.data or None
outage.updated_by = current_user.get_id()
db.session.add(outage)
db.session.commit()
# 类sql
db.session.query(PartType2.parttype1_id, PartType2.parttype2_id, PartType3.parttype3_id).select_from(
PartType2).join(PartType3).filter(PartType3.parttype3_id == self.parttype3_id).first()
db.session.query(PartType1.p1_name + "->" + PartType2.p2_name + "->" + PartType3.p3_name).select_from(
PartType1).join(PartType2).join(PartType3).filter(PartType3.parttype3_id == self.parttype3_id).scalar()
# subquery
query = query.filter(TTIL.parttype3_id == form.type3.data)
subquery = db.session.query(PartType3.parttype3_id).join(PartType2).join(PartType1)
if form.type2.data:
subquery = subquery.filter(PartType2.parttype2_id == form.type2.data).subquery()
else:
subquery = subquery.filter(PartType1.parttype1_id == form.type1.data).subquery()
ttil_list= query.filter(exists().where(TTIL.parttype3_id == subquery.c.parttype3_id)).all()
####
subquery = db.session.query(db.func.count()).filter(
Soshiki.dept_cd == DeptMaster.department_cd).label("regist_flg")
result = db.session.query(DeptMaster.department_cd, DeptMaster.department, subquery).filter(
DeptMaster.department_cd.like(dept_cd + "%"),
DeptMaster.department.like("%{}%".format(dept_name))
).all()

原生sql查询

result = db.session.execute(text(sqlstr))
# 返回值是一个sqlalchemy.engine.result.ResultProxy对象
# 该对象方法如下
>>>result.fetchall()
[('XCH05', '30788', 'China', 30788, 'J')]
>>>result.fetchone()
('XCH05', '30788', 'China', 30788, 'J')
# 注意，这里的fetchall，fetchone返回的都不是简单的元祖数组，而是一个sqlalchemy.engine.result.RowProxy对象。该对象可以看做一个字典，也可以看做一个object。实验如下：
>>>ss=turbine.next()
>>>ss.items()
[('plant_cd', 'XCH05'), ('plant_name', '30788'), ('country_nm', 'China'), ('turbine_id', 30788), ('data_type', 'J')]
>>>	ss.__class__
<class 'sqlalchemy.engine.result.RowProxy'>
>>>ss.plant_cd
'XCH05'
>>>ss['plant_cd']
'XCH05'
#显然，该对象也可以使用字典生成器来生成
>>>{k: v for k, v in ss.items()}
{'plant_cd': 'XCH05', 'plant_name': '30788', 'country_nm': 'China', 'turbine_id': 30788, 'data_type': 'J'}

需要注意的是，ResultProxy内部有一个游标，相当于一个生成器对象，它是不可以重复遍历的。

调用函数

connection = db.engine.raw_connection()
cursor = connection.cursor()
result = cursor.callfunc('pkg_turbine_db_util.signin_to_turbinedb', str, (user_id_in, session_id_in))
cursor.close()
connection.close()

…未完待续

最后

以上就是阳光银耳汤最近收集整理的关于flask养吾剑总结的全部内容，更多相关flask养吾剑总结内容请搜索靠谱客的其他文章。