flask_day01
Posted xiao-fu
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了flask_day01相关的知识，希望对你有一定的参考价值。
目录
Flask和pythonweb框架介绍
python中的web框架，本质都是一样的
Django：大而全，内置的app多，第三方app也多
Flask：小而精，没有过多的内置组件，只完成web框架最基本的功能，需要借助于第三方，完成更丰富的功能
web.py：是一个小巧灵活的python框架，它简单而且功能强大(国内几乎没有用的)
异步web框架
fastapi：python的异步web框架啊，不少公司在用。链接：https://fastapi.tiangolo.com/zh/
sanic：python的异步web框架，供支持异步高并发请求的web服务
tornado：异步框架，用的比较少
同步框架和异步框架区别
Django是同步框架还是异步框架，Django 3.x以后支持异步
同步框架的意思：一个线程只处理一个请求
异步框架的意思：一个线程可以处理多个请求
异步框架可以显著的提高并发量
flask介绍
Flask是一个基于python开发并且依赖jinjia2模板和WerKzeug WSGI服务的一个微型框架
jinjia2 魔板语法，Django的dtl，非常像
Werkzeug WSGI 符合wsgi协议的web服务器，Django使用的是wsgiref
# wegiref写web
from wsgiref.simple_server import make_server
# mya 就等同于Django
def mya(environ, start_response):
# 把environ包装成了request
print(environ)
stat_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, how are you!</h1>\'
return [data] # 做成了response
if __name == \'__main__\':
myserver = make_server(\'\', 8001, mya)
print(\'监听着8001\')
myserver.server_forever()
# 使用werkzegu写web
from werkzeug,wrappers import Requestm Response
@Request.application
def hello(request):
return Response(\'Hello World!\')
if __name == \'__main__\':
from werkzeug.serving import run_simple
run_simlpe(\'localhost\', 4000, hello)
flask快速使用
安装：pip install flask 同时安装的依赖有：MarkupSafe，Werkzeug, Jinja2，flask
flask版本就一共有：1.x没有本质区别 2.x没有本质区别，源码上动了，用起来一样
from flask import Flask
app = Flask(__name__)
# app=Flask(\'test\')
# 注册路由 --->>> 装饰器
@app.route(\'/index\')
def index():
return \'hello web\'
if __name__ == \'__main__\':
# app.run(\'127.0.0.1\', 5000)
app.run()
登录、显示用户信息小案例
login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form method="post">
<p>用户名：<input type="text" name="username"></p>
<p>密码：<input type="password" name="password"></p>
<input type="submit" value="登录"> error
</form>
</body>
</html>
home.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>用户列表</h1>
<table>
% for k,v in user_dict.items() %
<tr>
<td>k</td>
<td>v.name</td>
<td>v[\'name\']</td>
<td>v.get(\'name\')</td>
<td><a href="/detail/k">查看详细</a></td>
</tr>
% endfor %
</table>
</body>
</html>
detail.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>名字是：user.name</p>
<p>年龄是：user[\'age\']</p>
<p>性别是：user.get(\'gender\')</p>
<p>user.text</p>
</body>
</html>
py文件
from flask import Flask,request,render_template,redirect,session,jsonify
app = Flask(_name__)
# 要使用session，必须设置秘钥，秘钥是配置信息
app.secrest_key = \'adafagateagafagaffssd\'
USERS =
1:\'name\': \'张三\', \'age\': 18, \'gender\': \'男\', \'text\': \'没有那么简单\',
2:\'name\': \'李四\',\'age\': 28, \'gender\': \'男\', \'text\':\'安全第一\',
3:\'name\': \'王五\', \'age\': 18, \'gender\': \'女\', \'text\': \'开怀大笑\'
# 创建templates文件夹，写login.html
@app.route(\'/login\', methods=[\'GET\', \'POST\'])
def index():
# 没有request对象，使用全局的request
# get请求，返回模板
if request.method == \'GET\':
return
render_template(\'login.html\') # 新手四件套之一：返回模板
else:
# post请求，校验数据
# 取出前端传入的用户密码，校验
username = rquest.form.get(\'username\') # 等同于Django的request.POST
password = request.form.get(\'password\')
if username == \'lqz\' and password == \'123\':
# 登录成功，保存登录状态 重定向到根路径 新手四件套之一：重定向
# 保存到session中，session是全局的
session[\'name\'] = username
return redirect(\'/\')
else:
return
render_template(\'login.html\', error=\'用户名或密码错误\') # 注意跟Django的render区分，要模板渲染的数据，直接key=value传即可
@app.route(\'/\')
def home():
# 校验，登录成功，才能过来，不登录，重定向到登录页面
if session.get(\'name\'): # 有值说明登录了，没有值说明没有登录
return render_template(\'home.html\', user_dict=USERs)
else:
return redirect(\'login\')
@app.route(\'/detail/<int:pk>\')
def detail(pk):
if session.get(\'name\'): # 有值书说明登录了，没有值说明没有登录
user_detail = USERS.get(pk)
return render_template(\'detail.html\', user=user_detail)
else:
return redirect(\'/login\')
@app.route(\'/test\')
def test():
return
jsonify([\'name\': \'lqz\', \'age\':19])
if __name__ == \'__main__\':
app.run()
学到的有：
1.注册路由，app.route(路径，methods=[请求方式get,post])
2.新手四件套：
- render_template， 渲染模板，跟Django是有区别的
- redirect， 重定向
- return字符串， 返回字符串
- jsonify，返回json返回格式
3.请求的request对象，是全局的，直接导入使用即可，在不同的视图函数中不会出现混乱
- rquest.method 请求发方式
- request,form post请求的body体的内容转成了字典
4.session 全局的，直接导入使用即可，一定要指定秘钥app.secret_key = \'afagegfgrdaseve\'
- 放值：session[\'name\'] = \'lqz\'
- 取值：session.get(\'name\')
5.模板的渲染
- 兼容Django的dtl
- 更强大，可以加括号，字典可以.get.values() .items()
6.转换器
@app.route(\'/detail/< int:pk >\')
配置文件方式
# Django 有个settings
# flask 也有配置方式，但是它的使用方式有多种：
设置的方式一: (测试用)
# app.debug=True # 调试模式，提示信息更详细，修改代码不需要重启，自动重启
# app.secret_key=\'dasdfasdfasd\' # 秘钥，只能 放debug和secret_key
设置方式二：直接使用app.config设置
# app.config[\'DEBUG\'] = True
# app.config[\'SECRET_KEY\'] = \'afgasfgrg\'
# print(app.config)
设置方式三：使用py文件(不常用)
app.congfig.form_pyfile(\'settings.py\')
print(app.config)
设置方式四：常用的、类的方式
# app.config.form_object(\'settings.DevelopmentConfig\')
# app.config.from_object(\'settings.ProductionConfig\')
# print(app.config)
----------其他方式：-----------
# 通过环境变量配置
# app.config.from_envvar("环境变量名称")
# json
# app.config.from_json("json文件名称")
# JSON文件名称，必须是json格式，因为内部会执行json.loads
# 字典格式---》配置中心
# app.config.from_mapping(\'DEBUG\': True)
# 内置的配置字段，其他可以写自己的，比如redis的连接地址，mysql的连接地址
DEBUG
SECRET_KET
SESSION_COOKIE_NAME 可以cookie中session这个key
PERMANENT_SESSION_LIFETIME cookie过期时间
常用的：
这样就相当于配置了两套配置环境
settings.py 例如：
SESSION_COOKIE_NAME
路由系统
路由本质
django中配置路由 在url.py中，写path，是写在列表中
flask是基于装饰器的，大部分都用装饰器来做，少量可以抽取到一个urls.py中
# 路由的装饰器源码分析
# 路由的装饰器源码分析
# 咱们这样写
@app.route(\'/login\')
def index():
pass
#本质是---》index=app.route(\'/login\')(index)
# app.route(\'/login\')的执行结果 decorator 函数
-rule是路径
-其他参数都给了options
# 然后 decorator(index)--->在执行
# f是index
endpoint = options.pop("endpoint", None) # 目前没有endpoint，是None
# 核心，本质--》self就是实例化得到的app对象，flask对象
# app对象中有个方法add_url_rule，这是在添加路由
# 不使用装饰器，自己注册路由
self.add_url_rule(rule, endpoint, f, **options)
return f
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
def decorator(f: T_route) -> T_route:
endpoint = options.pop("endpoint", None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
# 可以不使用装饰器的方式，注册路由
app.add_url_rule(\'/\', endpoint=None, view_func=home, methods=[\'GET\'])
# flask路由的本质是app对象的add_url_rule完成路由的注册
路由参数add_url_rule
# rule URL规则
# view_func 视图函数名称
# defaults = None 默认值, 当URL中无参数，函数需要参数时，使用defaults = \'k\': \'v\'为函数提供参数
# endpoint = None, 路径的别名，名称，用于反向解析URL，即： url_for(\'名称\')
# methods = None, 允许的请求方式，如：["GET", "POST"]
#对URL最后的 / 符号是否严格要求
strict_slashes = None
\'\'\'
@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 = None,
\'\'\'
@app.route(\'/index/<int:nid>\', redirect_to=\'/home/<nid>\')
\'\'\'
# 需要记住的
# rule
# view_func
# defaults
# endpoint
# methods
转换器
\'default\': UnicodeConverter,
\'string\': UnicodeConverter,
\'any\': AnyConverter,
\'path\': PathConverter,
\'int\': IntegerConverter,
\'float\': FloatConverter,
\'uuid\': UUIDConverter,
# 了解：让路由支持正则（忽略掉）
不写就默认
flask、Django都推荐使用转换器
可以默认不写，路径后面可以跟int str类型的后缀路由
flask路由的本质就是app对象的add_url_rule完成路由的注册
class装饰器，类装饰器，
flask_day05：信号 Django信号 flask-script sqlalchemy 创建操作数据表
目录
鲁棒性
链路，链路追踪，
上下游，大的单体应用，上游还是前端，后端是Django写的
回顾
1.导出项目依赖 pipreqs
2.函数和方法
3.local对象
并发编程中的一个对象，它可以保证多线程并发访问数据安全
本质原理是：不同的线程，操作的是自己的数据
不支持协程
4.自己定义local，支持线程和协程
注意点一：
try:
# 只要解释没有装greenlet，这句话就会报错
# 一旦装了，有两种情况，使用了协程和没用协程，无论使用不使用，用getcurrent都能拿到协程id号
from greenlet import getcurrent as get_ident
except Exception as e:
from threading import get_ident
# 注意点二：重写类的 __setattr__ 和 __getatte__
对象点属性 取值 不存在会触发 __getattr__
对象点属性 设置值 不存在时会触发 __setattr__
# 注意点三：由于重写了__setattr__个__getattr__
类内部使用 self.storage 会递归
使用类调用对象的方法，他就是普通函数，有几个值传几个值
object.__setattr__(self,\'storage\',)
等同于：self.storage=
等价于：setattr(self,\'stotrage\',) 会递归
5.flask是如何实现这个local类的
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = name:value
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
def __setattr__(self, k, v):
ident = get_ident() #如果用协程，这就是协程号，如果是线程，这就是线程号
if ident in self.storage: #\'协程id号\':arg:1,\'协程id号\':arg:2,\'协程id号\':arg:3
self.storage[ident][k] = v
else:
self.storage[ident] = k: v
6.偏函数 ：提前传值，返回一个对象，后期可以调用这个对象，传入后续的值
7.请求上下文源码分析(ctx 对象)，整个flask的执行流程
-一旦请求来了----》会执行 Flask类的对象app()---》触发Flask __call__--->self.wsgi_app(environ, start_response)
-Flask类wsgi_app 方法 大约 2417行
def wsgi_app(self, environ, start_response):
#1 返回了一个ctx，请求上下文对象，RequestContext 的对象，里面有session，request
ctx = self.request_context(environ)
try:
try:
# 2 ctx.push---->RequestContext的push---》382行
# _request_ctx_stack.push(self)--self是ctx---》是全局变量
# 是LocalStack()的对象
ctx.push()
# 匹配路由执行视图函数，请求扩展
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
# 把结果返回给wsgi服务器
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
# 把当前放进去的ctx剔除，当次请求结束了
ctx.auto_pop(error)
-是LocalStack()的对象 的push ，传入了ctx
def push(self, obj):
# self._local是 Flask自己定义的兼容线程和协程的Local
#self._local中反射 stack，会根据不同线程或协程，返回不同线程的stack
#rv是None的
rv = getattr(self._local, "stack", None)
if rv is None:
# rv=[]
# self._local.stack=rv
#self._local=\'协程id号1\':stack:[],\'协程id号2\':stack:[]
self._local.stack = rv = []
rv.append(obj)
#self._local=\'协程id号1\':stack:[ctx,],\'协程id号2\':stack:[]
return rv
- 在视图函数中:request = LocalProxy(partial(_lookup_req_object, "request"))
-print(request.method) # 执行requets对象的 __getattr__
-LocalProxy的__getattr__-->核心：
-return getattr(self._get_current_object(), name)
-self._get_current_object() 是 ctx中的真正request对象，那method，自如就拿到当次请求的method
-def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
#object.__setattr__(self, "_LocalProxy__local", local)，初始化传入的
# local 是 partial(_lookup_req_object, "request")
#
# getattr(_lookup_req_object(\'request\'), \'method\')
# getattr(当次请求的reuqest, \'method\')
return self.__local() # self中的 __local,隐藏属性
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
-def _lookup_req_object(name):
# 这里把当前线程下 的ctx取出来了
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name) # 去ctx中反射request，返回的就是当次请求的requets
# django flask 同步框架，部署的时候，使用uwsgi部署，uwsgi是进程线程架构，并发量不高
# 可以通过uwsgi+gevent，部署成异步程序
信号
Flask框架中的信号基于blinker(安装这个模块)，其主要就是让开发者可以在flask请求过程中定制一些用户行为，flask 和 Django都有
观察者模式，又叫发布 -订阅(Publish/Subscribe) 23种设计模式之一
安装：pip install blinker
信号：signal 翻译过来的，并发编程中学过，信号量Semaphore，是两个不同的概念
比如：用户表新增一条记录时，就记录一下日志
方案一：在每个增加后，都写一行代码 --->>> 后期要删除，比较麻烦
方案二：使用信号量，写一个函数，绑定内置信号，只要程序执行到这，就会执行这个函数
内置信号：flask少一些，Django多一些
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在其中添加数据时，自动触发
使用内置信号量的步骤
1.写一个函数
2.绑定内置信号
3.等待被触发
自定义信号
1.定义出信号
session_set = _signals.signal(\'session_set\')
2.写一个函数
def test1(*args, **kwargs):
print(args)
print(kwargs)
print(\'session设置值了\')
3.绑定自定义的信号
session_set.connect(test1)
4.触发信号的执行(咱们做)
session_set.send(\'lqz\') # 触发信号的执行
Django信号
https://www.cnblogs.com/liuqingzheng/articles/9803403.html
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 # 请求异常后，自动触发
Database Wrappers
connection_created # 创建数据库连接时，自动触发
django中使用内置信号
1.写一个函数
def callBack(*args, **kwargs):
print(args)
print(kwargs)
2.绑定信号
方式一：
post_save.connect(callBack)
方式二：
from django.db.models.signals import pre_save
from django.dispatch import receiver
@receiver(pre_save)
def my_callback(senderm,**kwargs):
print(\'对象创建成功\')
print(sender)
print(kwargs)
3.等待触发
flask-script
django中有命令：python manage.py runserver。。。
flask启动项目，像Django一样，通过命令启动
Flask==2.2.2
Flask_Script==2.0.3
借助于：flask-script 实现
安装：pip install flask-script
修改代码：
from flask_script import Manager
manager=Manager(app)
manager.run()
用命令启动
python manage.py runserver
自定制命令
1.简单自定制命令
@manager.command
def custom(arg):
# 命令的代码，比如：初始化数据库， 有个excel表格，使用命令导入到mysql中
print(arg)
2.复杂一些的自定制命令
@manager.option(\'-n\', \'--name\', dest=\'name\')
@manager.option(\'-u\', \'--url\', dest=\'url\')
def cmd(name, url):
# python run.py cmd -n lqz -u xxx
# python run.py cmd --name lqz --url uuu
print(name, url)
django 中如何自定制命令？
sqlalchemy 快速使用
原生操作的快速使用
# 先不是orm，而是原生sql
# 第一步：导入
from sqlalchemy import create_engine
# 第二步：生成引擎对象
engine = create_engine(
"mysql+pymysql://root@127.0.0.1:3306/cnblogs",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 第三步：使用引擎获取连接，操作数据库
conn = engine.raw_connection()
cursor=conn.cursor()
cursor.execute(\'select * from aritcle\')
print(cursor.fetchall())
创建操作数据表
# 第一步：导入
from sqlalchemy import create_engine
import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
# 第二步：执行declarative_base，得到一个类
Base = declarative_base()
# 第三步：继承生成的Base类
class User(Base):
# 第四步：写字段
id = Column(Integer, primary_key=True) # 生成一列，类型是Integer，主键
name = Column(String(32), index=True, nullable=False) # name列varchar32，索引，不可为空
email = Column(String(32), unique=True)
# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
# 第五步：写表名 如果不写以类名为表名
__tablename__ = \'users\' # 数据库表名称
# 第六步：建立联合索引，联合唯一
__table_args__ = (
UniqueConstraint(\'id\', \'name\', name=\'uix_id_name\'), # 联合唯一
Index(\'ix_id_name\', \'name\', \'email\'), # 索引
)
class Book(Base):
__tablename__ = \'books\'
id = Column(Integer, primary_key=True)
name = Column(String(32))
# 第七步：把表同步到数据库中
# 不会创建库，只会创建表
engine = create_engine(
"mysql+pymysql://root@127.0.0.1:3306/aaa",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 把表同步到数据库 （把被Base管理的所有表，都创建到数据库）
Base.metadata.create_all(engine)
# 把所有表删除
# Base.metadata.drop_all(engine)
以上是关于flask_day01的主要内容，如果未能解决你的问题，请参考以下文章
如何防止 pandas.to_datetime() 函数将 0001-01-01 转换为 2001-01-01
01.01教程简介01.02软件的基本工作界面01.03模型设计的一般过程