信号
内置信号的使用
# 第一步:写一个函数
def test(app, **kwargs):
print(app)
print(type(kwargs))
# 请求地址是根路径,才记录日志,其它都不记录
print(kwargs['context']['request'].path)
if kwargs['context']['request'].path == '/':
print('记录日志了')
# 第二步:跟内置信号绑定
# signals 中有很多内置信号
signals.before_render_template.connect(test)
# 第三步:等待信号被触发(不需要咱们做)--->只要执行到内置信号位置,绑定的函数就会执行
有哪些内置信号
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 第三步:函数跟自己定义信号绑定
# 4 第四步:触发自定义信号---》我们做文章来源:https://www.toymoban.com/news/detail-632327.html
文章来源地址https://www.toymoban.com/news/detail-632327.html
from flask import Flask, session, render_template, signals
from flask.signals import _signals
# pip3 install blinker
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdfasdf'
#### 自定义信号---》session每次放一个值,我们就执行信号
# 1 定义信号
# 自定义信号
session_input = _signals.signal('session_input')
# 2 写个函数
def test2(*args, **kwargs):
print(args) # app
print(kwargs) # {session kk}
print('session放值了')
if kwargs.get('kk').get('name') == 'xx':
print('记录日志')
# 3 绑定信号
session_input.connect(test2)
# 4 触发信号
@app.route('/')
def index():
session['uu'] = '00'
session_input.send(app, session=session, kk={'name': 'uu'})
return render_template('index.html')
@app.route('/home')
def home():
return render_template('home.html')
@app.route('/order')
def order():
session['xx'] = 'xx'
session_input.send(app, session=session, kk={'name': 'xx'})
return "order"
if __name__ == '__main__':
app.run(port=8080)
信号的作用(信号量--》Semaphore)
# 对代码进行解耦
# 1 记录日志:只要是张三用户,访问index页面,就记录日志
# 2 只要用户表中,插入一条记录,就给用户发个短信通知
User.object.create--->调用发短信方法--》找到10个地址--》改10个地方
-如果有个内置信号---》只要表中增加记录,就会触发这个信号----》通过信号内判断这个表是不是User表,决定要不要发短信
-flask中没有这个内置信号---》自定义
信号量
# acquire():消耗信号量
# release():释放信号量
import threading
import time
def run(n):
semaphore.acquire() # 计数器获取锁
time.sleep(5) # 程序休眠5秒
print (n)
semaphore.release() # 计数器释放锁
if __name__ == '__main__':
# 添加一个计数器,最大并发线程数量5(最多同时运行5个线程)
semaphore = threading.Semaphore(5)
for i in range(50):
t = threading.Thread(target=run, args=(i, )) # 创建线程
t.start()
django的信号
# 内置信号:
#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 # 创建数据库连接时,自动触发
# 内置信号使用(当user表创建用户,就给用户发个邮件)
1 写个函数 #放到__init__里
from django.db.models.signals import pre_save
import logging
def callBack(sender, **kwargs):
logging.debug('%s创建了一个%s对象'%(sender._meta.model_name,kwargs.get('instance').title))
2 绑定内置信号
pre_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)
# 自定义信号干过什么?
-做双写一致性的缓存更新
flask-script
# django中,有命令
python manage.py runserver
python manage.py makemigrations
...自定制命令(django如何自定制命令)...
-python manage.py init_db excel文件路径 指定表名
# flask启动项目,像djagno一样,通过命令启动
Flask==2.2.2
Flask_Script==2.0.3
#借助于:flask-script 实现
-安装:pip3.8 install flask-script
-修改代码:
from flask_script import Manager
manager=Manager(app)
manager.run()
-用命令启动
python38 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 框架----》django orm--》只能用在django中,不能独立使用
# python界的orm框架
-peewee
-sqlalchemy:企业级
-djagno rom
-Tortoise ORM
-GINO
# go 界orm框架
-gorm 国人写的
-Xorm
# java界orm框架
-ssh 框架springmvc structs Hibernate(java的orm框架)
-ssh spring springmvc Hibernate
-ssm Spring SpringMVC MyBatis (orm框架)
-springboot :sb框架 ---》java工程师就是spring工程师
-spring cloud
# 分层:
Engine,框架的引擎
Connection Pooling ,数据库连接池
Dialect,选择连接数据库的DB API种类(sqlite,mysql...)
Schema/Types,架构和类型
SQL Exprression Language,SQL表达式语言
#操作不同数据库
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多:http://docs.sqlalchemy.org/en/latest/dialects/index.html
# 了解
orm不能创建数据库---》只能创建表,删除表---》sqlalchemy不能增加删除字段--》借助于第三方插件实现
sqlalchemy的原生操作
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
# 第一步:创建engine对象
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/cars?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第二步:通过engine获得链接
conn=engine.raw_connection()
cursor = conn.cursor()
cursor.execute(
"select * from news"
)
result = cursor.fetchall()
print(result)
cursor.close()
conn.close()
到了这里,关于flask----内置信号的使用/django的信号/ flask-script/sqlalchemy介绍和快速使用/sqlalchemy介绍和快速使用的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!