![[记录]基于Flask Web全栈开发实战（黄勇·著）](https://cssjc.com/zb_users/upload/2024/03/20240310135009171004980994772.jpg "[记录]基于Flask Web全栈开发实战（黄勇·著）")
Flask 学习
flask 项目配置
Debug模式 Host、Port配置
python">from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
在pycharm中，设置debug模式和port端口号
# 在非pycharm中，设置debug模式
if __name__=='__main__':
app.run(debug=True)
# 在非pycharm中，设置host和post
if __name__=='__main__':
app.run(debug=True,host='0.0.0.0',port=9000)
# port必须是整数，不能设置为字符串
在app.config中添加配置
from flask import Flask
app = Flask(__name__)
# cofig 配置项，配置项的名称必须大写
app.config['SECRET_KEY']='sknrek349Lx!@#'
# 使用python配置
# 在文件夹中创建一个config.py文件
# 文件内容的代码如：
TOKEN_KEY = "123456"
加载配置文件
from flask import Flask
import config
app = Flask(__name__)
app.config.from_object(config)
print(app.config["TOKEN_KEY"])
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
URL 和 视图
定义无参URL
@app.route('/profile')
def profile():
return "这是个人中心"
定义有参URL
@app.route('/blog/<blog_id>')
def blog_detail(blog_id):
return f'您查找的博客ID为{blog_id}
指定参数类型
语法：<类型:参数名>
如：
@app.route(‘/blog/int:blog_id’)
def blog_detail(blog_id):
return f’您查找的博客ID为{blog_id}
@app.route('/blog/list/<any(python,flask,django):category>')
def blog_list_with_category(category):
return f'您获取的博客分类为{category}'
# any 备选值中的任一个
# URL中需要传递多个参数，则只要用斜杠（/）分隔开来即可。
@app.route('/blog/list/<int:user_id>/<int:page>')
def blog_list(user_id,page):
return f'您查找的用户是{user_id}，博客分页为{page}'
# url 拼接：http://127.0.0.1:9000/blog/list?user_id=8&page=9
# request 是一个线程隔离的全局对象
@app.route('/blog/list')
def blog_list_query_str():
user_id = request.args.get('user_id')
page=request.args.get("page")
return f'您查找的用户是{user_id},博客分页为{page}'
HTTP请求方法
@app.route("/blog/add",methods=['POST'])
def blog_add():
return "使用POST方法添加博客"
@app.route('/blog/add/post/get',methods=['POST',"GET"])
def blog_add_post_get():
if request.method =='GET':
return "使用GET方法添加博客"
else:
return "使用POST方法添加博客"
页面重定向
from flask import url_for, redirect
@app.route('/login')
def login():
return 'login page'
@app.route('/profile')
def profile():
name = request.args.get('name')
if not name:
# 如果没有name 说明没有登录成功，重定向到登录页面
return redirect("/login")
else:
return name
构造URl
# 构造URL
@app.route('/blog<int:blog_id>')
def blog_detail(blog_id):
return f'您查找的博客ID为{blog_id}'
@app.route('/urlfor')
def get_url_for():
url = url_for("blog_detail",blog_id=2,user='admin')
return url
# 输出结果为：/blog2?user=admin
jinja2模版
模版的基本使用
渲染模版
在视图函数中使用render_template函数渲染index.html模版
from flask import render_template
# 模版的基本使用
@app.route('/index')
def idnex():
return render_template('index.html')
渲染变量
# 渲染变量
@app.route('/variable')
def variable():
hobby = "游戏"
return render_template("variable.html",hobby=hobby)
# <!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>我的兴趣爱好是{{ hobby }}</h1>
</body>
</html>
传递字典类型，person和对象类型user
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>我的兴趣爱好是{{ hobby }}</h1>
<p>person的姓名是{{ person.name }},person的年龄是{{ person.age }}</p>
<p>user的姓名是{{ user.username }},user的邮箱是{{ user.email }}</p>
</body>
</html>
class User:
def __init__(self,username,email):
self.username = username
self.email = email
@app.route('/variable')
def variable():
hobby='游戏'
person = {
"name":"张三",
"age":18
}
user = User("李四","XXXX@qq.***")
return render_template('variable.html',hobby=hobby,person=person,user=user)
-----
# 也可以用这种方式
class User:
def __init__(self,username,email):
self.username = username
self.email = email
@app.route('/variable')
def variable():
hobby='游戏'
person = {
"name":"张三",
"age":18
}
user = User("李四","XXXX@qq.***")
context = {
"hobby":hobby,
"person":person,
'user':user
}
# return render_template('variable.html',hobby=hobby,person=person,user=user)
return render_template('variable.html',**context)
过滤器和测试器
自定义过滤器
import datetime
# 获取当前时间
_now = datetime.datetime.now()
# 自定义时间过滤器
@app.template_filter("time_filter")
def time_filter(time):
if not isinstance(time,datetime.datetime):
return time
_period = (_now-time).total_seconds()
if _period<60:
return '刚刚'
elif 60<= _period< 60*60:
return f'{int(_period/60)}分钟之前'
elif 60*60 <= _period < 86400:
return f'{int(_period/3660)}小时之前'
elif 86400 <= _period < 2592000:
return f'{int(_period / 86400)}天之前'
else:
return time.strftime('%Y-%m-%d %H:%M')
@app.route('/filter')
def time_filter():
timelist=[
'abcd',
_now,
_now-datetime.timedelta(minutes=5),
_now-datetime.timedelta(hours=10),
_now-datetime.timedelta(days=5),
_now-datetime.timedelta(days=150)
]
return render_template('timefilter.html',timelist=timelist)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>智能显示时间</title>
{% for time in timelist %}
<ul>
<li>
<p>{{ time }}</p>
<p>{{ time|time_filter }}</p>
</li>
</ul>
{% endfor %}
</head>
<body>
</body>
</html>
Jinja2 内置过滤器
# 内置过滤器
@app.route('/filter')
def filter():
context={
'username':'Sleet',
'password':123456,
'age':-18,
'sex':'女',
'script':'<script>alert("hello");</script>',
"hobbies":{
'basketball':'great',
'football':"not bad",
'table tennis':'great',
'tennis':'bad'
},
'introduction':'I am a hangsome boy,I like playing basketball.'
}
return render_template('filter.html',**context)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>{{ username }}</p>
<p>{{ password }}</p>
<p>{{ age|abs }}</p>
<p>{{ sex|default('男')}}</p>
{#关闭自动转义#}
{% autoescape off %}
<p>{{ script }}</p>
{% endautoescape %}
<p>{{ script|safe }}</p>
<p>{{ hobbies|first }}</p> <!--获取序列第一个元素-->
{#返回元素的长度#}
<p>{{ hobbies|first|length }}</p>
<p>{{ hobbies|last }}</p>
<p>{{ username|replace('Sleet','Snow') }}</p> <!--替换字符串-->
<!--截取字符串-->
<p>{{ introduction|truncate(length=13,killwords=False) }}</p>
<p>{{ introduction|wordcount }}</p>
<p>{{ "%s and %s"|format('Sleet',"Snow" )}}</p>
<p>{{ hobbies|join(d=' and ') }}</p>
<p>{{ script|striptags }}</p>
</body>
</html>
测试器
控制语句
if判断语句
# if判断语句
@app.route('/if')
def if_statement():
age =18
return render_template('if.html',age=age)
-------------
# 或
# if判断语句
@app.route('/if/<int:age>')
def if_statement(age):
# age =18
return render_template('if.html',age=age)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>if 语句</title>
</head>
<body>
{% if age >18 %}
<div>已成年！</div>
{% elif age < 18 %}
<div>未成年！</div>
{% else %}
<div>刚成年！</div>
{% endif %}
</body>
</html>
for循环语句
# for循环语句
@app.route('/for')
def for_statement():
books=[{
"name":"三国演义",
"author":"罗贯中",
"price":100
},{
"name":"水浒传",
"author":"施耐庵",
"price":99
},{
"name":"红楼梦",
"author":"曹雪芹",
"price":101
},{
"name":"西游记",
"author":"吴承恩",
"price":102
}]
return render_template('for.html',books=books)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<table>
<thead>
<tr>
<th>书名</th>
<th>作者</th>
<th>价格</th>
</tr>
</thead>
<tbody>
{% for book in books %}
<tr>
<th>{{ book.name }}</th>
<th>{{ book.author }}</th>
<th>{{ book.price }}</th>
</tr>
{% else %}
<tr>
<td colspan="3" style="text-align: center">无数据</td>
</tr>
{% endfor %}
</tbody>
</table>
</body>
</html>
模版结构
宏和import语句
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{% macro input(name,value='',type='text') %}
<input type="{{ type }}" value="{{ value|escape }}" name="{{ name }}">
{% endmacro %}
{% macro textarea(name,value='',rows=10,cols=40) %}
<textarea name="{{ name }}" id="" cols="{{ cols }}" rows="{{ rows }}">{{ value|escape }}</textarea>
{% endmacro %}
</body>
</html>
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{% import 'forms.html' as forms %}
<dl>
<dt>Username</dt>
<dd>{{ forms.input('username' )}}</dd>
<dt>Password</dt>
<dd>{{ forms.input('password',type='password') }}</dd>
<p>{{ forms.textares('***ment') }}</p>
</dl>
</body>
</html>
模版继承
# base.html 父类模版
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="stylesheet" href="base.css">
<title>{% block title %}{% endblock %}</title>
{% block head %}{% endblock %}
</head>
<body>
<div id="body">{% block body %}{% endblock %}</div>
<div id="'footer" >
{% block footer %}
©Copyright 2008 by <a href="http://domain.invalid">you</a>
{% endblock %}
</div>
</body>
</html>
# 继承base模版
{% extends "base.html" %}
{% block title %} 首页{% endblock %}
{% block head %}
<style type="text/css">
.detail {
color: red;
}
</style>
{% endblock %}
{% block body %}
<h1>这里是首页</h1>
<p class="detail">首页的内容</p>
{% endblock %}
模版环境
Mysql数据库
flask_SQLAlchemy的基本使用
连接Mysql
# 连接数据库
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
# 配置连接数据库
hostname = '192.168.3.5'
port = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask'
app.config['SQLALCHEMY_DATABASE_URI']=f"mysql+pymysql://{USERNAME}:{PASSWORD}@{hostname}:{port}/{DATABASE}?charset=utf8mb4"
db = SQLAlchemy(app)
with app.app_context():
with db.engine.connect() as conn:
rs = conn.execute(text("select 1"))
print(rs.fetchone())
# 输出：<sqlalchemy.engine.base.Connection object at 0x10b13d910>
ORM模型
# 创建1个User模型
class User(db.Model):
__tablename__ ='user'
id = db.Column(db.Integer,primary_key = True,autoincrement=True)
username = db.Column(db.String(100))
password = db.Column(db.String(100))
# 运行
with app.app_context():
db.create_all()
CRUD操作
Create操作
# app中完整的代码
from flask import Flask
from flask import request
from flask import url_for, redirect
from flask import render_template
# 连接数据库
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
app = Flask(__name__)
# 配置连接数据库
hostname = '192.168.3.5'
port = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'student'
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{hostname}:{port}/{DATABASE}?charset=utf8mb4"
db = SQLAlchemy(app)
class User(db.Model):
__tablename__ ='db_user01'
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
username = db.Column(db.String(100))
password = db.Column(db.String(100))
with app.app_context():
db.create_all()
# CRUD操作
# 1 Create操作
@app.route('/user/add/')
def user_add():
user1 = User(username="张三",password="111111")
user2 = User(username="李四",password="222222")
user3 = User(username="王五",password="333333")
db.session.add(user1)
db.session.add(user2)
db.session.add(user3)
db.session.***mit()
return '用户添加成功！'
if __name__ == '__main__':
app.run()
Read操作
# 2 Read操作
@app.route('/user/fetch/')
def user_fetch():
# 获取User中的所有数据
users = User.query.all()
for i in users:
print(f"用户名为：{i.username},密码为：{i.password}")
# 获取主键为1的User对象
user = User.query.get(1)
print(user.username)
# 获取第1条数据
user = User.query.first()
print(user.username)
return "数据提取成功"
# 过滤
@app.route('/user/filter')
def user_filter():
# filter
users = User.query.filter(User.username == '张三').all()
print(users)
# filter_by
users = User.query.filter_by(username ='张三').all()
print(users)
return '数据过滤成功！'
# order_by()使用方法
@app.route('/user/filter_orderby')
def user_filter_orderby():
# 正序排序
users = User.query.order_by("id")
# users = User.query.order_by(User.id)
for user in users:
print(user.id)
# 倒序排序
users = User.query.order_by(text("-id"))
# users = User.query.order_by(User.id.desc())
for user in users:
print(user.id)
return "数据过滤成功"
# groupby使用
from sqlalchemy import func
# group_by()使用方法
@app.route('/user/filter_groupby/')
def user_filter_groupby():
users = db.session.query(User.username,func.count(User.id)).group_by("username").all()
print(users)
return '数据过滤成功！'
update操作
# update操作
@app.route('/user/update/')
def user_update():
user = User.query.get(1)
user.username = '张三_修改后'
db.session.***mit()
return '更新成功！'
# update操作
@app.route('/user/update/')
def user_update():
user = User.query.get(1)
user.username = '张三_修改后'
db.session.***mit()
return '更新成功！'
# 针对多条的数据更新
@app.route('/user/update_filter/')
def user_update_filter():
User.query.filter(User.username.like("%张三%")).update({"password":User.password+"_被修改的"},synchronize_session=False)
db.session.***mit()
return '数据更新成功！'
delete操作
# delete操作
@app.route('/user/delete/')
def user_delete():
user = User.query.get(1)
db.session.delete(user)
db.session.***mit()
return '数据删除成功！'
# delete操作
@app.route('/user/delete/')
def user_delete():
user = User.query.get(1)
db.session.delete(user)
db.session.***mit()
return '数据删除成功！'
# 删除多条
@app.route('/user/delete_filter/')
def user_delete_filter():
User.query.filter(User.username.contains("张三")).delete(synchronize_session=False)
db.session.***mit()
return "数据删除成功！"
表关系
外键
# 外键
class Article(db.Model):
__tablename__="db_article"
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
title = db.Column(db.String(200),nullable=False)
content = db.Column(db.Text,nullable=False)
author_id = db.Column(db.Integer,db.ForeignKey("db_user01.id"))
with app.app_context():
db.create_all()
一对多关系
建立关系
# 一对多建立关系
# 这是单向关系
class Article_relationship1(db.Model):
__tablename__ = "db_article_relationship1"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey("db_user01.id"))
author = db.relationship("User")
@app.route('/article_relationship1/add')
def article_add_relationship1():
user = User.query.first()
article = Article_relationship1(title="aa",content="bb",author=user)
db.session.add(article)
db.session.***mit()
article = Article_relationship1.query.filter_by(title='aa').first()
print(article.author.username)
return "查询成功"
建立双向关系
# 这是双向关系
# 双向关系
class User_Article(db.Model):
__tablename__ ='db_user_article'
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
username = db.Column(db.String(100))
password = db.Column(db.String(100))
articles = db.relationship("Article_User",back_populates="author")
class Article_User(db.Model):
__tablename__ = "db_article_user"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey("db_user_article.id"))
author = db.relationship("User_Article",back_populates="articles")
with app.app_context():
db.create_all()
@app.route('/article_user/query')
def article_user_query():
user = User_Article.query.first()
for article in user.articles:
print(article.title)
return "双向关联查询成功"
简化关系定义
# 简化关系
class User_Article_jh(db.Model):
__tablename__ ='db_user_article_jh'
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
username = db.Column(db.String(100))
password = db.Column(db.String(100))
class Article_User_jh(db.Model):
__tablename__ = "db_article_user_jh"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey("db_user_article_jh.id"))
author = db.relationship("User_Article_jh",backref="articles")
表单
表单验证
# 创建1个 register.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>注册</title>
</head>
<body>
<form action="{{ url_for('register') }}" method="POST">
<table>
<tr>
<td>用户名：</td>
<td><input type="text" name="username"></td>
</tr>
<tr>
<td>邮箱：</td>
<td><input type="email" name="email"></td>
</tr>
<tr>
<td>密码：</td>
<td><input type="password" name="password"></td>
</tr>
<tr>
<td>确认密码：</td>
<td><input type="password" name="confirm_password"></td>
</tr>
<tr>
<td></td>
<td><input type="submit" value="提交"></td>
</tr>
</table>
</form>
</body>
</html>
# 主程序
from flask import Flask,request,render_template
app = Flask(__name__)
@app.route('/register',methods=['GET','POST'])
def register():
if request.method == 'GET':
return render_template('register.html')
else:
pass
if __name__ == "__main__":
app.run()
表单类编写
# 在根目录下，创建1个forms.py
from wtforms import Form,StringField
from wtforms.validators import length,email,equal_to
class RegisterForm(Form):
username = StringField(validators=[length(min=3,max=20,message="请输入正确长度的用户名！")])
email = StringField(validators=[email(message="请输入正确格式的邮箱")])
password = StringField(validators=[length(min=6,max=20,message="请输入正确的密码！")])
confirm_password = StringField(validators=equal_to("password",message="两次密码不一致！"))
# 其中，wtforms包，pip install wtforms
视图函数中使用表单
from flask import Flask,request,render_template,redirect,flash,url_for
from forms import RegisterForm
# from email_validator import validate_email,EmailNotValidError
app = Flask(__name__)
# 使用flash消息，必须在app三配置Secret_Key 或者，直接通过app.secret_key来设置秘钥。
app.secret_key = 'linql_secret' # 设置一个唯一且保密的密钥
@app.route('/register',methods=['GET','POST'])
def register():
if request.method == 'GET':
return render_template('register.html')
else:
form = RegisterForm(request.form)
# 如果表单通过验证
if form.validate():
email = form.email.data
username = form.username.data
password = form.password.data
# 下面数据可以保存到数据库里
print("email",email)
print("username",username)
print("password",password)
return "注册成功"
else:
for errors in form.errors.values():
for error in errors:
flash(error)
return redirect(url_for("register"))
if __name__ == "__main__":
app.run()
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>注册</title>
</head>
<body>
<form action="{{ url_for('register') }}" method="POST">
<table>
<tr>
<td>用户名：</td>
<td><input type="text" name="username"></td>
</tr>
<tr>
<td>邮箱：</td>
<td><input type="email" name="email"></td>
</tr>
<tr>
<td>密码：</td>
<td><input type="password" name="password"></td>
</tr>
<tr>
<td>确认密码：</td>
<td><input type="password" name="confirm_password"></td>
</tr>
<tr>
<td></td>
<td><input type="submit" value="提交"></td>
</tr>
</table>
<ul>
{% for message in get_flashed_messages() %}
<li>{{ message }}</li>
{% endfor %}
</ul>
</form>
</body>
</html>
自定义验证字段
from wtforms import Form,StringField,ValidationError
registed_email = ['aa@example.***','bb@example.***']
class RegisterForm(Form):
def validate_email(self,field):
email = field.data
if email in registed_email:
raise ValidationError("邮箱已被注册！")
return True
渲染表单
# 视图函数里
from flask import Flask,request,render_template,redirect,flash,url_for
from forms import RegisterForm,LoginForm
@app.route('/login',methods=['GET','POST'])
def login():
# csrf(cross site request forgery,跨站请求伪造)
form = LoginForm(meta={"csrf":False})
if form.validate_on_submit():
email=form.email.data
password = form.password.data
return redirect("/")
return render_template('login.html',form=form)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post">
<table>
<tbody>
<tr>
<td>{{ form.email.label }}</td>
<td>{{ form.email }}</td>
</tr>
{% for error in form.email.errors %}
<tr>
<td></td>
<td>{{ error }}</td>
</tr>
{% endfor %}
<tr>
<td>{{ form.password.label }}</td>
<td>{{ form.password}}</td>
</tr>
{% for error in form.password.errors %}
<tr>
<td></td>
<td>{{ error }}</td>
</tr>
{% endfor %}
<tr>
<td>{{ form.remember.label }}</td>
<td>{{ form.remember() }}</td>
</tr>
<tr>
<td></td>
<td>{{ form.submit }}</td>
</tr>
</tbody>
</table>
</form>
</body>
</html>
CRSF攻击
Flask进阶
类视图
基本使用
from flask.views import View
class ListView(View):
def get_template_name(self):
raise NotImplementedError()
def render_template(self,context):
return render_template(self.get_template_name(),**context)
def dispatch_request(self):
context = {'objects':self.get_objects()}
return self.render_template(context)
class UserView(ListView):
def get_template_name(self):
return 'users.html'
def get_objects(self):
return Users.query.all()
