Flask-Session
from flask import Flask,views,request,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)
Session(app)
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
session["user"] = "123"
return "post"
return "get"
if __name__ == '__main__':
app.run(debug=True) View Code
WTForms
from flask import Flask, request, render_template
from wtforms import Form, validators, widgets
from wtforms.fields import simple, core
app = Flask(__name__)
class RegForm(Form):
username = simple.StringField(
label="使用者名",
validators=[
validators.DataRequired(message="使用者名不能為空"),
],
render_kw={"class": "my_class"}
)
pwd = simple.PasswordField(
label="密碼",
validators=[
validators.DataRequired(message="密碼不能為空"),
validators.length(min=6, max=16, message="長度必須大于等于6，小于等于16")
],
render_kw={"class": "form-control"},
widget = widgets.PasswordInput(),
)
pwd_confim = simple.PasswordField(
label="确認密碼",
validators=[
validators.EqualTo("pwd", message="兩次密碼不一緻"),
validators.length(min=6, max=16, message="長度必須大于等于6，小于等于16")
],
)
gender = core.SelectField(
label="性别",
choices=(
(1, "女"),
(2, "男")
),
default=1,
coerce=int # 限制是int類型的
)
city = core.RadioField(
label="城市",
choices=(
("1", "北京"),
("2", "上海")
),
default=2,
coerce=str
)
email = simple.StringField(
label="郵箱",
validators=[
validators.DataRequired(message="郵箱不能為空"),
validators.Email(message="格式不對")
]
)
hobby = core.SelectMultipleField(
label="愛好",
choices=(
(1, "電影"),
(2, "音樂"),
(3, "畫畫"),
(4, "看書")
),
coerce=int,
default=(1, 4)
)
favor = core.SelectMultipleField(
label="喜好",
choices=(
(1, '籃球'),
(2, '足球'),
),
widget=widgets.ListWidget(prefix_label=False),
option_widget=widgets.CheckboxInput(),
coerce=int,
default=[1, 2]
)
def __init__(self,*args,**kwargs): #這裡的self是一個RegisterForm對象
'''重寫__init__方法'''
super().__init__(*args, **kwargs) #繼承父類的init方法
self.favor.choices =((1, '籃球'), (2, '足球'), (3, '羽毛球')) #吧RegisterForm這個類裡面的favor重新指派
def validate_pwd_confim(self,field,):
'''
自定義pwd_config字段規則，例：與pwd字段是否一緻
:param field:
:return:
'''
# 最開始初始化時，self.data中已經有所有的值
if field.data != self.data['pwd']:
# raise validators.ValidationError("密碼不一緻") # 繼續後續驗證
raise validators.StopValidation("密碼不一緻") # 不再繼續後續驗證
class LoginForm(Form):
username = simple.StringField(
label="使用者名",
validators=[
validators.DataRequired(message="使用者名不能為空"),
validators.length(min=4, max=8, message="長度必須大于等于4，小于等于8")
]
)
password = simple.PasswordField(
label="密碼",
validators=[
validators.DataRequired(message="密碼不能為空"),
validators.length(min=4, max=8, message="長度必須大于等于4，小于等于8")
]
)
@app.route('/register',methods=["GET","POST"])
def register():
if request.method=="GET":
form = RegForm(data={'gender': 1}) #預設是1,
return render_template("register.html",form=form)
else:
form = RegForm(formdata=request.form)
if form.validate(): #判斷是否驗證成功
print('使用者送出資料通過格式驗證，送出的值為：', form.data) #所有的正确資訊
else:
print(form.errors) #所有的錯誤資訊
return render_template('register.html', form=form)
@app.route("/reg", methods=["GET", "POST"])
def reg():
if request.method == "GET":
rf = RegForm()
return render_template("reg.html", rf=rf)
else:
rf_data = RegForm(request.form)
if rf_data.validate():
print(rf_data.data)
return "OK"
else:
return render_template("reg.html", rf=rf_data)
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "GET":
login_form = LoginForm()
return render_template("login.html", lf=login_form)
else:
login_form_data = LoginForm(request.form)
if login_form_data.validate():
user = login_form_data.data.get("username")
return str(user)
else:
return render_template("login.html", lf=login_form_data)
if __name__ == '__main__':
app.run(debug=True) View Code
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post" novalidate>
{% for field in rf %}
<p>{{ field.label }}{{ field }}{{ field.errors[0] }}</p>
{% endfor %}
<input type="submit" value="submit">
</form>
<form action="" method="post" novalidate>
<p>{{ lf.username.label }}{{ lf.username }}{{ lf.username.errors.0 }}</p>
<p>{{ lf.password.label }}{{ lf.password }}{{ lf.password.errors[0] }}</p>
<input type="submit" value="submit">
</form>
</body>
</body>
</html> View Code
Flask-請求上下文
# 偏函數partial
from functools import partial
def my_sum(a, b):
print(a)
print(b)
return a + b
# new_my_sum = partial(my_sum, 10)
# new_my_sum此時就是下面這樣的
# def my_sum(b):
# a=10
new_my_sum = partial(my_sum,
b=10) # 位置參數一定要在關鍵字參數前面 不能寫成new_my_sum = partial(my_sum, a=10),如果非要這樣寫，下面就改成res = new_my_sum(b=20)
# new_my_sum此時就是下面這樣的
# def my_sum(b):
# b=10
res = new_my_sum(20)
print(res)
class My:
def __call__(self, *args, **kwargs):
print(666, '__call__')
def __setattr__(self, key, value):
print(key, value, '__setattr__')
def __getattr__(self, item):
print(item, '__getattr__')
def __setitem__(self, key, value):
print(key, value, '__setitem__')
def __getitem__(self, item):
print(item, '__getitem__')
a = My()
a() # 執行__call__
a.name = 'tom' # 執行__setattr__
a['age'] = 18 # __setitem__
a['name'] # __getitem__
a.qweqwewqe # __getattr__
# 空間換時間
# 線程安全 Flask上下文機制就是使用的Threading.local
# 線程進來，開辟一個空間給這個線程，線程操作的所有任務，都會複制一份到空間中
from threading import local
class Foo(local):
pass
print('##############################')
# 棧Stack
class Mystack(object):
data = []
def __setattr__(self, key, value):
self.push(key, value)
def push(self, key, value):
self.data.append({key:value})
def top(self):
return self.data.pop()
my_stack = Mystack()
my_stack.name = 'tom'
my_stack.name = 'rose'
print(my_stack.data) #[{'name': 'tom'}, {'name': 'rose'}]
#LocalStack
# import threading
# threading.current_thread().ident #目前線程的ID
import threading
from threading import get_ident #和上面是一樣的
class MyLocalStack(object):
data=[]
storage={}
def push(self, item):
try:
self.storage[get_ident()].append(item)
except:
self.storage[get_ident()]=[item]
def top(self):
return self.storage[get_ident()].pop()
my_local_stack=MyLocalStack()
import time
def go(i):
my_local_stack.push(i)
time.sleep(1)
# my_local_stack.top()
for i in range(5):
t = threading.Thread(target=go,args=(i,))
t.start()
print(my_local_stack.storage)
###########
from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple
@Request.application
def app(req):
print(req)
return Response("200ok")
run_simple("0.0.0.0", 9527, app)
"""
run_simple("0.0.0.0",9527,app) 會執行app函數
from flask import Flask
app=Flask(__name__)
if __name__ == '__main__':
app.run()
""" 鋪墊
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from flask import Flask,request
app=Flask(__name__)
if __name__ == '__main__':
app.run()
#run_simple(host, port, self, **options)
#app()=obj()=obj.__call__
#app.wsgi_app
"""
上文
self._local=={"__storage__":{},"__ident_func__":get_ident}
ctx = requestcontext對象 此對象中有 request/session
此時 self._local=={"__storage__":{3721:{"stack":[ctx] }},"__ident_func__":get_ident}
_request_ctx_stack == LocalStack()==self._local=={"__storage__":{3721:{"stack":[ctx] }},"__ident_func__":get_ident}
"""
"""
下文源碼
request = LocalProxy(partial(_lookup_req_object, "request")#傳回的是request對象)
class LocalProxy(object):
__slots__ = ("__local", "__dict__", "__name__", "__wrapped__")
def __init__(self, local, name=None): ####local==request偏函數==partial(_lookup_req_object, "request")#傳回的是request對象
object.__setattr__(self, "_LocalProxy__local", local) #__local=local=request偏函數
object.__setattr__(self, "__name__", name)
if callable(local) and not hasattr(local, "__release_local__"):
object.__setattr__(self, "__wrapped__", local)
def __getattr__(self, name): ##request.method執行的就是這
if name == "__members__":
return dir(self._get_current_object())
return getattr(self._get_current_object(), name) #從request對象中取出 name ，name=method
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
return self.__local() ##request偏函數的執行，傳回的是request對象
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
def _lookup_req_object(name): #name==request
top = _request_ctx_stack.top #top==ctx ctx = requestcontext對象 此對象中有 request/session
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name) #傳回的是request對象 ，top是ctx name是request
@property
def top(self):
try:
return self._local.stack[-1] ##
except (AttributeError, IndexError):
return None
"""
""" 上文源碼
def run(self, host=None, port=None, debug=None, load_dotenv=True, **options): # self==app==Flask
from werkzeug.serving import run_simple
try:
run_simple(host, port, self, **options) #相當于執行self() 也就是 Flask.__call__() # self==app==Flask
Flask.__call__()
def __call__(self, environ, start_response): #environ==request
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ) # self==app==Flask
## ctx = requestcontext對象 此對象中有 request/session ##
error = None
try:
try:
ctx.push()### 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
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
def push(self): ##self==ctx
top = _request_ctx_stack.top ###_request_ctx_stack == LocalStack()=={"__storage__":{},"__ident_func__":get_ident}
if top is not None and top.preserved:
top.pop(top._preserved_exc)
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, "exc_clear"):
sys.exc_clear()
_request_ctx_stack.push(self)######### self==ctx ctx = requestcontext對象 此對象中有 request/session
###{"__storage__":{3721:{"stack":[ctx] }},"__ident_func__":get_ident}### 上文
def push(self, obj): ######### obj就是ctx self==_request_ctx_stack == LocalStack()=={"__storage__":{},"__ident_func__":get_ident}
rv = getattr(self._local, "stack", None) ##此時 self._local=={"__storage__":{},"__ident_func__":get_ident}
if rv is None:
self._local.stack = rv = [] ##此時 self.local=={"__storage__":{3721:{"stack":rv=[] }},"__ident_func__":get_ident}
rv.append(obj) ##obj==ctx ctx = requestcontext對象 此對象中有 request/session 此時 {"__storage__":{3721:{"stack":[ctx] }},"__ident_func__":get_ident}
return rv ##rv==ctx
class Local(object):
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
class LocalStack(object):
def __init__(self):
self._local = Local()
class Local(object):
__slots__ = ("__storage__", "__ident_func__")
def __init__(self):
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
def request_context(self, environ):
return RequestContext(self, environ) # self==app==Flask
class RequestContext(object):
def __init__(self, app, environ, request=None, session=None): #self==requestcontext對象 app==app==flask
self.app = app
if request is None:
request = app.request_class(environ) #request_class = Request
self.request = request #######
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session
class Request(RequestBase, JSONMixin):
pass
class BaseRequest(object):
def __init__(self, environ, populate_request=True, shallow=False):
self.environ = environ
if populate_request and not shallow:
self.environ["werkzeug.request"] = self
self.shallow = shallow
""" View Code
Flask-SQLAlchemy
Flask-Script
Flask-Migrate
轉載于:https://www.cnblogs.com/bubu99/p/11229544.html
![【MySQL資料庫】資料庫索引事務1.索引2.事務[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)