标签：tools safari name 限制 回顾 tps 类型 array 标签
Model ORM
ModelForm
amdin
大而全
浪费资源https://www.cnblogs.com/DragonFire/p/
session
小而精，三方组件全
稳定性不好
1.HttpResponse: return ""
from flask import Flask
app = Flask(__name__)
@app.route('/') # flask中的app.route()可以理解为django主url中注册的路由
def index():
return 'hello world' # return 就是HttpResponse
app.run(debug=True) # debug=true自动重启代码
2.redirect
from flask import redirect
@app.route('/')
def index():
return redirect("/login") # 重定向
@app.route("/login")
def login():
return render_template("login.html")
3.render > render_template
from flask import render_template
return render_template("login.html") # 返回模板由浏览器渲染
4.
@app.route("/json")
def json2():
dict1 = {"name": "alex"}
return dict1 # 允许返回字典格式
5.jsonify # 在响应头中加了一个Content_Type:application/json，返回json标准的字符串
@app.route("/json")
def json2():
dict1 = {"name": "alex"}
return jsonify(dict1)
6.send_file # 打开文件并自定识别文件格式，方便浏览器
from flask import send_file
@app.route('/')
def index():
return send_file('app.py') # 返回图片，文件，视频 send_file(文件路径)
7.render_template # 返回一个html页面1.request
from flask import request
@app.route("/req", methods=["post"])
def req():
print(request.method)
print(request.form)
print(request.form["user"])
print(request.form["pwd"])
print(list(request.form.keys()))
return 'ok'
# 返回结果
POST
ImmutableMultiDict([('user', 'alex'), ('pwd', 'maxhope888')])
alex
maxhope888
['user', 'pwd']
2.args # 获取url中的数据
http://127.0.0.1:5000/req1?id=1&name=alex
@app.route("/req1", methods=["get"])
def req1():
print(request.args)
# 结果ImmutableMultiDict([('id', '1'), ('name', 'alex')])
return 'ok'
3.request.values # 全部参数
<form action="/req1?id=1&age=make" method="post">
<input type="text" name="user">
<input type="text" name="pwd">
<input type="submit" value="提交">
</form>
@app.route("/req1", methods=["get", "post"])
def req1():
print(request.values) # 直接将post参数和url上的参数全部拿出来，如果url和form中的Key重名的话,form中的同名的key中value会被url中的value覆盖
return 'ok'
# 结果
CombinedMultiDict([ImmutableMultiDict([('id', '1'), ('age', 'make')]), ImmutableMultiDict([('user', 'liu'), ('pwd', '1')])])
4.cookie # cookie
print(request.cookies) # 获取
res = Response()
res.set_cookie("name", "alex") # 设置
5.request.headers
# 获取请求头信息
Connection: keep-alive
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36
Sec-Fetch-User: ?1
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9
Cookie: name=alex
If-Modified-Since: Mon, 30 Dec 2019 09:45:06 GMT
6.request.data
# 处理不了的数据就放当这个里面
7.request.files
file = request.files["file"]
8.混合
# 获取当前的url路径
print(request.path)# /req
# 当前url路径的上一级路径
print(request.script_root) #
# 当前url的全部路径
print(request.url) # http://127.0.0.1:5000/req?name=alex&age=12
# 当前url的路径的上一级全部路径
print(request.url_root ) # http://127.0.0.1:5000/
# 显示主机端口
request.host
# 加上http://的主机端口
request.host_url
# 所有args，forms的参数
request.values
# to_dict 转化为字典,用来查看数据，而不用来获取数据
request.forms.to_dict
request.args.to_dict
request.values.to_dict
# 文件
myfile = request.files["myfile"]
myfile.filename
# 请求方式
request.method
#
requst.cookies
#
request.headers
#
request.data # 如果提交时请求头中的content-type不能识别的时候，显示原始数据，bytes
#
request.json # 如果提交时请求头中的content-type:application/json,自动转化为json数据Markup == django中的mark_safe | safe
{{}} 引用变量，执行函数
{%%} 逻辑代码中执行
#
@app.template_global()
def func(a,b):
return a + b
{{func(1,2)}} # 前
#
@qpp.template_filter
def func(a,b,c):
return a + b + c
{{10|func(2,3)}} # 前
#
{% macro create_input(na,ty) %}
{{ na }}: <input type="{{ ty }}" name="na">
{% endmacro %}
{{ create_input("test", "username") }}index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>Welcome OldboyEDU</h1>
<h2>下面的内容是不一样的</h2>
{% block content %}
{% endblock %}
<h2>上面的内容是不一样的,但是下面的内容是一样的</h2>
<h1>OldboyEDU is Good</h1>
</body>
</html>
---------------------------------------另一个页面--------------------------------------------------
{% extends "index.html" %}
{% block content %}
<form>
用户名:<input type="text" name="user">
密码:<input type="text" name="pwd">
</form>
{% endblock %}flask中的session是存放在服务端的cookie中的，而且随着你存放的值越多session越长
app.secret_key = "fdsfadsfsaf" # 加密字符串，用于序列化和反序列化session信息，由于Flask中默认的session存放位置-客户端的Cookies中，所以Sessio你需要加密，用到secret_key,请求进入视图函数带上cookie将session从cookie拿出来，通过secret_key,反序列化成字典
request.session["user] = request.forms.get("username") # 设置session from flask import session
def outer(func):
def inner(*args, **kwargs):
if session.get("user", None):
ret = func(*args, **kwargs)
return ret
else:
return redirect("/login")
return inner1.from flask import Flask
app = Flask(__name__)
app.run("0.0.0.0", 5000, debug=True)
2.路由
@app.route(methods=["get", "post"])
def index():
return
3.Response三剑客
HttResponse return "三剑客"
render render_template
redirct return redrict("/路由地址")
Flask小儿子:
jsonify # from flask import json ,响应头中加Content-Type:appliction/json
send_file # from flask import send_file # 打开文件并返回文件内容，自动识别类型Content-Type:文件类型，二进制文件时背景颜色特殊
4.request
公共变量
from flask import request
request.method # 获取请求方式
request.from # 请求FormData 数据， to_dict（）转换为字典
request.args # 获取url中的数据， to_dict（）转换为字典
request.json # 请求头中的Content-Type:application/json，将数据序列化到json中
request.data # 请求头中的Content-Type:无法识别，将数据序列化到data中, 原始数据
request.files # 获取文件属性，save("文件名")
request.values # 查看form和url中的数据，不要使用to_dict会覆盖掉form中的数据
5.jinjia2
{{}}
{%%}
6.Flask中的session
公共变量
from flask import session
app.secret_key='432432@$@$'
存在cookie中的键值对（序列化后的数据），节省flask的开销，相对安全
request.session["user"]="fadsf"
if session.get("key")# ssertionError: View function mapping is overwriting an existing endpoint function: inner
解决1：
def outer(func):
@functools.wraps(func) # 这里
def inner(*args, **kwargs):
if session.get("user", None):
ret = func(*args, **kwargs)
return ret
else:
return redirect("/login")
return inner
解决2：
@app.route('/index1', endpoint="index1") # 设置不同的endpoint，这个其实就是django url()中的name方向解析参数
@outer # inner = outer(index) index=inner 执行index的时候就是执行inner
def index():
print(url_for(index))
return send_file('app.py')
# url_for
url_for("index1") # /index1@app.route('/index/<int:page>', endpoint="index1", defaults={"nid": 123}, redirect_to="login")
1.endpoint 反向生成url地址标志，默认是如函数名
2.methods 试图允许的请求方式
3.defaults={"nid": "123"} 默认参数
4.strict_slashes=True # 是否严格遵循路由规定，就是'/index1'，为True就是如果你访问的是后后面加了/就会访问不到，False(不严格)可以
5.redirect_to="/login" # 永久重定向，进入视图函数就之前进入，下面代码不执行，301
6.'/index/<int:page>' # 动态路由参数，视图中必须定义接收，def index(page):https://www.cnblogs.com/DragonFire/p/9260299.html
app = Flask(__name__, template_folder="temp", static_folder="", static_url_path="/")
1.template_folder="temp" # 指定模板路径
2.static_folder="statics" # 指定静态文件路径，当静态文件目录不是static的时候使用，只使用他一个的时候，默认/statics
3.static_url_path='/static' # 与static_folder配合使用，访问/static就是访问statics目录
# 不常用，下面
4.static_host=None # 指定静态文件服务器地址
5.host_matching = False, # 如果不是特别需要的话,慎用,否则所有的route 都需要host=""的参数,不是在对象配置中配置SERVER_NAME
6.subdomain_matching = False, # 理论上来说是用来限制SERVER_NAME子域名的,但是目前还没有感觉出来区别在哪里
7.instance_path = None, # 指向另一个Flask实例的路径
8.instance_relative_config = False # 是否加载另一个实例的配置
9.root_path = None # 主模块所在的目录的绝对路径,默认项目目录1.'DEBUG': False, # 是否开启Debug模式
app.debug=True or app.config["DEBUG"] = False
2.'TESTING': False, # 是否开启测试模式,不会重启服务
3.'SECRET_KEY': None, # 之前遇到过,在启用Session的时候,一定要有它,开启flash的时候使用
app.secret_key='fasdf' or app.config["SECRET_KEY"] = 'fasdf'
4.'PERMANENT_SESSION_LIFETIME': 31, # days , Session的生命周期(天)默认31天
5.'SESSION_COOKIE_NAME': 'session', # 在cookies中存放session加密字符串的名字
# 下面的补充要
'SESSION_COOKIE_DOMAIN': None, # 在哪个域名下会产生session记录在cookies中，多个为None
6.'JSON_AS_ASCII': True,
# 默认情况下 Flask 使用 ascii 编码来序列化对象。如果这个值被设置为 False ，
# Flask不会将其编码为 ASCII，并且按原样输出，返回它的 unicode 字符串。
# 比如 jsonfiy 会自动地采用 utf-8 来编码它然后才进行传输。
7.'JSON_SORT_KEYS': True,
#默认情况下 Flask 按照 JSON 对象的键的顺序来序来序列化它。
# 这样做是为了确保键的顺序不会受到字典的哈希种子的影响，从而返回的值每次都是一致的，不会造成无用的额外 HTTP 缓存。
# 你可以通过修改这个配置的值来覆盖默认的操作。但这是不被推荐的做法因为这个默认的行为可能会给你在性能的代价上带来改善。
app.config.from_object(FlaskDebug) # 通过对象导入配置配
class FlaskDebug(object):
DEBUG = True
SECRET_KEY = "debug"
PERMANENT_SESSION_LIFETIME = 7
SESSION_COOKIE_NAME = "debug_session"# ./myapp.view
from flask import Blueprint
bapp = Blueprint("bapp", __name__, template_folder="temp", url_prefix="/prfix")
@bapp.route("/index")
def index():
return "index"
# ./app.py
from flask import Flask
from bapp import views
app = Flask(__name__)
app.register_blueprint(views.bapp) # 注册蓝图
# app.register_blueprint(views.bapp2) # 可以有多个app
if __name__ == '__main__':
app.run()
# url_prefix="/prfix" 路由的前缀，127.0.0.1/prfix/index flash("alex", "ts1")
flash("alex2", "ts2")
get_flashed_messages(with_categories=True, category_filter=("ts1", "ts2")) # 获取多个
get_flashed_messages(category_filter="ts2") # 获取单个1.
@bapp.before_request # 请求进入之前
def before_request():
if request.path == "/login":
return None # 跳过本次验证
if session.get("username"):
return None
return redirect("/login")
# return render_template("error.html")
2.
@bapp.after_request # 响应返回客户端之前
def after_request(args): # 必须接受参数，和返回
return args
正常：be1 -> be2 -> be3 -> af3 -> af2 -> af1
异常：be1 -> af3 -> af2 -> af1
3.
@bapp.errorhandler(404) # 重定义错误页面返回信息
def error404(error_info):
return "error"
# return render_template("error.html")# 路由
1. endpoint 反向生成url ，url_for(endpoint)
2. methods 允许的请求方式，默认GET， 多重请求["GET", "POST"]
3. /index/<int:page> 动态路由匹配
4.strict_slashes=Flase 是否遵循夜歌路由匹配
5.redirect_to="/login" 永久重定向
6.defaults={"nid": 1}
def index(nind):
_________________________________________________________________________
# flask实例参数
1.template_floder="templates" 默认模板存放路径
2.static_folder="static" 默认静态文件存放目录
3.static_url_path="/static" 静态文件访问路径 "/" +static_folder
蓝图的url_prefix特殊配置url前缀
--------------------------------------------------------------------------
# False对象配置
app=Flask(__name__)
app.config.form_object(class)
-------------------------------------------------------------------------
# 蓝图
from flask import Blueprint
bapp = Blueprint("app01", __name__)
@bapp.route("/login", methods=["POST", "GET"])
def login():
# 导入
app.register_blueprint(myapp01.view)
------------------------------------------------------------------------
# 特殊装饰器
1.before_request # 在请求进入函数之前
2.after_request # 在响应返回之前
3.app.errorHanlder（code_or_excepitons） # 错误处理重定向函数
------------------------------------------------------------------------
# 三剑客
HttpResponse: return "hello"
render: return render_template("模板")
redirect: return redirect("/路由")
-------------------------------------------------------------------------
# 小儿子
jsonify: # 返回json格式的对象，响应头中加入Content——Type：application/json
send_file: # 打开文件并返回文件内容，自动识别文件类型
---------------------------------------------------------------------------
# request
request.form # 存储FormData的数据，to_dict()
request.args # 存储url中的数据,to_dict()
request.json # 请求头中的Content-Type：application/json，自动序列化到requset.json中
request.data # 不能识别Content-Type的时候，将原始数据存放在data中，bytes
request.method # 请求方式
request.files # 获取上传文件
-----------------------------------------------------------------------------
# session
from flask import session
app.secret_key = "fdsaf"
session["user"] = "name"
存放在浏览器的cookie中了，有secret-key加密之后存放
user:123 + secret_key -> response_cookie -> 拿到cookie存储
请求带上cookie-flaskfrom flask import Flask, views
app = Flask(__name__)
class Login(views.MethodView):
def get(self):
return "hello"
def post(self):
return "post"
app.add_url_rule("/", view_func=Login.as_view("login")) # 路由注册Http连接，请求+响应=断开
服务器不知道你是谁-http session -无记忆无状态
http 聊天室
用户1 - 用户1http - 服务器转发给用户1 - 传达室 - 用户2区传达室 - 用户1写的信 - 收信
轮询：
用户1不定的传达室 - 传达室 - 一秒100次
开启告诉公路（带宽） - 占用网路带宽
前后端一秒交互多次
前后端压力极大
并且带宽消耗
资源浪费极其严重
长轮询：
在传达室安排几个哨兵
向服务器保存我的一个连接状态，用于快速传递消息
节省带宽，释放压力
数据实时性
长连接：
websocket
用户1安电话了，并且把电话号码方在传达室了
服务端及客户端节省极大地资源
能保证数据实时性
带宽：
同时可以跑多少 1Mbp/s = 128kb/s
拨号调制调节器 - 25kbp/s = 3 - 4 kb/s
ISDN = 2电话 - 56kbp/s = 10kbp/s
ADSL = 128kbp/s = 25kbp/s
256
512
1
2
10Mbp/s = 2Mbp/s
VDSL 光纤 贵非常贵# 简单使用
from flask import Flask, session
from flask_session import Session
from redis import Redis
app = Flask(__name__)
app.config["SESSION_TYPE"] = "redis"
app.config["SESSION_REDIS"] = Redis(host="127.0.0.1", port=6379, db=6)
Session(app) # 这个必须在上面的配置下面，否则会让你使用secret_key
@app.route("/")
def index():
session["user"] = "alex"
return "index"
if __name__ == '__main__':
app.run()
-------------------------------------------------------------------------------------
其中Session是如何替换，Flask的原生的session，
class Session(object):
def __init__(self, app=None):
self.app = app
if app is not None:
self.init_app(app)
def init_app(self, app):
app.session_interface = self._get_interface(app) # 这里直接将其替换成了，第三方组件的Sessionhttps://www.cnblogs.com/wdliu/p/10183645.html # 这个博客里面有基本用法
from flask import Flask, render_template, request, session
from wtforms.fields import simple, core
from wtforms import validators, widgets
from wtforms import Form
app = Flask(__name__)
class LoginForm(Form):
username = simple.StringField(
label="用户名", # 描述标记
validators=[
validators.data_required(message="用户名不能为空"),
validators.length(min=8, max=15, message="最小8位，最多15位"),
], # 验证器
description='', # 描述标记
id="user_name", # 标签id
default=None, # 默认值
widget=None, # 默认（input type = "text"）
render_kw={"class": "my_user"} # {"class": "my_login"}
)
password = simple.PasswordField(
label="密码", # 描述标记
validators=[
validators.data_required(message="密码不能为空"),
validators.length(min=8, max=15, message="最小8位，最多15位"),
validators.email(message="密码必须符合邮箱规则")
], # 验证器
description='', # 描述标记
id="pwd", # 标签id
default=None, # 默认值
widget=None, # 默认（input type = "text"）
render_kw={"class": "pws"} # {"class": "my_login"}
)
class RegForm(Form):
username = simple.StringField(
label="用户名", # 描述标记
validators=[
validators.data_required(message="用户名不能为空"),
validators.length(min=8, max=15, message="最小8位，最多15位"),
], # 验证器
)
password = simple.PasswordField(
label="密码", # 描述标记
validators=[
validators.data_required(message="密码不能为空"),
], # 验证器
)
re_password = simple.PasswordField(
label="确认密码", # 描述标记
validators=[
validators.equal_to(message="两次密码不一致", fieldname="password")
], # 验证器
)
gender = core.SelectField(
label="性别",
coerce=int,
choices=(
(1, "女"),
(0, "男"),
),
default=1,
)
hobby = core.SelectMultipleField(
label="爱好",
validators=[
validators.length(min=2, max=4, message="最多4个，最少2个")
],
coerce=int,
choices=(
(1, "foo"),
(2, "apple"),
(3, "banana"),
(4, "basket"),
(5, "running"),
(6, "eating"),
),
default=(1, 3, 5), # 有时候就算你更新了代码，页面显示还是会出问题，所以多刷新
)
s = simple.SubmitField(label="提交")
class Meta:
csrf = True
csrf_field_name = "csrf_token"
csrf_secret = "fda#13!@"
csrf_context = {"session": "fdsa"}
@app.route(rule="/reg", methods=["GET", "POST"])
def reg():
if request.method == "GET":
reg = RegForm()
return render_template("register.html", **locals())
else:
reg = RegForm(request.form)
if reg.validate():
return reg.data.get("password")
else:
return render_template("register.html", **locals())
@app.route(rule="/", methods=["GET", "POST"])
def index():
if request.method == "GET":
wtf = LoginForm()
return render_template("wtform.html", **locals())
else:
wtf = LoginForm(request.form)
if wtf.validate():
return wtf.data.get("password")
else:
return render_template("wtform.html", **locals())
if __name__ == '__main__':
app.run(debug=True)# 正常方式
import pymysql
mysql = pymysql.connect(
host="127.0.0.1",
port=3306,
user="root",
password="1",
db="myapp001",
charset="utf8"
)
c = mysql.cursor(cursor=pymysql.cursors.DictCursor)
# c = mysql.cursor()
sql1 = "select * from user where name='赵'"
c.execute(sql1)
data = c.fetchall()
c.close()
mysql.close()
--------------------------------------------------------------------------------------
# 连接池方式
import pymysql
from DBUtils.PooledDB import PooledDB, SharedDBConnection
Pool = PooledDB(
creator=pymysql,
maxconnections=20,
mincached=2, # 初始化，连接池至少创建的空闲连接，0标识不创建
maxcached=5, # 连接池中最多的限制连接，0和None不限制
maxshared=3, # 最大共享连接数
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待，True，等待，False,不等待
maxusage=None, # 一个连接最多被重复使用的次数，None标识无限，
setsession=[], # 开始会话前执行的命令连接，如：["set datestyle to"]["set time Zoon"]
ping=0,
# ping mysql服务器，检查时候服务可用，
# 如：
# 0=None=Never, # 永远不ping
# 1=default=whenever it is requested 在第一次链接的时候ping一次
# 2 = where a cursor is created 当cursor创建的时候ping一次数据库
# 4 = when a query is executed 当查询语句被执行的时候
# 7 = always 每查询一次，ping一次
host="127.0.0.1",
port=3306,
user="root",
password="1",
db="myapp001",
charset="utf8"
)
pool = Pool.connection() # pymysql - conn
c = pool.cursor(cursor=pymysql.cursors.DictCursor)
sql1 = "select * from user where name='赵'"
c.execute(sql1)
data = c.fetchall()
print(data)
--------------------------------------------------------------------------------------
# 使用
import pymysql
from a6 import Pool
def create_conn():
conn = Pool.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
return conn, cursor
def close_conn(conn, cursor):
cursor.close()
conn.close()
def insert(sql, args):
conn, cursor = create_conn()
print(args)
res = cursor.execute(sql, args)
conn.commit()
close_conn(conn, cursor)
return res
def fetch_one(sql, args):
conn, cursor = create_conn()
print(args)
res = cursor.execute(sql, args)
close_conn(conn, cursor)
return res
sql = 'insert into user(name, age, gender, addr, height) values (%s, %s, %s, %s, %s)'
# 关于这个sql的使用里面的字段不需要加引号，它会自己转成字符串，%s也一样，如果你加上了可能会报错
# insert(sql, ("make", "22", "女", "汗水", 180))
sql = "select * from user where name=%s and age=%s"
fetch_one(sql, ("make", 22))https://www.cnblogs.com/DragonFire/tag/websocket/
import socket, base64, hashlib
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 8000))
sock.listen(5)
# 获取客户端socket对象
conn, address = sock.accept()
# 获取客户端的【握手】信息
data = conn.recv(1024)
print(data)
"""
b'GET /ws HTTP/1.1\r\n
Host: 127.0.0.1:9527\r\n
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:62.0) Gecko/20100101 Firefox/62.0\r\n
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n
Accept-Language: zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2\r\n
Accept-Encoding: gzip, deflate\r\n
Sec-WebSocket-Version: 13\r\n
Origin: http://localhost:63342\r\n
Sec-WebSocket-Extensions: permessage-deflate\r\n
Sec-WebSocket-Key: jocLOLLq1BQWp0aZgEWL5A==\r\n # 浏览器给服务端发送一个key值，服务端通过加密之后传输给浏览器，浏览器进行比较对就通信，不对就拒绝通信，需要这个东西
Cookie: session=6f2bab18-2dc4-426a-8f06-de22909b967b\r\n
Connection: keep-alive, Upgrade\r\n
Pragma: no-cache\r\n
Cache-Control: no-cache\r\n
Upgrade: websocket\r\n\r\n'
"""
# magic string为：258EAFA5-E914-47DA-95CA-C5AB0DC85B11
magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
def get_headers(data):
header_dict = {}
header_str = data.decode("utf8")
for i in header_str.split("\r\n"):
if str(i).startswith("Sec-WebSocket-Key"):
header_dict["Sec-WebSocket-Key"] = i.split(":")[1].strip()
return header_dict # {"Sec-WebSocket-Key": "jocLOLLq1BQWp0aZgEWL5A=="}
headers = get_headers(data) # 提取请求头信息
# # 对请求头中的sec-websocket-key进行加密
response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" "Upgrade:websocket\r\n" "Connection: Upgrade\r\n" "Sec-WebSocket-Accept: %s\r\n" "WebSocket-Location: ws://127.0.0.1:8000\r\n\r\n"
#
value = headers['Sec-WebSocket-Key'] + magic_string
ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())
print(ac.decode("utf-8"))
response_str = response_tpl % (ac.decode('utf-8'))
# # 响应【握手】信息
conn.send(response_str.encode("utf8"))
while True:
msg = conn.recv(8096)
print(msg.decode("utf-8"))<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
</body>
</html>
<script>
var ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onmessage = function (data) {
console.log(data.data)
};
</script>websocket解密：
b'\x81\x86\x02\x16\x81\x91\xe6\xab!t\xa7\xab' # 你好
b'\x81\x85[\x98C/3\xfd/C4' # hello
b'\x81\x89\xe7\xbd\xe5\xaa\x015tMo\x0c\x01\x17G' # 我爱你
# b'\x81\x83\xceH\xb6\x85\xffz\x85'
hashstr = b'\x81\x89\xe7\xbd\xe5\xaa\x015tMo\x0c\x01\x17G'
# 将第二个字节也就是 \x83 第9-16位 进行与127进行位运算
payload = hashstr[1] & 127
print(payload)
if payload == 127:
extend_payload_len = hashstr[2:10]
mask = hashstr[10:14]
decoded = hashstr[14:]
# 当位运算结果等于127时,则第3-10个字节为数据长度
# 第11-14字节为mask 解密所需字符串
# 则数据为第15字节至结尾
if payload == 126:
extend_payload_len = hashstr[2:4]
mask = hashstr[4:8]
decoded = hashstr[8:]
# 当位运算结果等于126时,则第3-4个字节为数据长度
# 第5-8字节为mask 解密所需字符串
# 则数据为第9字节至结尾
if payload <= 125:
extend_payload_len = None
mask = hashstr[2:6]
decoded = hashstr[6:]
# 当位运算结果小于等于125时,则这个数字就是数据的长度
# 第3-6字节为mask 解密所需字符串
# 则数据为第7字节至结尾
str_byte = bytearray()
for i in range(len(decoded)):
byte = decoded[i] ^ mask[i % 4]
str_byte.append(byte)
print(str_byte.decode("utf8"))websocket加密：
加密出错源码阅读：
# 请求上文
1. app.run()
只看run_simple(host, port, self, **options)这一句，说实话其他的看不懂
而且看到这里run就别向下了,这里的self其实就是app=Flask(__name__)
2.app创建去了我们猜测他要执行app(),执行它的__call__方法
app.__call__()
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response) # 就执行了这一句
3.self.wsgi_app
ctx = self.request_context(environ) # 这一句给ctx赋值，返回个ResquestContent类
def request_context(self, environ)：
return RequestContext(self, environ)
4.初始化request和session，在RequestContext中
def __init__(self, app, environ, request=None, session=None)
self.app = app # 就是传进来的self=app=Flask()
if request is None:
request = app.request_class(environ) # 这个时候就是一个Resquest()类
self.request = request # 这里给request赋值
self.session = session # 这里对session赋值
5.继续执行ctx=RequestContext() -- request,session
ctx.push() # 看这里
# push中的
top = _request_ctx_stack.top
_request_ctx_stack = _request_ctx_stack = LocalStack() = \
class LocalStack(object):
def __init__(self):
self._local = Local() # 他里面有这么一个属性，一个类
class Local(object):
__slots__ = ("__storage__", "__ident_func__")
def __init__(self): # {__storage__:{},__ident_func__: get_index }
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
# 最后的结果就是：
_request_ctx_stack = LocalStack() = {_local:{__storage__:{},__ident_func__: get_index }}
6.现在来看top方法
_request_ctx_stack.top= LocalStack().top
@property
def top(self):
try:
return self._local.stack[-1] # 这里.statck，self._local去看看__getattr__方法
except (AttributeError, IndexError):
return None
7.self._local的__getattr__方法
def __getattr__(self, name): # name=stack， self=LocalStack()
try:
return self.__storage__[self.__ident_func__()][name] # self.__storage__中压根就没有一个线程id叫做stack,self.__storage__[线程id][stack],返回None
except KeyError:
raise AttributeError(name)
8.直接看push函数中这一句，中间的跳过，因为根本看不懂
_request_ctx_stack.push(self) # self=ctx=RequestContext() -- request,session
def push(self, obj):
rv = getattr(self._local, "stack", None) # 刚才就说了拿不到，直接None
if rv is None:
self._local.stack = rv = [] # k这里进行复制操作看，self._local的__setattr__方法
9.self._local的__setattr__方法
def __setattr__(self, name, value): # name=stack, value=[]
ident = self.__ident_func__() # 拿到一个id号
storage = self.__storage__
try:
storage[ident][name] = value # 赋值{__storage__:{9527:{stack:[ctx]}},__ident_func__: get_index }
except KeyError:
storage[ident] = {name: value}
10.接着看下面的代码
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
self._local.stack = rv = []
rv.append(obj) # obj=obj = self = ctx = ResquestContent() # 第一次request进来到这里就没了session先不看
return rv
# 应用上下文
1.request.methods
request = LocalProxy(partial(_lookup_req_object, "request"))
我们分开来看
2.partial(_lookup_req_object, "request")
def _lookup_req_object(name): # name = reuqest
top = _request_ctx_stack.top # 可以看到在执行这一句的是偶其实就是获取每个进程对应的ctx，这个在上文之中过再看一遍
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name) # 获取到每个ctx的request
3.
@property
def Xtop(self):
try:
return self._local.stack[-1] # __getattr__方法
except (AttributeError, IndexError):
return None
4.
def __getattr__(self, name): # self._local.stack[-1]
try:
return self.__storage__[self.__ident_func__()][name] # self.__storage__[9527][stack]，通过每个进程的编号拿取对应的ctx对象
except KeyError:
raise AttributeError(name)
# 然后再第二部中的最后一步获取到，requset
5.request = LocalProxy(partial(_lookup_req_object, "request"))
# 然后继续看最后的一层代理LocalProxy
def __init__(self, local, name=None):
object.__setattr__(self, "_LocalProxy__local", local) # local = request
object.__setattr__(self, "__name__", name)
if callable(local) and not hasattr(local, "__release_local__"):
object.__setattr__(self, "__wrapped__", local)
6.这个时候request.methods其实就是触发LocalProxy的__getarrt__方法
def __getattr__(self, name): # request.methods name=methods
if name == "__members__":
return dir(self._get_current_object())
return getattr(self._get_current_object(), name) # name=methods
7._get_current_object
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
return self.__local() # 这个__local里面存放就是拿取到的request偏函数，加上（）执行 得到的就是request,返回
8.
# 然后你再第六步中俄最后还可以看到getattr方法其实就是拿到属性，最后返回Tornado:
异步 IO非阻塞 原生websocket
干净Sanic:
资料太少计算通过网段，计算可用地址范围
192.168.130.12/24
255.255.255.0
11111111 11111111 11111111 00000000 上面的24就是前面1的位数
192.168.130.1 1，255 网关 0 用不了
192.168.130.1 - 192.168.130.254 可用地址范围
-------------------------------------------------------------------------------
192.168.130.12/16
11111111 11111111 00000000 00000000
192.168.0.0
192.168.0.0 - 192.168.255.255
-------------------------------------------------------------------------------
192.168.130.12/17
11111111 11111111 10000000 00000000
子网掩码： 255.255.128.0
192.168.0.0 - 192.168.127.255
-------------------------------------------------------------------------------
192.168.130.12/18
11111111 11111111 11000000 00000000
子网掩码： 255.255.192.0
192.168.0.0 - 192.168.64.255
-------------------------------------------------------------------------------
192.168.130.12/19
11111111 11111111 11100000 00000000
子网掩码： 255.255.224.0
192.168.0.0 - 192.168.31.255
-------------------------------------------------------------------------------
192.168.130.12/30
11111111 11111111 11111111 111111100
子网掩码： 255.255.255.252
192.168.130.0 - 192.168.130.3
-------------------------------------------------------------------------------
192.168.130.12/31
11111111 11111111 11111111 111111110
子网掩码： 255.255.255.254
192.168.130.0 - 192.168.130.1
-------------------------------------------------------------------------------
如何使用python：
192.168.130.12/22A: 255.0.0.0 10.0.0.0 - 10.254.254.254
B: 255.255.0.0 172.10.0.0. - 172.10.254.254
C: 255.255.255.0 168.192.11.0 - 168.192.11.254 class Foo():
def __call__(self, *args, **kwargs):
return 666
def __setattr__(self, key, value):
return f"{value}:{key}"
def __setitem__(self, key, value):
return f"{value}:{key}"
foo = Foo()
print(foo()) # 属性加（）执行__call__方法
偏函数：
# django model中可以使用
from functools import partial
def ab(a, b):
return a + b
par_ab = partial(ab, 1)
print(par_ab(2))
例子：设置默认值时使用偏函数，保证与这次的数据的创建时间一致，如果直接使用函数，他会在你们次访加载模型的时候就执行，和数据创建的时间不一致。更有可能的是你第一访问的是登录页面，但是它也回加载model，在你过了两个小时之后的此创建数据的时候使用还是几个小时之前的创建。在时间上表现得比特明显。
cid = models.CharField(
max_length=32,
unique=True,
default=functools.partial(get_random_string, 12),
verbose_name="cid",
help_text="cid"
)
key = models.CharField(
max_length=64,
default=functools.partial(get_random_string, 30),
verbose_name="key",
help_text="key"
)import time
from threading import local
class Foo(local):
pass
foo = Foo()
foo.num = 1
def add(i):
foo.num = i
time.sleep(1)
print(foo.num, i, threading.current_thread().ident)
for i in range(10):
t = threading.Thread(target=add, args=(i,))
t.start()
执行结果：
0 0 2248
3 3 11888
1 1 4248
2 2 13620
4 4 16184
6 6 1384
7 7 13828
5 5 6068
9 9 17252
8 8 20388
# 在一开始开启进程的时候，结果只有最后一个是对的，但是当时用了local之后，就正确了，保证了线程安全，大概解释一些，他会将每个线程的当前id和值存放在一个字典当中，每次执行的时候还是按照线程生成的时间前后来执行，但是计算机执行速度很快一秒内，二十个线程都执行完成了，将值存进对应的线程id字典中，因为有了线程id所以保证每一个线程每次拿到的结果是自己的。标签：tools safari name 限制 回顾 tps 类型 array 标签
原文地址：https://www.cnblogs.com/liuzhanghao/p/12157413.html
