Flask
1 python界的web框架
-Django：大而全，你要的东西都有 [orm，缓存，认证]
django-ninja
-flask：小而精，所有web开发需要的东西，借助于第三方集成
-web.py
-----同步框架--进程线程架构--3.x以后支持异步--
tornado
sanic
fastapi：高性能小而精，借助于第三方：orm
- Starlette 和 Pydantic
----异步框架--协程架构-----性能高
2 Flask介绍
重要的：Starlette Pydantic
Flask是一个用Python编写的Web应用框架。它由Armin Ronacher开发，他领导着一个名为Pocco的国际Python爱好者团队。Flask基于Werkzeug WSGI工具包和Jinja2模板引擎。两者都是Pocco项目
Flask是一个基于Python开发并且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架，对于Werkzeug本质是Socket服务端，其用于接收http请求并对请求进行预处理，然后触发Flask框架，开发人员基于Flask框架提供的功能对请求进行相应的处理，并返回给用户，如果要返回给用户复杂的内容时，需要借助jinja2模板来实现对模板的处理，即：将模板和数据进行渲染，将渲染后的字符串返回给用户浏览器
“微”(micro) 并不表示你需要把整个 Web 应用塞进单个 Python 文件（虽然确实可以），也不意味着 Flask 在功能上有所欠缺。微框架中的“微”意味着 Flask 旨在保持核心简单而易于扩展。Flask 不会替你做出太多决策——比如使用何种数据库。而那些 Flask 所选择的——比如使用何种模板引擎——则很容易替换。除此之外的一切都由可由你掌握
默认情况下，Flask 不包含数据库抽象层、表单验证，或是其它任何已有多种库可以胜任的功能。然而，Flask 支持用扩展来给应用添加这些功能，如同是 Flask 本身实现的一样。众多的扩展提供了数据库集成、表单验证、上传处理、各种各样的开放认证技术等功能。Flask 也许是“微小”的，但它已准备好在需求繁杂的生产环境中投入使用
Flask depends on the Werkzeug WSGI toolkit, the Jinja template engine, and the Click CLI toolkit
2.1 Werkzeug介绍
Werkzeug是一个WSGI工具包，他可以作为一个Web框架的底层库。这里稍微说一下， werkzeug 不是一个web服务器，也不是一个web框架，而是一个工具包，官方的介绍说是一个 WSGI 工具包，它可以作为一个 Web 框架的底层库，因为它封装好了很多 Web 框架的东西，例如 Request，Response 等等
#1 wsgi :python web 网关接口 协议
#2 具体实现：
wsgiref
uwsgi
gunicorn
#3 Werkzeug 工具包---》遵循wsgi协议---》又做了一些封装
# 4 Werkzeug介绍
Werkzeug is a comprehensive WSGI web application library. It began as a simple collection of various utilities for WSGI applications and has become one of the most advanced WSGI utility libraries.
Werkzeug 是一个综合性 WSGI Web 应用程序库。它最初是 WSGI 应用程序的各种实用程序的简单集合，现已成为最先进的 WSGI 实用程序库之一。
Werkzeug doesn’t enforce any dependencies. It is up to the developer to choose a template engine, database adapter, and even how to handle requests
Werkzeug 不强制执行任何依赖关系。由开发人员选择模板引擎、数据库适配器，甚至如何处理请求
# https://werkzeug.palletsprojects.com/en/3.0.x/
# 5 Werkzeug 快速体验
from werkzeug.wrappers import Request, Response
@Request.application
def hello(request):
return Response('Hello World!')
if __name__ == '__main__':
from werkzeug.serving import run_simple
run_simple('localhost', 4000, hello)
2.2 wsgiref介绍
最简单的Web应用就是先把HTML用文件保存好，用一个现成的HTTP服务器软件，接收用户请求，从文件中读取HTML，返回。
如果要动态生成HTML，就需要把上述步骤自己来实现。不过，接受HTTP请求、解析HTTP请求、发送HTTP响应都是苦力活，如果我们自己来写这些底层代码，还没开始写动态HTML呢，就得花个把月去读HTTP规范。
正确的做法是底层代码由专门的服务器软件实现，我们用Python专注于生成HTML文档。因为我们不希望接触到TCP连接、HTTP原始请求和响应格式，所以，需要一个统一的接口协议来实现这样的服务器软件，让我们专心用Python编写Web业务。这个接口就是WSGI：Web Server Gateway Interface。而wsgiref模块就是python基于wsgi协议开发的服务模块
from wsgiref.simple_server import make_server
def mya(environ, start_response):
print(environ)
start_response('200 OK', [('Content-Type', 'text/html')])
if environ.get('PATH_INFO') == '/index':
with open('index.html','rb') as f:
data=f.read()
elif environ.get('PATH_INFO') == '/login':
with open('login.html', 'rb') as f:
data = f.read()
else:
data=b'<h1>Hello, web!</h1>'
return [data]
if __name__ == '__main__':
myserver = make_server('', 8011, mya)
print('监听8010')
myserver.serve_forever()
2.3 click介绍（django中得命令）
# 1 Click is a Python package for creating beautiful command line interfaces in a composable way with as little code as necessary. It’s the “Command Line Interface Creation Kit”. It’s highly configurable but comes with sensible defaults out of the box
Click 是一个 Python 包，用于以可组合的方式使用尽可能少的代码创建漂亮的命令行界面。它是“命令行界面创建工具包”。它具有高度可配置性，但具有开箱即用的合理默认值
# 2 It aims to make the process of writing command line tools quick and fun while also preventing any frustration caused by the inability to implement an intended CLI API
它的目的是使编写命令行工具的过程变得快速而有趣，同时也防止因无法实现预期的 CLI API 而造成的任何挫败感
# 3 Click in three points:
arbitrary nesting of commands
automatic help page generation
supports lazy loading of subcommands at runtime
Click三点：
命令的任意嵌套
自动生成帮助页面
支持运行时延迟加载子命令
https://click.palletsprojects.com/en/8.1.x/
# pip3 install click
import click
@click.command()
@click.option('--count', default=1, help='Number of greetings.')
@click.option('--name', prompt='Your name',help='The person to greet.')
def hello(count, name):
for x in range(count):
click.echo(f"Hello {name}!")
if __name__ == '__main__':
hello()
# 1 python3 app.py --count=3
# 2 python3 app.py --help
# 3 python3 app.py --count=3 --name=lqz
3 flask创建和运行
3.1 python-dotenv
存入环境变量，直接可以取出，上线会修改
pip install python-dotenv
import os
from dotenv import load_dotenv
from dotenv import dotenv_values
## 1 加载配置文件
# 必须在根路径下新建一个 .env 的文件，并写入配置才能返回True,会把.env下的配置文件设置进环境变量
res=load_dotenv() # take environment variables from .env
print(res)
print(os.environ.get('DOMAIN'))
# You will probably want to add .env to your .gitignore, especially if it contains secrets like a password.
## 2 获取环境变量字典
config = dotenv_values(".env")
print(config)
print(config.get('DOMAIN'))
3.2 watchdog
pip install watchdog
# 当前目录下文件修改会被监控到，打印日志
import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
path = sys.argv[1] if len(sys.argv) > 1 else '.'
event_handler = LoggingEventHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
3.3 创建虚拟环境
Mac/linux
# 创建虚拟环境
mkdir myproject
cd myproject
python3 -m venv .venv
# 激活虚拟环境
. .venv/bin/activate
Win
# 创建虚拟环境
mkdir myproject
cd myproject
py -3 -m venv .venv
# 激活虚拟环境
.venv\Scripts\activate
4 快速入门
4.1 多种方式运行flask
方式一 ： python -m flask --app 文件名 run
方式二 ： flask --app 文件名 run
方式三 ： flask run --host=0.0.0.0 py文件必须叫app.py
方式四 ： 使用pycharm快速运行 配置一个 flaskserver
方式五 ： 直接右键运行，必须加入如下代码
if __name__ == '__main__':
app.run()
from flask import Flask
app = Flask(__name__)
@app.route("/")
def index():
return "ok!heart!"
if __name__ == "__main__":
app.run()
4.2 debug模式
debug模式 ： 自动重启，错误提示
flask --app hello run --debug
5 fastapi快速使用
https://fastapi.tiangolo.com/zh/
from fastapi import FastAPI
from pydantic import BaseModel
app=FastAPI()
class Item(BaseModel):
name: str
age:int
@app.get('/')
async def index():
return {'code':100,'msg':'成功'}
@app.post('/login')
async def login(item:Item):
print(item.name)
print(item.age)
return item
if __name__ == '__main__':
import uvicorn
uvicorn.run(app='myfirstfastapi:app',port=8001)
6 显示用户案例
''' 学到的
1 返回新手四件套
- 返回字符串--》return '字符串'
- 返回模板--》render_template('detail.html',info=user)
- 返回重定向--》redirect('/login')
- 返回json--》jsonify({'code':100,'msg':'成功'})
2 request对象，是全局的--》from flask import request
-request.path
-request.method
.... 有很多--》记django的request---》类似
3 前端传入的：请求体 请求参数
-请求体：request.form--->编码格式是 urlencoded
-请求参数：request.args-->路径后的 /demo01?name=lqz&age=19
4 渲染 模板--》跟django 90%---》它比django更强大--》可以加括号--》字典可以.get []
for {{}}
5 session 使用
- 设置 app.secret_key='adsfa33w9SAD##@sdr434l'
- 设置值：session['username']
- 取值：session.get('username')
6 路由转换器
@app.route('/detail/<int:pk>', methods=['GET'])
'''
6.1 app.py
from flask import Flask, render_template, request, redirect, session,jsonify
app = Flask(__name__, template_folder="templates", static_folder="static")
app.secret_key='adsfa33w9SAD##@sdr434l' #如果使用session，必须加秘钥
# 用户信息
USERS = {
1: {'name': '刘亦菲', 'age': 18, 'gender': '男', 'text': "刘亦菲，1987年8月25日出生于湖北省武汉市，华语影视女演员、歌手，毕业于北京电影学院2002级表演系本科",'img':'https://img2.woyaogexing.com/2021/10/16/e3ccba623848430ba83209c0621a2256!400x400.jpeg'},
2: {'name': '彭于晏', 'age': 28, 'gender': '男', 'text': "彭于晏，1982年3月24日出生于中国台湾省澎湖县，毕业于不列颠哥伦比亚大学，华语影视男演员。。。。。。。。",'img':'https://img2.woyaogexing.com/2021/10/16/e71aa35728c34313bccb4c371192990f!400x400.jpeg'},
3: {'name': '迪丽热巴', 'age': 38, 'gender': '女', 'text': "迪丽热巴（Dilraba），1992年6月3日出生于中国新疆乌鲁木齐市，毕业于上海戏剧学院，中国内地影视女演员",'img':'https://img2.woyaogexing.com/2021/10/30/6a34146dde2d4f1c832463a5be1ed027!400x400.jpeg'},
4: {'name': '亚瑟', 'age': 38, 'gender': '男', 'text': "亚瑟，是腾讯手游《王者荣耀》中一名战士型英雄角色，也是《王者荣耀》的新手英雄之一，既可攻又可守",'img':'https://img2.woyaogexing.com/2021/10/30/371b2aa7a03c4f53b7b1bc86f877d7d1!400x400.jpeg'},
}
@app.route('/')
def index():
# 7 判断用户是否登录，如果登录了，返回模板
if session.get('username'):
# 1 返回模板，渲染到模板中
return render_template('index.html',user_dict=USERS)
# 如果没登录，重定向到login
else:
return redirect('/login')
@app.route('/login', methods=['GET', 'POST'])
def login():
# 2 判断是get请求还是post请求，需要使用request对象
## 2.1 request 正常每次请求都是新的request---》现在flask中全局就一个request--》它会不会乱套？
if request.method == 'GET':
return render_template('login.html')
else:
# 3 取出用户名密码--》请求对象中取--》request
username = request.form.get('username')
password = request.form.get('password')
if username == 'lqz' and password == '123':
# 6 登录成功--》登录信息写入cookie--》使用session写入
session['username'] = username
# 4 登录成功--》重定向到 index
return redirect('/')
else:
# 5 登录失败--》返回login.html 页面，但是页面中渲染错误信息
return render_template('login.html', error='用户名密码错误')
@app.route('/detail/<int:pk>', methods=['GET'])
def detail(pk):
if session.get('username'):
user = USERS.get(pk)
return render_template('detail.html',info=user)
else:
return redirect('/login')
@app.route('/demo01', methods=['GET'])
def demo01():
request.args
return jsonify({'code':100,'msg':'成功'})
if __name__ == '__main__':
app.run()
6.2 index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css">
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.bundle.min.js"></script>
<title>Title</title>
</head>
<body>
<div class="container">
<!-- 头部-->
<div class="sticky-top">
<header class="d-flex flex-wrap justify-content-center py-3 mb-4 border-bottom">
<a href="/" class="d-flex align-items-center mb-3 mb-md-0 me-md-auto text-dark text-decoration-none">
<svg class="bi me-2" width="40" height="32">
<use xlink:href="#bootstrap"></use>
</svg>
<span class="fs-4">交友平台</span>
</a>
<ul class="nav nav-pills">
<li class="nav-item"><a href="#" class="nav-link active" aria-current="page">首页</a></li>
<li class="nav-item"><a href="#" class="nav-link">女生</a></li>
<li class="nav-item"><a href="#" class="nav-link">男生</a></li>
<li class="nav-item"><a href="#" class="nav-link">国产</a></li>
<li class="nav-item"><a href="#" class="nav-link">欧美</a></li>
</ul>
</header>
</div>
<!--轮播图-->
<div>
<div class="bd-example-snippet bd-code-snippet">
<div class="bd-example">
<div id="carouselExampleCaptions" class="carousel slide" data-bs-ride="carousel">
<div class="carousel-indicators">
<button type="button" data-bs-target="#carouselExampleCaptions" data-bs-slide-to="0" class=""
aria-label="Slide 1"></button>
<button type="button" data-bs-target="#carouselExampleCaptions" data-bs-slide-to="1"
aria-label="Slide 2" class="active" aria-current="true"></button>
<button type="button" data-bs-target="#carouselExampleCaptions" data-bs-slide-to="2"
aria-label="Slide 3" class=""></button>
</div>
<div class="carousel-inner">
<div class="carousel-item">
<img src="https://img.zcool.cn/community/[email protected]" alt=""
width="100%" height="300">
<div class="carousel-caption d-none d-md-block">
<h5>激情绿荫</h5>
<p>Some representative placeholder content for the first slide.</p>
</div>
</div>
<div class="carousel-item active">
<img src="https://img2.baidu.com/it/u=2951612437,4135887500&fm=253&fmt=auto&app=138&f=JPEG"
alt="" width="100%" height="300">
<div class="carousel-caption d-none d-md-block">
<h5>品牌雨伞</h5>
<p>Some representative placeholder content for the second slide.</p>
</div>
</div>
<div class="carousel-item">
<img src="https://img1.baidu.com/it/u=1417689082,3333220267&fm=253&fmt=auto&app=138&f=JPEG"
alt="" width="100%" height="300">
<div class="carousel-caption d-none d-md-block">
<h5>家装节</h5>
<p>Some representative placeholder content for the third slide.</p>
</div>
</div>
</div>
<button class="carousel-control-prev" type="button" data-bs-target="#carouselExampleCaptions"
data-bs-slide="prev">
<span class="carousel-control-prev-icon" aria-hidden="true"></span>
<span class="visually-hidden">Previous</span>
</button>
<button class="carousel-control-next" type="button" data-bs-target="#carouselExampleCaptions"
data-bs-slide="next">
<span class="carousel-control-next-icon" aria-hidden="true"></span>
<span class="visually-hidden">Next</span>
</button>
</div>
</div>
</div>
</div>
<!-- 内容-->
<div class="row row-cols-md-2" style="padding: 10px">
{% for k,v in user_dict.items() %}
<div class="card">
<div class="row " style="padding: 10px">
<img src="{{v.get('img')}}" alt="" class="col-md-4">
<div class="col-md-8">
<div class="card-body">
<h5 class="card-title">{{v['name']}}</h5>
<p class="card-text">性别：{{v.gender}}</p>
<p class="card-text">年龄：{{v.age}}</p>
<p class="card-text">
<a href="/detail/{{k}}" class="btn btn-danger">查看详细</a>
</p>
</div>
</div>
</div>
</div>
{%endfor%}
</div>
<!-- table-->
<div class="bd-example" style="margin-top: 30px">
<table class="table table-hover table-striped table-bordered">
<thead>
<tr class="table-danger">
<th colspan="3" class="text-center">更多交友</th>
</tr>
</thead>
<tbody>
<tr class="table-success">
<th>杨幂</th>
<td>女</td>
<td>33</td>
</tr>
<tr class="table-warning">
<th scope="row">刘亦菲</th>
<td>未知</td>
<td>40</td>
</tr>
<tr class="table-success">
<th scope="row">彭于晏</th>
<td>男</td>
<td>23</td>
</tr>
<tr class="table-warning">
<th scope="row">陈奕迅</th>
<td>男</td>
<td>44</td>
</tr>
<tr class="table-success">
<th scope="row">薛之谦</th>
<td>男</td>
<td>36</td>
</tr>
<tr class="table-warning">
<th>李清照</th>
<td>女</td>
<td>未知</td>
</tr>
</tbody>
</table>
</div>
<!--分页-->
<div class="d-flex justify-content-center">
<ul class="pagination pagination-lg">
<li class="page-item">
<a class="page-link" href="#" aria-label="Previous">
<span aria-hidden="true">«</span>
</a>
</li>
<li class="page-item"><a class="page-link" href="#">1</a></li>
<li class="page-item"><a class="page-link" href="#">2</a></li>
<li class="page-item"><a class="page-link" href="#">3</a></li>
<li class="page-item"><a class="page-link" href="#">4</a></li>
<li class="page-item"><a class="page-link" href="#">5</a></li>
<li class="page-item"><a class="page-link" href="#">6</a></li>
<li class="page-item">
<a class="page-link" href="#" aria-label="Next">
<span aria-hidden="true">»</span>
</a>
</li>
</ul>
</div>
<!-- 尾部-->
<div>
<footer class="py-3 my-4">
<ul class="nav justify-content-center border-bottom pb-3 mb-3">
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">首页</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">特性</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">联系我们</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">资料获取</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">关于</a></li>
</ul>
<p class="text-center text-muted">Copyright © 1998 - 2029 liuqingzheng. All Rights Reserved. </p>
</footer>
</div>
</div>
</body>
</html>
6.3 login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css">
<title>登录</title>
</head>
<body>
<div class="container col-xl-10 col-xxl-8 px-4 py-5">
<div class="row align-items-center g-lg-5 py-5">
<div class="col-lg-7 text-center text-lg-start">
<h1 class="display-4 fw-bold lh-1 mb-3">亚洲最大交友平台</h1>
<p class="col-lg-10 fs-4">Bootstrap是Twitter推出的一个用于前端开发的开源工具包。它由Twitter的设计师Mark
Otto和Jacob Thornton合作开发,是一个CSS/HTML框架。目前,Bootstrap最新版本为5.0</p>
</div>
<div class="col-md-10 mx-auto col-lg-5">
<form class="p-4 p-md-5 border rounded-3 bg-light" method="post">
<div class="form-floating mb-3">
<input type="text" class="form-control" id="floatingInput" placeholder="[email protected]" name="username">
<label for="floatingInput">用户名</label>
</div>
<div class="form-floating mb-3">
<input type="password" class="form-control" id="floatingPassword" placeholder="Password" name="password">
<label for="floatingPassword">密码</label>
</div>
<div class="checkbox mb-3">
<label>
<input type="checkbox" value="remember-me"> 记住密码
</label>
</div>
<button class="w-100 btn btn-lg btn-primary" type="submit">登录</button>
<hr class="my-4">
<small class="text-muted">{{error}}</small>
</form>
</div>
</div>
</div>
</body>
</html>
6.4 detai.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css">
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.bundle.min.js"></script>
<title>Title</title>
</head>
<body>
<div class="container">
<div class="sticky-top">
<header class="d-flex flex-wrap justify-content-center py-3 mb-4 border-bottom">
<a href="/" class="d-flex align-items-center mb-3 mb-md-0 me-md-auto text-dark text-decoration-none">
<svg class="bi me-2" width="40" height="32">
<use xlink:href="#bootstrap"></use>
</svg>
<span class="fs-4">交友平台</span>
</a>
<ul class="nav nav-pills">
<li class="nav-item"><a href="#" class="nav-link active" aria-current="page">首页</a></li>
<li class="nav-item"><a href="#" class="nav-link">女生</a></li>
<li class="nav-item"><a href="#" class="nav-link">男生</a></li>
<li class="nav-item"><a href="#" class="nav-link">国产</a></li>
<li class="nav-item"><a href="#" class="nav-link">欧美</a></li>
</ul>
</header>
</div>
<div class="position-relative overflow-hidden p-3 p-md-5 m-md-3 text-center bg-light">
<div class="col-md-5 p-lg-5 mx-auto my-5">
<h1 class="display-4 fw-normal">{{info.name}}</h1>
<img src="{{info.img}}" alt=""
width="300px" height="300px" style="margin: 20px">
<p class="lead fw-normal">{{info.text}}</p>
<a class="btn btn-outline-secondary" href="#">收藏</a>
</div>
<div class="product-device shadow-sm d-none d-md-block"></div>
<div class="product-device product-device-2 shadow-sm d-none d-md-block"></div>
</div>
<div>
<footer class="py-3 my-4">
<ul class="nav justify-content-center border-bottom pb-3 mb-3">
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">首页</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">特性</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">联系我们</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">资料获取</a></li>
<li class="nav-item"><a href="#" class="nav-link px-2 text-muted">关于</a></li>
</ul>
<p class="text-center text-muted">Copyright © 1998 - 2029 liuqingzheng. All Rights Reserved. </p>
</footer>
</div>
</div>
</body>
</html>
6.5 登录认证装饰器
endpoint ：路由的别名
# 问题一：登录装饰器，放上还是下 放下面
# 问题二：每个路由有个别名【反向解析】--》这个别名如果不写--》会以函数名作为别名--》一旦加了登录认证装饰器，所有别名都变成了 inner
# 1 使用装饰器装饰装饰器 @wraps(func)
# 2 每个人都指定别名
from flask import Flask, render_template, request, redirect, session, jsonify
app = Flask(__name__, template_folder="templates", static_folder="static")
app.secret_key = 'adsfa33w9SAD##@sdr434l' # 如果使用session，必须加秘钥
from functools import wraps
# 用户登录，写个装饰器
def login_required(func):
# @wraps(func)
def inner(*args, **kw):
# 判断有没有登录
if session.get('username'):
res = func(*args, **kw)
return res
else:
return redirect('/login')
return inner
# 用户信息
USERS = {
1: {'name': '刘亦菲', 'age': 18, 'gender': '男', 'text': "刘亦菲，1987年8月25日出生于湖北省武汉市，华语影视女演员、歌手，毕业于北京电影学院2002级表演系本科",
'img': 'https://img2.woyaogexing.com/2021/10/16/e3ccba623848430ba83209c0621a2256!400x400.jpeg'},
2: {'name': '彭于晏', 'age': 28, 'gender': '男', 'text': "彭于晏，1982年3月24日出生于中国台湾省澎湖县，毕业于不列颠哥伦比亚大学，华语影视男演员。。。。。。。。",
'img': 'https://img2.woyaogexing.com/2021/10/16/e71aa35728c34313bccb4c371192990f!400x400.jpeg'},
3: {'name': '迪丽热巴', 'age': 38, 'gender': '女', 'text': "迪丽热巴（Dilraba），1992年6月3日出生于中国新疆乌鲁木齐市，毕业于上海戏剧学院，中国内地影视女演员",
'img': 'https://img2.woyaogexing.com/2021/10/30/6a34146dde2d4f1c832463a5be1ed027!400x400.jpeg'},
4: {'name': '亚瑟', 'age': 38, 'gender': '男', 'text': "亚瑟，是腾讯手游《王者荣耀》中一名战士型英雄角色，也是《王者荣耀》的新手英雄之一，既可攻又可守",
'img': 'https://img2.woyaogexing.com/2021/10/30/371b2aa7a03c4f53b7b1bc86f877d7d1!400x400.jpeg'},
}
# 问题一：登录装饰器，放上还是下 放下面
# 问题二：每个路由有个别名【反向解析】--》这个别名如果不写--》会以函数名作为别名--》一旦加了登录认证装饰器，所有别名都变成了 inner
# 1 使用装饰器装饰装饰器 @wraps(func)
# 2 每个人都指定别名
@app.route('/',endpoint='index')
@login_required
def index():
return render_template('index.html', user_dict=USERS)
@app.route('/login', methods=['GET', 'POST'])
def login():
# 2 判断是get请求还是post请求，需要使用request对象
## 2.1 request 正常每次请求都是新的request---》现在flask中全局就一个request--》它会不会乱套？
if request.method == 'GET':
return render_template('login.html')
else:
# 3 取出用户名密码--》请求对象中取--》request
username = request.form.get('username')
password = request.form.get('password')
if username == 'lqz' and password == '123':
# 6 登录成功--》登录信息写入cookie--》使用session写入
session['username'] = username
# 4 登录成功--》重定向到 index
return redirect('/')
else:
# 5 登录失败--》返回login.html 页面，但是页面中渲染错误信息
return render_template('login.html', error='用户名密码错误')
@app.route('/detail/<int:pk>', methods=['GET'],endpoint='detail')
@login_required
def detail(pk):
user = USERS.get(pk)
return render_template('detail.html', info=user)
@app.route('/demo01', methods=['GET'])
def demo01():
request.args
return jsonify({'code': 100, 'msg': '成功'})
if __name__ == '__main__':
app.run()
7 flask的配置文件
PERMANENT_SESSION_LIFETIME : session 过期时间
DEBUG : DEBUG模式
修改配置文件如下：
app.config['DEBUG']=True
app.config['SECRET_KEY']='aaaa'
选择类中的配置app.config.from_object('settings.TestingConfig')
flask中的配置文件是一个flask.config.Config对象（继承字典）,默认配置为：
{
'DEBUG': get_debug_flag(default=False), # 是否开启Debug模式
'TESTING': False, # 是否开启测试模式
'PROPAGATE_EXCEPTIONS': None, # 是否传播异常，None表示使用默认行为
'PRESERVE_CONTEXT_ON_EXCEPTION': None, # 是否在异常发生时保留上下文，None表示使用默认行为
'SECRET_KEY': None, # 应用的密钥，用于安全相关操作
'PERMANENT_SESSION_LIFETIME': timedelta(days=31), # 永久会话的生命周期
'USE_X_SENDFILE': False, # 是否使用X-Sendfile头
'LOGGER_NAME': None, # 日志记录器的名称
'LOGGER_HANDLER_POLICY': 'always', # 日志处理器策略，'always'表示总是记录
'SERVER_NAME': None, # 服务器名称
'APPLICATION_ROOT': None, # 应用的根目录
'SESSION_COOKIE_NAME': 'session', # 会话Cookie的名称
'SESSION_COOKIE_DOMAIN': None, # 会话Cookie的域名
'SESSION_COOKIE_PATH': None, # 会话Cookie的路径
'SESSION_COOKIE_HTTPONLY': True, # 会话Cookie是否仅通过HTTP传输
'SESSION_COOKIE_SECURE': False, # 会话Cookie是否仅通过HTTPS传输
'SESSION_REFRESH_EACH_REQUEST': True, # 每次请求是否刷新会话
'MAX_CONTENT_LENGTH': None, # 最大请求内容长度
'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12), # 发送文件的默认缓存时间
'TRAP_BAD_REQUEST_ERRORS': False, # 是否捕获坏请求错误
'TRAP_HTTP_EXCEPTIONS': False, # 是否捕获HTTP异常
'EXPLAIN_TEMPLATE_LOADING': False, # 是否解释模板加载
'PREFERRED_URL_SCHEME': 'http', # 首选URL方案
'JSON_AS_ASCII': True, # JSON是否作为ASCII编码
'JSON_SORT_KEYS': True, # JSON键是否排序
'JSONIFY_PRETTYPRINT_REGULAR': True, # 是否美化打印JSON
'JSONIFY_MIMETYPE': 'application/json', # JSON响应的MIME类型
'TEMPLATES_AUTO_RELOAD': None, # 是否自动重新加载模板
}
7.1 修改配置
方式一 : app.debug 直接在app上改，实际他们在config属性上（开发阶段用）
app.debug=True
app.secret_key='asdfasdf'
方式二 : app.config['DEBUG']=True 在app.config上改
app.config['DEBUG']=True
app.config['SECRET_KEY']='aaaa'
方式三 : 写一个settings.py，直接在里面写key:value
-app.config.from_pyfile('settings.py')
settings.py
DEBUG=True
SECRET_KEY = 'hello~!ssrheart~!welcome~!'
DB_USER = 'heart'
DB_PASSWORD = 'heart123'
app.py
app.config.from_pyfile('settings.py')
方式四 ： app.config.from_object('settings.TestingConfig')
settings.py
class Config(object):
DEBUG = False
TESTING = False
DATABASE_URI = '127.0.0.1'
DATABASE_PORT = 3306
class ProductionConfig(Config):
DATABASE_URI = '192.168.1.11'
DATABASE_PORT = 3308
class DevelopmentConfig(Config):
DEBUG = True
方式其他
app.config.from_envvar("环境变量名称")
app.config.from_json("json文件名称")
app.config.from_mapping({'DEBUG': True}) #可能会用
- 配置中心---》服务--》有个地址，发送请求会返回一对json格式配置文件
-nocos
8 flask的路由
route的参数，其实就是add_url_rule的参数
@app.route('/',methods=['GET']) #--》decorator--》index=decorator(index)
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
def decorator(f: T_route) -> T_route:
#1 如果没有传 endpoint ，就是None
endpoint = options.pop("endpoint", None)
#2 self 是 app对象，是Flask类的对象--》Flask类中得add_url_rule方法是路由的本质
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
# 于是，注册路由，我们可以不用装饰，自己写
app.add_url_rule('/index',endpoint=None,view_func=index,methods=['GET']) # 等同于django的path
rule: URL规则，路径view_func: 视图函数名称defaults: 默认值，当URL中无参数，函数需要参数时，使用defaults = {'k': 'v'}-为函数提供参数，django中也有，叫kwargsendpoint: 名称[别名]，用于反向生成URL，即: url_for('名称')methods: 允许的请求方式，如: ["GET", "POST"]strict_slashes: 对URL最后的 / 符号是否严格要求subdomain: 子域名访问
@app.route('/index', strict_slashes=False)
#访问http://www.xx.com/index/ 或http://www.xx.com/index均可
@app.route('/index', strict_slashes=True)
#仅访问http://www.xx.com/index
redirect_to: 重定向到指定地址
@app.route('/index/<int:nid>', redirect_to='/home/<nid>')
app.add_url_rule('/index/<int:pk>'): 转换器
DEFAULT_CONVERTERS = {
'default': UnicodeConverter,
'string': UnicodeConverter,
'any': AnyConverter,
'path': PathConverter,
'int': IntegerConverter,
'float': FloatConverter,
'uuid': UUIDConverter,
}
9 cbv
9.1 基本使用
from flask import Flask, url_for
from flask.views import MethodView
app = Flask(__name__)
app.debug = True
class IndexView(MethodView):
def get(self):
return 'get请求'
def post(self):
return 'post请求'
# app.add_url_rule('/index', endpoint='index', view_func=IndexView.as_view(name='lqz'))
如果传了endpoint，以endpoint为准，如果没传，以函数名为准，函数名是view，如果name改了，以name为准
app.add_url_rule('/index', view_func=IndexView.as_view(name='heart')) #
# IndexView.as_view(name='heart') 必须传name，而name是路由的别名
if __name__ == '__main__':
app.run()
9.2 源码分析
# 0 app.add_url_rule('/index', view_func=IndexView.as_view(name='heart'))
# 1 as_view 源码
@classmethod
def as_view(cls, name, *class_args) :
if cls.init_every_request:
def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
self = view.view_class( # type: ignore[attr-defined]
*class_args, **class_kwargs
)
# current_app.ensure_sync 不用看
# 当成 return self.dispatch_request(**kwargs)
return current_app.ensure_sync(self.dispatch_request)(**kwargs)
else:
self = cls(*class_args, **class_kwargs)
def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
return current_app.ensure_sync(self.dispatch_request)(**kwargs)
if cls.decorators:
# 把view的名字改为 传入的name，否则它一直叫view，endpoint没传，就报错
# view.__name__ = name
for decorator in cls.decorators:
view = decorator(view)
view.__name__ = name
return view
# 2 变成了 app.add_url_rule('/index', view_func=view) ，view的名字是传入的name
# 3 请求来了-->执行view()-->有没有参数取决于谁？有没有转换器，有没有defaults
# 4 执行view本质在执行 self.dispatch_request()--> self是 视图类的对象
# 5 MethodView 的dispatch_request
def dispatch_request(self, **kwargs: t.Any) -> ft.ResponseReturnValue:
# 如果是get请求，meth 就是 get方法
meth = getattr(self, request.method.lower(), None)
# 执行get加括号-->跟django没有区别
return meth(**kwargs)
9.3 cbv加装饰器
from flask import Flask
from flask.views import MethodView
app = Flask(__name__)
app.debug = True
### 登录认证--->不能公用--》要么只能给fbv用，要么只能给cbv用
def auth(func):
def inner(*args, **kwargs):
print(args) #cbv,会有第一个参数 self # 如果是fbv，就是空的
res = func(*args, **kwargs)
print('装饰器走了')
return res
return inner
@app.route('/', endpoint='index')
@auth
def index():
return 'hello flask'
# class UserView(MethodView):
# @auth
# def get(self):
# return 'user-get'
class UserView(MethodView):
decorators = [auth] # 顺序从下往上
methods=['POST']
def get(self):
return 'user-get'
def post(self):
return 'user-post'
app.add_url_rule('/user', endpoint='user', view_func=UserView.as_view('user'))
if __name__ == '__main__':
app.run()
10 请求响应
10.1 请求对象
request.method 提交的方法
request.args get请求提及的数据
request.form post请求提交的数据
request.values post和get提交的数据总和
request.cookies 客户端所带的cookie
request.headers 请求头
request.path 不带域名，请求路径
request.full_path 不带域名，带参数的请求路径
request.script_root
request.url 带域名带参数的请求路径
request.base_url 带域名请求路径
request.url_root 域名
request.host_url 域名
request.host 127.0.0.1:500
request.files
obj = request.files['the_file_name']
obj.save('/var/www/uploads/' + secure_filename(f.filename))
10.2 响应对象
set_cookie
from flask import Flask, url_for, request,make_response,jsonify
from flask.views import MethodView, View
app = Flask(__name__)
app.debug = True
class IndexView(MethodView):
def get(self):
# 1 新手四件套：字符串，模板，重定向，json
# 2 向cookie中写入数据
res=make_response('get请求') # make_response 放四件套之一都可
res.set_cookie('name','heart',httponly=True,path='/')
# 3 向响应头中写入数据
res.headers['xxxx']='sss'
return res
def post(self):
return 'post请求'
app.add_url_rule('/index', view_func=IndexView.as_view(name='heart'))
if __name__ == '__main__':
app.run()
11 session
11.1 基本使用
session['name'] = name
name = session.get('name')
session.pop('name')
session.clear()
11.2 原理及源码
# open_session
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
s = self.get_signing_serializer(app)
if s is None:
return None
val = request.cookies.get(self.get_cookie_name(app))
if not val:
return self.session_class()
max_age = int(app.permanent_session_lifetime.total_seconds())
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
# save_session
def save_session(
self, app: Flask, session: SessionMixin, response: Response
) -> None:
name = self.get_cookie_name(app)
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
secure = self.get_cookie_secure(app)
samesite = self.get_cookie_samesite(app)
httponly = self.get_cookie_httponly(app)
# Add a "Vary: Cookie" header if the session was accessed at all.
if session.accessed:
response.vary.add("Cookie")
# If the session is modified to be empty, remove the cookie.
# If the session is empty, return without setting the cookie.
if not session:
if session.modified:
response.delete_cookie(
name,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
httponly=httponly,
)
response.vary.add("Cookie")
return
if not self.should_set_cookie(app, session):
return
expires = self.get_expiration_time(app, session)
val = self.get_signing_serializer(app).dumps(dict(session)) # type: ignore
response.set_cookie(
name,
val, # type: ignore
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
)
response.vary.add("Cookie")
12 请求扩展
flask中叫请求扩展，本质作用实现像django中间件的作用一样
flask也有中间件，但是一般不用，用请求扩展即可
12.1 before_request
- 请求来进视图函数之前执行
- 多个会从上往下依次执行
- 如果返回None，表示继续下一个
- 如果返回了四件套：表示结束，不继续往后走
@app.before_request
def before01():
print('来了老弟1')
# 向请求对象中，放值
request.name='lqz'
@app.before_request
def before02():
print('来了老弟2')
12.2 after_request
- 视图函数执行完，走
- 多个会从下往上依次执行
- 必须有返回值，是响应对象
- 处理跨域，在响应头中加，就用它
@app.after_request
def after01(response):
print('走了老弟1')
return response
@app.after_request
def after02(response):
print('走了老弟2')
response.headers['ssss']='sss'
return response
12.3 teardown_request
- 无论视图函数执行成功或失败，都会走它
- 即便视图函数执行出异常,也会走
- 一般用来记录日志
@app.teardown_request
def teardown(exc):
# exc是视图函数错误对象--》记录错误日志
print(exc)
12.4 errorhandler
- 监听http响应状态码
- 全局异常处理
@app.errorhandler(404)
def error_404(arg):
return jsonify({'code':'xxx'})
@app.errorhandler(500)
def error_500(arg):
return jsonify({'code':'500错误了'})
13 开源项目 sql审核平台
-https://gitee.com/cookieYe/Yearning
-https://gitee.com/rtttte/Archery
14 闪现(flash)
闪现是存放数据的地方，一旦取了，数据就没了
django中也有类似的东西，message消息框架
要用闪现，必须指定secret_key-->闪现内部使用session实现的
# 基本使用
设置：flash('欢迎你：heart')
取：get_flashed_messages()
# 根据标签设置和取值
flash('heart',category="x1")
get_flashed_messages(category_filter=['x1'])
from flask import Flask, request, render_template, redirect,flash,get_flashed_messages
app = Flask(__name__)
app.debug = True
# 要用闪现，必须指定secret_key--》闪现内部使用session实现的
app.secret_key='asdfasdf'
@app.route('/login', endpoint='login',methods=['GET','POST'])
def login():
if request.method == 'GET':
return render_template('login.html')
else:
username = request.form.get('username')
password = request.form.get('password')
if username == 'heart' and password == '123':
# 使用闪现，放个数据
flash('欢迎你：heart')
flash('ss')
flash('ee')
return redirect('/')
else:
flash('用户名密码错误')
return redirect('/')
@app.route('/')
def index():
# 从闪现中取出数据
# print(get_flashed_messages())
return render_template('index.html') # {{ get_flashed_messages() }}
if __name__ == '__main__':
app.run()
15 g对象
g对象，是一个全局对象，global的缩写
g在当次请求中，可以放入值，可以取出值
g和request对象都在当次请求中有效
from flask import Flask, g, request
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdf'
@app.before_request
def before():
if 'index' in request.full_path:
g.name = 'index'
request.name='index'
else:
g.name = '其他'
request.name = '其他'
def add():
print(g.name) # index
print(request.name) # index
@app.route('/index')
def index():
print(g.name) # index
add()
return 'hello'
@app.route('/home')
def home():
print(g.name) # 其他
return 'home'
if __name__ == '__main__':
app.run()
15.1 g和session区别
g 只在当前请求中有效
session可以跨请求
闪现可以跨请求-->本质就是session-->用一次就没了
16 蓝图
Flask 蓝图（Blueprint）是 Flask 中的一种组织和结构化代码的方式，允许将应用程序拆分为更小的、可重用的模块。
通过使用蓝图，可以更容易地管理大型应用程序，将路由、视图函数、模板和静态文件组织到不同的逻辑部分。
蓝图使用步骤
1 蓝图类实例化得到一个对象
from flask import Blueprint
# 第一个参数是蓝图名
order_blue=Blueprint('order',__name__)
2 在视图函数上，使用蓝图注册路由-->蓝图也有请求扩展
@order_blue.before_request
def before():
print('来了')
@order_blue.route('/order')
def order():
return 'order-order-order'
3 在app中注册蓝图
app.register_blueprint(order_blue)
16.1 不使用蓝图
main.py
from core import app
if __name__ == '__main__':
app.run()
core/views/user.py
from core import app
from flask import render_template
@app.route('/login')
def login():
return 'login'
@app.route('/register')
def register():
return render_template('register.html')
init.py
from flask import Flask
app = Flask(__name__,template_folder='../templates',static_folder='../static')
app.config.from_object('core.settings.DevelopmentConfig')
from .views import user
settings.py
class Config(object):
DEBUG = False
TESTING = False
DATABASE_URI = 'sqlite://:memory:'
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
class DevelopmentConfig(Config):
DEBUG = True
16.2 蓝图小项目
views/users.py
from flask import render_template
from flask import Blueprint
# 第一个参数是蓝图名
user_blue=Blueprint('user',__name__)
@user_blue.route('/login')
def login():
return 'login'
@user_blue.route('/register')
def register():
return render_template('register.html',name='heart')
views/order.py
from flask import Blueprint,request,session,g
# 第一个参数是蓝图名
order_blue=Blueprint('order',__name__)
@order_blue.before_request
def before():
print('来了')
@order_blue.route('/order')
def order():
return 'order-order-order'
init.py
from flask import Flask
app = Flask(__name__, template_folder='../templates', static_folder='../static')
app.config.from_object('flask_little.settings.DevelopmentConfig')
from .views.users import user_blue
from .views.order import order_blue
# 注册蓝图，并指定 URL 前缀
app.register_blueprint(user_blue, url_prefix='/user')
app.register_blueprint(order_blue, url_prefix='/order')
16.2 蓝图大项目
apps/goods/init.py
from flask import Blueprint
goods_blue = Blueprint('goods',
__name__, template_folder="../templates", static_folder="../static"
)
from . import views
apps/goods/views.py
from . import goods_blue
@goods_blue.route("/goods")
def goods():
return "this is goods!"
apps/user/init.py
from flask import Blueprint
user_blue = Blueprint(
"user", __name__, template_folder="../templates", static_folder="../static"
)
from . import views
apps/user/views.py
from . import user_blue
@user_blue.route("/login")
def login():
return "this is login!"
flask_big/init.py
from flask import Flask
app = Flask(__name__)
from flask_big.apps.user import user_blue
from flask_big.apps.goods import goods_blue
app.register_blueprint(user_blue, url_prefix="/user")
app.register_blueprint(goods_blue, url_prefix="/goods")
manage.py
from flask_big import app
if __name__ == "__main__":
app.run()
17 flask-session
可以放在redis中，数据库中，文件中
pip3 install flask-session
17.1 方式一
from flask_session.redis import RedisSessionInterface
import redis
from flask import Flask, session
app = Flask(__name__)
app.secret_key = "adsfasdvzxv123esdfadsasda"
conn = redis.Redis(host="127.0.0.1", port=6379)
# 1 client:redis链接对象
# 2 key_prefix:放到redis中得前缀
# 3 use_signer:是否使用secret_key 加密
# 4 permanent:关闭浏览器，cookie是否失效
# 5 生成session_key的长度
app.session_interface = RedisSessionInterface(
app,
client=conn,
key_prefix="session",
use_signer=True,
permanent=True,
sid_length=32,
)
@app.route("/")
def home():
session["name"] = "heart"
return "ok,home!"
if __name__ == "__main__":
app.run(port=5001)
17.2 方式二
py
from flask import Flask, session
from flask_session import Session
app = Flask(__name__)
app.config.from_pyfile("settings.py")
Session(app)
@app.route("/")
def home():
session["name"] = "heart"
return "ok,home!"
if __name__ == "__main__":
app.run(port=5001)
settings.py
from redis import Redis
SESSION_TYPE = "redis"
SESSION_REDIS = Redis(host="127.0.0.1", port="6379")
SESSION_KEY_PREFIX = "heart" # 如果不写，默认以：SESSION_COOKIE_NAME 作为key
17.3 源码分析
# 1 open_session
def open_session(self, app, request):
# 1 取到前端传入，在cookie中得随机字符串
sid = request.cookies.get(app.config["SESSION_COOKIE_NAME"])
if not sid:
sid = self._generate_sid(self.sid_length)
# 不为空--》把sid传入--》得到了session对象
return self.session_class(sid=sid, permanent=self.permanent)
if self.use_signer:
try:
sid = self._unsign(app, sid)
except BadSignature:
sid = self._generate_sid(self.sid_length)
return self.session_class(sid=sid, permanent=self.permanent)
return self.fetch_session(sid)
def fetch_session(self, sid):
# 随机字符串
prefixed_session_id = self.key_prefix + sid
# 从redis中，取出 key 为 前缀+随机字符串 对应的value的值
value = self.redis.get(prefixed_session_id)
if value is not None:
try:
# 解密成字符串
session_data = self.serializer.loads(value)
# 把解密后的数据，组装到 session对象中
return self.session_class(session_data, sid=sid)
except pickle.UnpicklingError:
return self.session_class(sid=sid, permanent=self.permanent)
return self.session_class(sid=sid, permanent=self.permanent)
# 2 save_session
def save_session(self, app, session, response):
if not self.should_set_cookie(app, session):
return
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
if not session:
if session.modified:
self.redis.delete(self.key_prefix + session.sid)
response.delete_cookie(
app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
)
return
expiration_datetime = self.get_expiration_time(app, session)
serialized_session_data = self.serializer.dumps(dict(session))
# 放到redis中
self.redis.set(
name=self.key_prefix + session.sid,
value=serialized_session_data,
ex=total_seconds(app.permanent_session_lifetime),
)
# 把session 对应的 随机字符串放到 cookie中
self.set_cookie_to_response(app, session, response, expiration_datetime)
# 两个点
- session的前缀如果不传，默认：config.setdefault('SESSION_KEY_PREFIX', 'session:')
- session过期时间：通过配置，如果不写，会有默认
'PERMANENT_SESSION_LIFETIME': timedelta(days=31),#这个配置文件控制
-设置cookie时，如何设定关闭浏览器则cookie失效
permanent=False
app.config['SESSION_PERMANENT'] = False
18 wtforms
做数据校验，渲染模版，渲染错误信息用的，前后端混合使用
pip install wtforms
18.1 py
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='用户名长度必须大于%(min)d且小于%(max)d')
],
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-z\d$@$!%*?&]{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()
18.2 html
<!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>
19 数据库连接池
19.1 flask操作mysql
from flask import Flask, jsonify
import pymysql
app = Flask(__name__)
app.debug=True
# 使用全局的链接对象，会有数据安全问题
# 1 拿到mysql链接对象
# 分析--》cursor 和conn是全局的--》在并发情况下有问题
# conn = pymysql.connect(
# user='root',
# password="1234",
# host='127.0.0.1',
# database='blog',
# port=3306,
# autocommit=False)
# cursor = conn.cursor(pymysql.cursors.DictCursor)
#
#
# @app.route('/')
# def index():
# sql = 'select * from blog_tag where id >%s'
# cursor.execute(sql, 1)
# res = cursor.fetchall()
# print(res)
# return jsonify(res)
#
# @app.route('/home')
# def home():
# sql = 'select * from blog_commit where id >%s'
# cursor.execute(sql, 1)
# res = cursor.fetchall()
# print(res)
# return jsonify(res)
# 每个视图函数中创建一个链接对象 数据库链接数过多
@app.route('/')
def index():
conn = pymysql.connect(
user='root',
password="1234",
host='127.0.0.1',
database='blog',
port=3306,
autocommit=False)
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = 'select * from blog_tag where id >%s'
cursor.execute(sql, 1)
res = cursor.fetchall()
print(res)
return jsonify(res)
@app.route('/home')
def home():
conn = pymysql.connect(
user='root',
password="1234",
host='127.0.0.1',
database='blog',
port=3306,
autocommit=False)
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = 'select * from blog_commit where id >%s'
cursor.execute(sql, 1)
res = cursor.fetchall()
print(res)
return jsonify(res)
if __name__ == '__main__':
app.run()
19.2 数据库连接池
pip install dbutils
# 1 先创建出一批链接，每个请求从池中取链接操作
-每个请求用自己的链接对象
-又有池的存在
-数据并发安全，并且链接数不会过高
# 2 dbutils模块，实现数据库连接池
# 1 安装 pip install dbutils
# 新建一个文件
# 2 使用：实例化得到一个池对象---》池是单例
from dbutils.pooled_db import PooledDB
import pymysql
POOL = PooledDB(
creator=pymysql, # 使用链接数据库的模块
maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数
mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
maxcached=5, # 链接池中最多闲置的链接，0和None不限制
maxshared=3,
# 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制
setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
ping=0,
# ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
host='127.0.0.1',
port=3306,
user='root',
password='123456',
database='cnblogs',
charset='utf8'
)
# 3 在视图函数中导入使用
# 数据库连接池
@app.route('/')
def index():
conn = POOL.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = 'select * from blog_tag where id >%s'
cursor.execute(sql, 1)
res = cursor.fetchall()
print(res)
return jsonify(res)
19.3 测试(使用池和不用池)
from flask import Flask, jsonify
from pool import POOL
import pymysql
import time
app = Flask(__name__)
app.debug=True
# 数据库连接池
@app.route('/tag')
def index():
conn = POOL.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = 'select * from blog_tag where id >%s'
cursor.execute(sql, 1)
time.sleep(1)
res = cursor.fetchall()
print(res)
conn.close()
return jsonify(res)
# 没有使用连接池
@app.route('/tag_no')
def tag_no():
conn = pymysql.connect(
user='root',
password="1234",
host='127.0.0.1',
database='blog',
port=3306,
autocommit=False)
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = 'select * from blog_tag where id >%s'
cursor.execute(sql, 1)
time.sleep(1)
res = cursor.fetchall()
print(res)
conn.close()
return jsonify(res)
if __name__ == '__main__':
app.run()
20 flask定制命令
20.1 flask-script
老版本，已经不使用了
# flask 老版本中，没有命令运行项目，自定制命令
# flask-script 解决了这个问题：flask项目可以通过命令运行，可以定制命令 1.x 2.x
# 新版的flask--》官方支持定制命令 click 定制命令，这个模块就弃用了 2.x 3.x
# flask-migrate 老版本基于flask-script，新版本基于flask-click写的
# 使用步骤
-1 pip3 install Flask-Script==2.0.3
-2 pip3 install flask==1.1.4
-3 pip3 install markupsafe=1.1.1
-4 使用
from flask_script import Manager
manager = Manager(app)
if __name__ == '__main__':
manager.run()
-5 自定制命令
@manager.command
def custom(arg):
"""自定义命令
python manage.py custom 123
"""
print(arg)
- 6 执行自定制命令
python manage.py custom 123
20.2 新版本定制命令
from flask import Flask
import click
app = Flask(__name__)
@app.cli.command("create-user")
@click.argument("name")
def create_user(name):
# from pool import POOL
# conn=POOL.connection()
# cursor=conn.cursor()
# cursor.excute('insert into user (username,password) values (%s,%s)',args=[name,'123456'])
# conn.commit()
print(name)
@app.route('/')
def index():
return 'index'
if __name__ == '__main__':
app.run()
# 运行项目的命令是：flask --app py文件名字:app run
# 命令行中执行
# flask --app 7-flask命令:app create-user lqz
# 简写成 前提条件是 app所在的py文件名字叫 app.py
# flask create-user lqz
20.3 django中自定制命令
# 1 app下新建文件夹
management/commands/
# 2 在该文件夹下新建py文件，随便命名（命令名）
# 3 在py文件中写代码
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = '命令提示'
def add_arguments(self, parser):
parser.add_argument('path', nargs='*', type=str,)
def handle(self, *args, **kwargs):
print('开始导入')
print(args)
print(kwargs)
# 4 使用命令
python manage.py py文件(命令名)
21 flask-cache
https://flask.palletsprojects.com/en/3.0.x/patterns/caching/
# pip install Flask-Caching
from flask import Flask
from flask_caching import Cache,SimpleCache
config = {
"DEBUG": True, # some Flask specific configs
"CACHE_TYPE": "SimpleCache", # Flask-Caching related configs ,可以缓存到redis
"CACHE_DEFAULT_TIMEOUT": 300
}
app = Flask(__name__)
app.config.from_mapping(config)
cache = Cache(app)
@app.route('/')
def index():
cache.set('name', 'xxx')
return 'index'
@app.route('/get')
def get():
res=cache.get('name')
return res
if __name__ == '__main__':
app.run()
22 信号
22.1 信号是什么
用信号的好处 : 代码侵入性低，解耦
# 1 Flask框架中的信号基于blinker，其主要就是让开发者可以在flask请求过程中定制一些用户行为
# 2 信号是典型的 观察者模式
-触发某个事执行【模板准备渲染】
-绑定信号：可以绑定多个
只要模板准备渲染--》就会执行这几个绑定的新--》函数
# 3 面向切面编程(AOP)--》一种方案
-整个程序正常运行，但是我们可以把一部分代码，插入到某个位置执行
-钩子函数：只要写了，程序走到哪，就会执行，没写，就不会执行
-序列化类的校验
# 4 通过信号可以做什么事？
-在框架整个执行过程中，插入一些代码执行
比如：记录某个页面的访问量
比如：每次渲染 login.html --->都记录日志
比如：程序出异常---》记录日志
比如：用户表中有个用户创建--》给这个用户发点短信
比如：用户下了订单---》发个邮件通知，让它尽快付款
比如：轮播图表只要发生变化，就删缓存：django中内置信号
22.2 flask中内置信号的使用
# 1 flask中内置信号
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在其中添加数据时，自动触发
# 2 绑定内置信号，当程序执行到信号位置，就执行我们的函数
# 3 信号和请求扩展的关系
-有的信号可以完成之前在请求扩展中完成的事
-但他们机制不一样
-信号更丰富
from flask import Flask,render_template,signals
app = Flask(__name__)
app.debug=True
# 内置信号使用---》当模板渲染前[index.html]--》记录日志
# 1 写一个函数
def func1(*args,**kwargs):
print('模板渲染了')
print(args)
print(kwargs.get('template').name)
if 'index.html' == kwargs.get('template').name:
print('记日志了')
# from jinja2.environment import Template
# 2 跟内置信号绑定
signals.before_render_template.connect(func1)
# 3 等待触发(自动)
@app.route('/<string:name>')
def index(name):
return render_template('index.html',name=name)
@app.route('/login')
def login():
return render_template('login.html')
if __name__ == '__main__':
app.run()
22.3 自定义信号
# 步骤
# 1 定义一个自定义信号
# 2 写一个函数
# 3 跟内置信号绑定
# 4 等待触发(手动)-->只要blog_tag 插入一条记录，就触发
from flask import Flask, render_template, request
from flask.signals import _signals
import pymysql
from pool import POOL
import pymysql
app = Flask(__name__)
app.debug = True
# 自定义信号
# 1 定义一个自定义信号
create_user = _signals.signal('create_user')
# 2 写一个函数
def func1(*args, **kwargs):
print('自定义信号执行了')
if kwargs.get('table_name') == 'blog_tag':
print('记录日志，blog_tag增加了')
print(args)
print(kwargs)
# 3 跟内置信号绑定
create_user.connect(func1)
# 4 等待触发(手动)-->只要blog_tag 插入一条记录，就触发
def insert_data(sql, table_name, *args):
create_user.send(table_name=table_name)
conn = POOL.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
cursor.execute(sql, args)
conn.commit()
@app.route('/create_tag')
def create_tag():
name = request.args.get('name')
blog_id = request.args.get('blog_id')
sql = 'insert into blog_tag (name ,blog_id) values (%s,%s)'
insert_data(sql, 'blog_tag', name, blog_id)
return '创blog_tag成功'
@app.route('/create_category')
def create_category():
name = request.args.get('name')
blog_id = request.args.get('blog_id')
sql = 'insert into blog_category (name ,blog_id) values (%s,%s)'
insert_data(sql, 'blog_category', name, blog_id)
return '创blog_tag成功'
if __name__ == '__main__':
app.run()
22.4 django中信号使用
# 1 内置信号
Model signals
pre_init # django的modal执行其构造方法前，自动触发
post_init # django的modal执行其构造方法后，自动触发
pre_save # django的modal对象保存前，自动触发
post_save # django的modal对象保存后，自动触发
pre_delete # django的modal对象删除前，自动触发
post_delete # django的modal对象删除后，自动触发
m2m_changed # django的modal中使用m2m字段操作第三张表（add,remove,clear）前后，自动触发
class_prepared # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发
Management signals
pre_migrate # 执行migrate命令前，自动触发
post_migrate # 执行migrate命令后，自动触发
Request/response signals
request_started # 请求到来前，自动触发
request_finished # 请求结束后，自动触发
got_request_exception # 请求异常后，自动触发
Test signals
setting_changed # 使用test测试修改配置文件时，自动触发
template_rendered # 使用test测试渲染模板时，自动触发
Database Wrappers
connection_created # 创建数据库连接时，自动触发
# 内置信号使用
1 写一个函数
2 跟内置信号绑定
3 等待触发(自动的)
# 1 写个函数
# 放到__init__里
from django.db.models.signals import pre_save
import logging
def callBack(sender, **kwargs):
# 过滤banner表 ：kwargs就有表名
print('对象保存了')
# celery异步
# 2 绑定
post_save.connect(callBack)
# 3 绑定方式二，使用装饰器
from django.db.models.signals import pre_save
from django.dispatch import receiver
@receiver(pre_save)
def my_callback(sender, **kwargs):
print("对象创建成功")
print(sender)
print(kwargs)
# 自定义信号
# 1 定义信号（一般创建一个py文件）（toppings，size 是接受的参数）
import django.dispatch
pizza_done = django.dispatch.Signal(providing_args=["toppings", "size"])
#2 写个函数注册信号
def callback(sender, **kwargs):
print("callback")
print(sender,kwargs)
pizza_done.connect(callback)
# 3 触发信号
from 路径 import pizza_done
pizza_done.send(sender='seven',toppings=123, size=456)
22.5 信号和信号量
# 信号：signal
-flask，django中的观察者模式 --》信号机制
# 信号量：Semaphore
-并发编程中概念
在Python中，信号量（Semaphore）主要用来控制多个线程或进程对共享资源的访问。信号量本质上是一种计数器的锁，它维护一个许可（permit）数量，每次 acquire() 函数被调用时，如果还有剩余的许可，则减少一个，并允许执行；如果没有剩余许可，则阻塞当前线程直到其他线程释放信号量
23 sqlalchemy
介绍
# 1 sqlalchemy 企业级orm框架
# 2 python界的orm框架
-1 django-orm #只能django框架用
-2 peewee # 小型orm框架
https://docs.peewee-orm.com/en/latest/peewee/quickstart.html
-----同步orm框架-----
-3 sqlalchemy # 企业级，使用非常广泛：https://docs.sqlalchemy.org/en/20/orm/quickstart.html # 可以用在flask上，还可以用在fastapi
---中间态----
-4 Tortoise ORM # fastapi用的比较多
# 3 django ，flask---》同步框架--》新版本支持异步
# 4 fastapi，sanic，tornado---》异步框架--》支持同步的写法--》如果在异步框架上写同步
-众多第三方库---》都是同步的--》导致异步框架性能发挥不出来
-redis：aioredis --》redis-py
-mysql：aiomysql --》pymysql
23.1 快速使用
pip insatll sqlalchemy
# 引擎
# https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine
# 1 架构
Engine，框架的引擎
Connection Pooling ，数据库连接池
Dialect，选择连接数据库的DB API种类
SQL Exprression Language，SQL表达式语言
# 2 链接不同数据库
# 3.1 postgresql
# default
engine = create_engine("postgresql://scott:tiger@localhost/mydatabase")
# psycopg2
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/mydatabase")
# pg8000
engine = create_engine("postgresql+pg8000://scott:tiger@localhost/mydatabase")
# 3.2 Mysql
# default
engine = create_engine("mysql://scott:tiger@localhost/foo")
# mysqlclient (a maintained fork of MySQL-Python)
engine = create_engine("mysql+mysqldb://scott:tiger@localhost/foo")
# PyMySQL
engine = create_engine("mysql+pymysql://scott:tiger@localhost/foo")
# 3.3 oracle
engine = create_engine("oracle://scott:[email protected]:1521/sidname")
engine = create_engine("oracle+cx_oracle://scott:tiger@tnsname")
# 3.4 Microsoft SQL Server
# pyodbc
engine = create_engine("mssql+pyodbc://scott:tiger@mydsn")
# pymssql
engine = create_engine("mssql+pymssql://scott:tiger@hostname:port/dbname")
# 3.5 sqlite
# Unix/Mac - 4 initial slashes in total
engine = create_engine("sqlite:////absolute/path/to/foo.db")
# Windows
engine = create_engine("sqlite:///C:\\path\\to\\foo.db")
# Windows alternative using raw string
engine = create_engine(r"sqlite:///C:\path\to\foo.db")
23.2 sqlalchemy原生操作
import threading
# 1 创建 engin
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3307/test?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1, # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 2 原生操作mysql，用了连接池，就可以不用dbutils模块了
def task():
conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute("select * from test_table")
result = cursor.fetchall()
print(result)
cursor.close()
conn.close()
for i in range(20):
t = threading.Thread(target=task)
t.start()
23.3 sqlalchemy操作表
23.3.1 创建删除表
Base.metadata.create_all(engine) : 创建表
Base.metadata.drop_all(engine) : 删除表
models.py
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
import datetime
# 第一步：创建基类，以后所有类，都必须继承基类
# 1 老版本创建基类(不建议)
# from sqlalchemy.ext.declarative import declarative_base
# Base = declarative_base()
# 2 新版本创建基类
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
# 第二步：写个类，继承
class User(Base):
# 第三步：写字段：传统方式，类型方式
__tablename__='user'
id=Column(Integer,primary_key=True,autoincrement=True)
name = Column(String(32), index=True, nullable=False) # name列varchar32，索引，不可为空
email = Column(String(32), unique=True) # email 列，varchar32，唯一
# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text, nullable=True)
if __name__ == '__main__':
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy001?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 创建表(sqlalchemy默认不能创建库，只能新增或删除表，不能增加删除字段)
Base.metadata.create_all(engine) # 被他管理的表
# 删除表
Base.metadata.drop_all(engine)
23.3.2 增删查改数据
from models import User
# 创建engine
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy001?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 老方式(不推荐使用) 创建session对象
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session=Session()
# 新方式 创建session对象
from sqlalchemy.orm import Session
session=Session(engine)
增加
# 增加单条
user=User(name='heart1',email='[email protected]')
session.add(user)
session.commit() #提交
# 增加多条
user1=User(name='heart2',email='[email protected]')
user2=User(name='heart3',email='[email protected]')
session.add_all([user1,user2]) # 必须是列表
session.commit() #提交
session.close()
查询
res=session.query(User).filter_by(name='heart').all()
res=session.query(User).filter_by(id=1).all()
print(res) # [heart]
print(res[0]) # heart
删除
res=session.query(User).filter_by(name='heart1').delete()
print(res)
session.commit()
修改
# 方式一，直接修改
res=session.query(User).filter_by(id=2).update({'name':'zeus'})
session.commit()
# 方式二，查出来对象修改
user=session.query(User).filter_by(id=2).first()
user.name='god'
session.add(user) # [id字段在不在] 如果对象不存在，就是新增，如果对象存在，就是修改
session.commit()
23.4 一对多关系
23.4.1 新增
models.py
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
def __str__(self):
return self.caption
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名
hobby_id = Column(Integer, ForeignKey("hobby.id")) # 一个爱好，可以被多个人喜欢，一个人只能喜欢一个爱好
# 跟数据库无关，不会新增字段，只用于快速链表操作
# 类名，backref用于反向查询
hobby = relationship('Hobby', backref='pers')
def __str__(self):
return self.name
def __repr__(self):
return self.name
from models import Person,Hobby
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy001?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
from sqlalchemy.orm import Session
session=Session(engine)
# 笨办法增加
#1 先增加一个hobby
hobby=Hobby()
session.add(hobby)
session.commit()
#2 增加Person---》必须要有hobby_id
person=Person(name='heart',hobby_id=1)
session.add(person)
session.commit()
# 简便方法 增加person的同时 增加了Hobby
person = Person(name='heart', hobby=Hobby(caption='乒乓球'))
session.add(person)
session.commit()
# 先查出来hobby对象，然后增加给person，字段是relationship的
hobby = session.query(Hobby).filter_by(id=1).first()
person = Person(name="heart3", hobby=hobby)
session.add(person)
session.commit()
23.4.2 跨表查询正向反向
# 基于对象的跨表查询--->正向
person = session.query(Person).filter_by(nid=2).first()
print(person) # heart1
print(person.hobby_id) # 2
print(person.hobby) # 乒乓球
# 基于对象的跨表查询--->反向
hobby = session.query(Hobby).filter_by(id=1).first()
print(hobby.caption) # 篮球
print(hobby.pers) # [heart, heart1]
23.5 多对多关系
models.py
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
ctime = Column(DateTime, default=datetime.datetime.now)
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
def __repr__(self):
return self.name
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
girls = relationship('Girl', secondary='boy2girl', backref='boys')
def __repr__(self):
return self.name
from models import Boy,Girl,Boy2Girl
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy001?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
from sqlalchemy.orm import Session
session=Session(engine)
# 1 增加记录--笨办法
boy=Boy(name='张四')
boy1=Boy(name='张五')
girl=Girl(name='小粉')
girl1=Girl(name='小紫')
session.add_all([boy,girl,boy1,girl1])
session.commit()
# 2 往中间表增加记录
res=Boy2Girl(boy_id=1,girl_id=1)
session.add_all([res])
session.commit()
# 3 简便方式增加记录
girls=session.query(Girl).filter(Girl.id>=2).all()
print(girls)
# 3.1 让id为2的，跟上面俩女生约会
boy = session.query(Boy).filter(Boy.id == 2).all()[0]
print(boy)
boy.girls=girls
session.add(boy)
session.commit()
# 3.2 新用户跟上面俩女生约会
boy = Boy(name='彭于晏',girls=girls)
session.add(boy)
session.commit()
# 4 基于对象的跨表查--》正向
boy = session.query(Boy).filter(Boy.id == 2).all()[0]
print(boy.name)
print(boy.girls)
# 基于对象的跨表查--》反向
girl = session.query(Girl).filter(Girl.id == 2).all()[0]
print(girl.name)
print(girl.boys)
23.6 sqlalchemy的更多查询
23.6.1 filter_by查询
filter_by()括号内只能是等于的条件，不能有大于小于
res = session.query(User).filter_by(name='heart',id=1).all() # 返回列表
res = session.query(User).filter_by(name='heart',id=1).first() # 返回单条
23.6.2 filter/where查询
可以写条件
res = session.query(User).filter(id>=1) # 原生sql
res = session.query(User).filter(id>=1).all()
res = session.query(User).where(id==1).all() # 就是filter
23.6.3 between查询
res = session.query(User).where(User.id.between(1,10),User.name == 'heart').all()
23.6.4 in查询
因为in是关键字，所以是 in_
res = session.query(User).where(User.id.in_([1,3,4]).all()
23.6.5 取反查询
其实就是 not in
res = session.query(User).where(~User.id.in_([1,3,4]).all()
23.6.6 二次筛选
res = session.query(User).where(User.id.in_(session.query(User.id).filter_by(name='heart'))).all()
23.6.7 and,or条件
from sqlalchemy import and_, or_
or包裹的都是or条件，and包裹的都是and条件
res = session.query(User).filter(and_(User.id > 3, User.name == 'eric')).all()
res = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()
res = session.query(User).filter(or_(User.id < 2,and_(User.name == 'eric', User.id > 3),User.extra != ""))
print(res)
23.6.8 通配符，以e开头，不以e开头
res = session.query(User).filter(User.name.like('l%')).all()
res = session.query(User).filter(~User.name.like('l%'))
23.6.9 限制，用于分页，区间
res = session.query(User)[0:3] # 第一页，一页显示三条
23.6.10 排序
根据name降序排列（从大到小） desc asc
res = session.query(User).order_by(User.name.asc()).all()
第一个条件重复后 再按第二个条件升序排 字符串按什么排?
第一个条件相同时，将按照第二个条件进行排序。
res = session.query(User).order_by(User.name.desc(), User.id.asc()).all()
23.6.11 分组
from sqlalchemy.sql import func
res = session.query(User).group_by(User.name).all() # 分组完要统计 但是这个sql一般不写
分组之后取最大id，id之和，最小id
select max(id),sum(id),min(id),name from user group by user.name;
res = session.query(func.max(User.id),func.sum(User.id),func.min(User.id),User.name).group_by(User.name).all()
23.6.12 having筛选
res = session.query(func.max(User.id),func.sum(User.id),func.min(User.id),User.name).filter(User.id>=4).group_by(User.name).having(func.sum(User.id) > 400).all()
23.6.13 连表
默认用forinkey关联
笛卡尔积链表
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all() #笛卡尔积后过滤
join表 默认是inner join 没有指定on字段，默认按外键
ret = session.query(Person).join(Hobby)
isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
ret = session.query(Person).join(Hobby, isouter=True)
ret = session.query(Hobby).join(Person, isouter=True)
自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_，因为我们可以没有强外键关联
res = session.query(Person).join(Hobby, Person.nid == Hobby.id, isouter=True)
union和union all的区别？
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
res = q1.union(q2).all()
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
res = q1.union_all(q2).all()
何时使用
- 使用
UNION：当你希望合并结果集并去除重复记录时使用。例如，需要一个唯一值列表的情况。- 使用
UNION ALL：当你希望合并结果集而不关心重复记录时使用。例如，当你需要所有记录并且关注查询性能时。总结
UNION去除重复记录，但性能较慢。UNION ALL不去除重复记录，性能较快。
24 自己集成到项目中
# 1 先创建库
# 2 使用sqlalchemy 把表同步到库中
# 3 新建db.py--->实例化得到一个线程安全的session对象
# 4 以后在任意py文件中，导入，再视图函数中使用即可
25 flask-sqlalchemy使用
pip install flask-sqlalchemy
## 借助于flask-sqlalchemy
1 导入 from flask_sqlalchemy import SQLAlchemy
2 实例化得到对象
db = SQLAlchemy()
3 将db注册到app中
db.init_app(app)
------2,3 可以合并为db = SQLAlchemy(app)--------
4 视图函数中使用session
全局的db.session # 线程安全的
5 models.py 中继承Model
db.Model
6 写字段
username = db.Column(db.String(80), unique=True, nullable=False)
7 配置文件中加入
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://[email protected]:3306/ddd?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
src/init.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
app = Flask(__name__, template_folder="../templates", static_folder="../static")
app.config.from_pyfile("./settings.py")
db.init_app(app)
from .views import bp_user
app.register_blueprint(bp_user)
src/models.py
import datetime
from . import db
# 单表
class User(db.Model):
__tablename__ = 'users' # 表名
# 写字段
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id 主键
name = db.Column(db.String(32), index=True, nullable=False) # name列，索引，不可为空
email = db.Column(db.String(32), unique=True)
ctime = db.Column(db.DateTime, default=datetime.datetime.now)
extra = db.Column(db.Text)
def __str__(self):
return self.name
def __repr__(self):
return self.name
src/settings.py
DEBUG=True
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:[email protected]:3306/sqlalchemy003?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
src/views.py
from flask import Blueprint, render_template, jsonify
from .models import User
bp_user = Blueprint('user', __name__)
from . import db
@bp_user.route('/')
def index():
user=User(name='xxx',email='[email protected]')
db.session.add(user)
db.session.commit()
# 把用户表中所有数据都查出来，返回给前端
res = db.session.query(User).filter(User.id >= 1).all()
# 不行 json序列化，只能序列化 数字，字符串，布尔，列表，和字典
# res 列表套对象---》不能序列化
# l = []
# for item in res:
# l.append({'name': item.name, 'email': item.email})
# return jsonify({'code': 100, 'msg': "成功", 'results': l})
return render_template('index.html',users=res)
app.py
from src import app
if __name__ == '__main__':
app.run()
26 flask-migrate使用
https://github.com/miguelgrinberg/Flask-Migrate/
pip install Flask-Migrate
from flask_migrate import Migrate
app = Flask(__name__)
app.config.from_pyfile('./settings.py')
db = SQLAlchemy(app)
migrate = Migrate(app, db) # flask 就会多出好几个命令
flask --app manage:app db init # 初始化，第一次执行，以后再也不执行了，它执行完，会出现一个migrations文件夹
flask --app manage:app db migrate # django中的makemigrations 是一模一样
flask --app manage:app db upgrade # 跟django的migrate一样
# flask上其他第三方插件
cors
token
cache
restful
27 scoped线程安全
# 1 如果我们把session 做成全局 单例
-每个视图函数，用同一个session，有问题
#2 如果在每个视图函数中，都对session实例化一次
-代码有点麻烦
# 3 全局就用一个session对象，它在不同线程中---》都是这个线程自己的
from models import User
import threading
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy001?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 第二步：创建 线程安全的 session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
Session = sessionmaker(bind=engine)
session=scoped_session(Session)
# 第三步：正常使用-->再flask中，使用全局的session即可，实现：不同线程使用线程自己的session对象
def task(se,i):
session=se()
session.add(User(name='xxx',email=f'{i}@qq.com'))
session.commit()
print('=========',session)
if __name__ == '__main__':
l=[]
for i in range(10):
t=threading.Thread(target=task,args=[session,i])
t.start()
l.append(t)
for i in l:
i.join()
补充
1 偏函数
可以提前传值
from functools import partial
def add(a,b,c):
return a+b+c
res=add(1,2,3)
print(res)
res=partial(add,1) # 提前传值
print(res(2,3))
2 flask中可以使用的模块
跨域 flask-cors
jwt flask-jwt
后台管理admin flask-admin
前后端分离resful flask-resful
3 flask 项目
https://gitee.com/pear-admin/pear-admin-flask