Flask

werkzurg

#相似於django wsgiref
from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple

#實例一
def run(environ,start_response):
    return [b"asdfasdf"]

if __name__ == '__main__':
    run_simple('localhost', 4000, run)  #請求進來了會加括號執行第三個參數 若是是對象的話 會觸發__call__

    

 

示例二：
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)		    
		

 flask

from flask import Flask
app = Flask(__name__)

@app.route('/index')
def index():
    return "Hello World"

if __name__ == '__main__':  #防止被導入時候也被運行
    app.run()

 

from flask import Flask，render_template
app = Flask(__name__,template_folder='temlates') #在這改的模板文件

@app.route('/index')
def index():
    return render_template("login.html")  #默認在temlates下找 ，能夠在上面修改template默認文件夾

if __name__ == '__main__':  #防止被導入時候也被運行
    app.run()

登陸實例

from flask import Flask,render_template,request,redirect,session
app = Flask(__name__,template_folder="templates")

@app.route('/login',methods=["GET","POST"]) #容許提交的method  默認是Get
def login():
    if request.method == 'GET':
        return render_template('login.html')
    user = request.form.get('user')     #這個request必須導才能使用（上下文管理）form 是post 數據 
    pwd = request.form.get('pwd')       #get 數據在request.args 
    if user == 'oldboy' and pwd == '666':
        session['user'] = user
        return redirect('/index')  #重跳轉
    return render_template('login.html',error='用戶名或密碼錯誤')  #後端能夠拿到這個error ｛｛error｝｝
    # return render_template('login.html',**{"error":'用戶名或密碼錯誤'})

@app.route('/index')
def index():
    user = session.get('user')
    if not user:
        return redirect('/login')
    return render_template('index.html')

if __name__ == '__main__':
    app.run()
    

 模板傳遞 

{{ error}}    

傳遞的時候return  error='sss'    **{error:'ss'} #django 會自動加（）   

<img src='ssss/tupia.png'>   #就算有這個sss 也找不到  sss是前綴 static_url_path='/ssss' #能夠修改

app=Flask(static_folder='static') #默認是static下找圖片 而圖片要的是前綴是ssss的 

登陸認證

from flask import Flask，session

session["user"]=user
session["pwd"]=pwd

往cookie裏存 session實際上加密的cookie裏

app.secret_key='鹽' #加鹽

配置文件使用

給你一個路徑 字符串的 能夠找到類和類的靜態字段
實例 略#

cookie知道Key 嗎？ 咱們不知道 實際上是內部作了 因此咱們能session["key"]=v

app.config #在這裏
怎麼修改呢？
app.config.from_object('配置文件路徑')  #好比到線上的配置就是Pro

class Dev（）： #表示開發環境
    kEY=VALUES #小寫的沒用
　　 DEVUG=Flase

class Pro（）： #表示線上環境
    kEY=VALUES #小寫的沒用
　　 DEBUG=True

app.config.from_object('setting.Dev')  #好比到DEV

路由系統

@app.route('/index',method=['GET',"POST"])
def  index():
    return 'index'


起別名
from filask import url_from
@app.route('/index',method=['GET',"POST"],endpoint='n1') #重命名  不寫默認是函數名
def  index():
    url_from("n1")  #能夠經過url_form  拿到真正的url   
    return 'index'

　　

 

from filask import url_from
@app.route('/index/<int:nid>',method=['GET',"POST"],endpoint='n1') #不寫默認是函數名
def  index(nid):
    url_from("n1")  #能夠經過n1 拿到真正的url   
    return 'index'


#支持的有  str  any  path  int float  uuid   #默認 不支持正則

@app.route('/index/<int:nid>',method=['GET',"POST"],endpoint='n1') 
def  index(nid):
    url_from("n1",nid=nid)  #匹配完整url
    return 'index'

請求和相應相關-視圖

request 的參數



return的參數

#返回相應體
字符串
render_template()
redicrect  
jsonify  #內部自動序列化 


#返回字符串怎麼返回響應體
obj=make_response(''str')
obj.headers['xxxx']='123' #頭裏加東西
obj.set_cookie(key,value) #設置cookie
return obj 

 實例 股票管理

from flask import Flask,render_template,request,redirect,session
app = Flask(__name__,template_folder="templates")
app.config.from_object('配置文件路徑') 
STUDENT_DICT={
 1:{'name':'xzq'},
 2:{'ss':'ss'}
}
@app.route('/index/<int:nid>',method=['GET',"POST"],endpoint='n1') 
def  index(nid):
    url_from("n1",nid=nid)  #匹配完整url
    return render_template('dd.html',student_dic=STUDENT_DICT)

模板

tr #一行
th 表題
td       普通內容
tbody  內容體
theader 


{% for  k,v  in  stu_dic.items() %}
<tr>
<td>{{  v['name'] }}</td>  # v.name 也能拿到  v.get('name'，默認值) 
</tr>

<a href='del/{{k}}'>  #刪除

{%  endfor  %}

　

 

裝飾器的坑
若是使用裝飾器 若是再用反向解析 就不知道哪一個是哪一個了
由於全部的view都是一個inner

裝飾器導入順序
從下到下

　　

裝飾器不適合批量

應用場景：比較少的


第三版本：before_request  #用於執行真正視圖函數以前 進行一個攔截 進行判斷
@app.before_require
def xxxx():
    xxx   


@app.before_require
def xxxx():
    xxx   
    #return   xxx    #若是返回當即返回不走後面的視圖    

頁面看到的就是 xxx

　　

 模板

{{ list.0 }}  #索引取值方式一
{{ list[0] }} #索引取值方式二
{{ text }}  #若是傳遞過來的是html   仍是文本
 {{ text|safe }}  #若是傳遞過來的是html   仍是文本

#Python 用Markup  至關於makesafe

{{ func}}  #函數地址
{{ func（7）}}  #返回值


默認返回全部值
@app.template_global()
def sb(a1,a2):
    return a1+a2


@app.template_filter():
def db(a1,a2,a3):
    return a1+a2+a3
{{1|db(2,3)}}  #這個返回值能夠對其if判斷


模板繼承
{% extens 'layout.html' %}
{% end%}


定義宏  至關於定義函數
  {% macro ccc(name,type='text',value='')%}
      <h1>宏</h1>
      <input type="{{type}}" name="{{name}}" value="{{value}}">
      <input type="submit" value="提交">
  {% endmacro %}


  {{ ccc('n1') }}
  {{ ccc('n2') }}


基本數據類型  能夠執行Python語法　　

session

請求剛進來就建立一個空的字典
每次處理 都會內存拿那個字典
當要返回的時候 加密返回

返回前段的session的鍵  
配置文件能夠修改
SESSION_REFERSH_EACH_REQUEST #每次請求超時時間日後推 

　　

閃現

在session中存儲一個數據，讀取時經過pop將數據移除。

flash('臨時數據存儲','error')  # 值，分類
get_flashed_messages() # [  '存儲']
get_flashed_messages()  #[]   

print(get_flashed_messages(category_filter=['error']))  #經過category_filter 取error的分類裏找

flask啓動

請求進來了 會觸發app下的__call__ 方法

怎麼在請求以前作一個操做，call方法執行以後執行一個操做

重改 app.wsgi_app 
class Middleware(object):
    def __init__(self, old):
        self.old = old

    def __call__(self, *args, **kwargs):
        print('前')
        ret = self.old(*args, **kwargs)
        print('後')
        return ret


if __name__ == '__main__':
    app.wsgi_app = Middleware(app.wsgi_app)
    app.run()
    # app.__call__  # 請求來了才執行__call__方法

特殊的裝飾器

@app.befer_request
def x1():
    print("after")
    


@app.after_request
def x2(response):
     return response



before #先定義限制性

after  #後往上執行

 　　

 

 

@app.before_first_request   #flask 啓動執行第一次請求

@app。errorhandle(404)
def not_found(arg):
    return '沒找到'

路由設置的兩種方式

def index():
	return "index"
app.add_url_rule("/xxx",None,index)
			

- 不用讓endpoint重名
- 若是重名函數也必定要相同。

參數

b. 參數 
			rule,                       URL規則
			view_func,                  視圖函數名稱
			endpoint=None,              名稱，用於反向生成URL，即： url_for('名稱')
			methods=None,               容許的請求方式，如：["GET","POST"]
			strict_slashes=None,        對URL最後的 / 符號是否嚴格要求， False  便可baidu.com/1 也能夠 baidu.com/1/
			redirect_to=None,           重定向到指定地址，例如 開發完新系統後 應該重定向到新地址，而不是使用新的url redirect_to="xxx/"
			defaults=None,              默認值,當URL中無參數，函數須要參數時，使用defaults={'k':'v'}爲函數提供參數
			subdomain=None,             子域名訪問

　　

from flask import Flask, views, url_for

app = Flask(import_name=__name__)
app.config['SERVER_NAME'] = 'wupeiqi.com:5000'
"""
127.0.0.1   wupeiqi.com
127.0.0.1   web.wupeiqi.com
127.0.0.1   admin.wupeiqi.com

"""

# http://admin.wupeiqi.com:5000/ 才能訪問
@app.route("/", subdomain="admin")
def admin_index():
    return "admin.your-domain.tld"


# http://web.wupeiqi.com:5000/
@app.route("/", subdomain="web")
def web_index():
    return "web.your-domain.tld"


# http://sdsdf.wupeiqi.com:5000/
# http://sdfsdf.wupeiqi.com:5000/
# http://asdf.wupeiqi.com:5000/

@app.route("/dynamic", subdomain="<username>")
def username_index(username):
    """Dynamic subdomains are also supported
    Try going to user1.your-domain.tld/dynamic"""
    return username + ".your-domain.tld"


if __name__ == '__main__':
    app.run()

CBV

import functools
from flask import Flask,views
app = Flask(__name__)


def wrapper(func):
    @functools.wraps(func)
    def inner(*args,**kwargs):
        return func(*args,**kwargs)

    return inner



class UserView(views.MethodView):
    methods = ['GET']  #只支持GET 請求
    decorators = [wrapper,] #給每一個函數加裝飾器

    def get(self,*args,**kwargs):
        return 'GET'

    def post(self,*args,**kwargs):
        return 'POST'

app.add_url_rule('/user',None,UserView.as_view('uuuu'))  #uuu 就是endpoint

if __name__ == '__main__':
    app.run()

自定義路由 （略）

from flask import Flask,url_for

app = Flask(__name__)

# 步驟一：定製類
from werkzeug.routing import BaseConverter
class RegexConverter(BaseConverter):
	"""
	自定義URL匹配正則表達式
	"""

	def __init__(self, map, regex):
		super(RegexConverter, self).__init__(map)
		self.regex = regex

	def to_python(self, value):
		"""
		路由匹配時，匹配成功後傳遞給視圖函數中參數的值
		:param value:
		:return:
		"""
		return int(value)

	def to_url(self, value):
		"""
		使用url_for反向生成URL時，傳遞的參數通過該方法處理，返回的值用於生成URL中的參數
		:param value:
		:return:
		"""
		val = super(RegexConverter, self).to_url(value)
		return val

# 步驟二：添加到轉換器
app.url_map.converters['reg'] = RegexConverter

"""
1. 用戶發送請求
2. flask內部進行正則匹配
3. 調用to_python(正則匹配的結果)方法
4. to_python方法的返回值會交給視圖函數的參數

"""

# 步驟三：使用自定義正則
@app.route('/index/<reg("\d+"):nid>')
def index(nid):
	print(nid,type(nid))

	print(url_for('index',nid=987))
	return "index"

if __name__ == '__main__':
	app.run()

session實現原理（源碼） 08

藍圖（給開發者提供目錄結構）16-09

 

 

 

 

 

 

 

 

#counter 視圖
from flask import Blueprint,render_template

ac = Blueprint('ac',__name__)

@ac.before_request
def x1():
    print('app.before_request')


@ac.route('/login')
def login():
    return render_template('login.html')


@ac.route('/logout')
def logout():
    return 'Logout'

 

#user 視圖
from flask import Blueprint

uc = Blueprint('uc',__name__)


@uc.route('/list')
def list():
    return 'List'


@uc.route('/detail')
def detail():
    return 'detail'

 

#__init__  註冊藍圖
from flask import Flask
from .views.account import ac
from .views.user import uc

def create_app():

    app = Flask(__name__)

    app.register_blueprint(ac)
    app.register_blueprint(uc,url_prefix='/api')
    return app

　　

manage.py

from crm import create_app

app = create_app()

if __name__ == '__main__':
    app.run()

自定義靜態文件和模板路徑

ac = Blueprint('ac',__name__,template_folder ='',static_url_path='') 　　

加前綴 -有點像路由分發

app.register_blueprint(uc,url_prefix='/uc') #user/login

給藍圖加before_request 就至關於在局部才觸發

from flask import Blueprint

uc = Blueprint('uc',__name__)

@uc.before_request  #只在局部加裝飾器
def x1():
    xxxx

@uc.route('/login')
def login():
    xxx　　

threading.local【和flask無任何關係】 16-12

上下文管理 16-13

箱子 ：{
線程id：{ctx:ctx 對象}，
線程id2：{ctx:ctx 對象}，
}

視圖函數：
from flask  import request 


 請求結束後 根據線程惟一標識 移除空調上的數據上　　

上下文管理

- 面向對象中特殊方法 setattr/getattr注意事項：

class Foo(object):
def __init__(self):
    # self.storage = {}
    object.__setattr__(self,'storage',{})  #不能再init 的時候經過.的方式賦值

    def __setattr__(self, key, value):
        print(key,value,self.storage)

obj = Foo()
obj.xx = 123
    

基於列表實現的棧

class Stack(object):

    def __init__(self):
        self.data = []

    def push(self,val):
        self.data.append(val)

    def pop(self):
        return self.data.pop()
    
    def top(self):
        return self.data[-1]

_stack = Stack()

_stack.push('佳俊')
_stack.push('鹹魚')

print(_stack.pop())
print(_stack.pop())

棧

Local 對象（原理）

try:
    from greenlet import getcurrent as get_ident
except:
    from threading import get_ident

"""
class Local(object):
    
    def __init__(self):
        object.__setattr__(self,'storage',{})  #建立一個線程  定義一個空字典
    
    def __setattr__(self, key, value):
        ident = get_ident()   #獲取表示
        if ident not in self.storage:
            self.storage[ident] = {key:value} #若是這個表示不存在的話 就建立這個
        else:
            self.storage[ident][key] = value   #若是這個表示存在的話  修改
            
    def __getattr__(self, item):
        ident = get_ident()
        if ident in self.storage:              #有的話 返回這個
            return self.storage[ident].get(item)
"""　　

人家的local

__slots__  #規定了只能點哪些屬性

local 作了數據隔離、
class Local(object):
    __slots__ = ('__storage__', '__ident_func__')

    def __init__(self):
        # __storage__ = {1231:{'stack':[]}}
        object.__setattr__(self, '__storage__', {})
        object.__setattr__(self, '__ident_func__', get_ident)

    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

    def __setattr__(self, name, value):
        ident = self.__ident_func__()
        storage = self.__storage__
        try:
            storage[ident][name] = value
        except KeyError:
            storage[ident] = {name: value}

    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

flask-session 使用

pip3 install flask-session

app.config["session_type"]='redis'

 

 

　　

 

- 使用
    # by luffycity.com
    import redis
    from flask import Flask,request,session
    from flask.sessions import SecureCookieSessionInterface
    from flask_session import Session

    app = Flask(__name__)

    # app.session_interface = SecureCookieSessionInterface()
    # app.session_interface = RedisSessionInterface()
    app.config['SESSION_TYPE'] = 'redis'
    app.config['SESSION_REDIS'] = redis.Redis(host='140.143.227.206',port=6379,password='1234')
    Session(app)

    @app.route('/login')
    def login():
        session['user'] = 'alex'
        return 'asdfasfd'

    @app.route('/home')
    def index():
        print(session.get('user'))

        return '...'


    if __name__ == '__main__':
        app.run()
- 原理：
    - session數據保存到redis
        session:隨機字符串1:q23asifaksdfkajsdfasdf
        session:隨機字符串2:q23asifaksdfkajsdfasdf
        session:隨機字符串3:q23asifaksdfkajsdfasdf
        session:隨機字符串4:q23asifaksdfkajsdfasdf
        session:隨機字符串5:q23asifaksdfkajsdfasdf
    - 隨機字符串返回給用戶。
        隨機字符串和　

 work——home

 

 

 

 

 

 

 

代碼統計用戶列表 18-04

數據庫鏈接池DBUTILS 18-10

setting

pip3 install DBUtils


setting
from DBUtils.PooledDB import PooledDB, SharedDBConnection
import pymysql

class Config(object):
    SALT = b"sdf1123df"
    SECRET_KEY = 'asdf123sdfsdfsdf'
    MAX_CONTENT_LENGTH = 1024 * 1024 * 7


    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='s9day118',
        charset='utf8'
    )

　使用

conn=POOL.connect()

　　

handle

import pymysql

from settings import Config

def connect():
    conn = Config.POOL.connection()
    cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
    return conn,cursor


def connect_close(conn,cursor):
    cursor.close()
    conn.close()

def fetch_all(sql,args):
    conn,cursor = connect()

    cursor.execute(sql, args)
    record_list = cursor.fetchall()
    connect_close(conn,cursor)

    return record_list


def fetch_one(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    result = cursor.fetchone()
    connect_close(conn, cursor)

    return result


def insert(sql, args):
    conn, cursor = connect()
    row = cursor.execute(sql, args)
    conn.commit()
    connect_close(conn, cursor)
    return row

　加鎖

user1 

mysql  >  begin；
mysql  >  select * from ss for update ;  #這樣纔算加鎖 有結果

user  2  

mysql  >  begin；
mysql  >  select * from ss for update ;  #阻塞  鎖還沒釋放

釋放鎖的操做

commit;  支持行鎖

indndb 支持行鎖

各個框架加鎖

 

 

 

 

wtforms

pip3 install wtforms

from flask import Flask,request,render_template,session,current_app,g,redirect
from wtforms import Form
from wtforms.fields import simple
from wtforms.fields import html5
from wtforms.fields import core

from wtforms import widgets
from wtforms import validators

app = Flask(__name__)


class LoginForm(Form):
    name = simple.StringField(   #定義字段類型
        validators=[             #
            validators.DataRequired(message='用戶名不能爲空.'), #制定規則  
            # validators.Length(min=6, max=18, message='用戶名長度必須大於%(min)d且小於%(max)d')
        ],
        widget=widgets.TextInput(),  #制定input類型
        render_kw={'placeholder':'請輸入用戶名'}  #增長屬性  透明提示 
    )
    pwd = simple.PasswordField(
        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個特殊字符')

        ],
        render_kw={'placeholder':'請輸入密碼'}
    )


@app.route('/login',methods=['GET','POST'])
def login():
    if request.method == "GET":
        form = LoginForm()
        # print(form.name,type(form.name)) # form.name是StringField()對象, StringField().__str__
        # print(form.pwd,type(form.pwd))   # form.pwd是PasswordField()對象,PasswordField().__str__
        return render_template('login.html',form=form)  #返回表單

    form = LoginForm(formdata=request.form)   #生成有數據表單 
    if form.validate():  #校驗
        print(form.data)
        return redirect('https://www.luffycity.com/home')
    else:
        # print(form.errors)
        return render_template('login.html', form=form)


class RegisterForm(Form):
    name = simple.StringField(
        label='用戶名',
        validators=[
            validators.DataRequired()
        ],
        widget=widgets.TextInput(),
        render_kw={'class': 'form-control'}, #添加屬性
        default='alex'  #默認值
    )

    pwd = simple.PasswordField(
        label='密碼',
        validators=[
            validators.DataRequired(message='密碼不能爲空.')
        ],
        widget=widgets.PasswordInput(),
        render_kw={'class': 'form-control'}
    )

    pwd_confirm = simple.PasswordField(
        label='重複密碼',
        validators=[
            validators.DataRequired(message='重複密碼不能爲空.'),
            validators.EqualTo('pwd', message="兩次密碼輸入不一致")
        ],
        widget=widgets.PasswordInput(),
        render_kw={'class': 'form-control'}
    )

    email = html5.EmailField(
        label='郵箱',
        validators=[
            validators.DataRequired(message='郵箱不能爲空.'),
            validators.Email(message='郵箱格式錯誤')
        ],
        widget=widgets.TextInput(input_type='email'),
        render_kw={'class': 'form-control'}
    )

    gender = core.RadioField(
        label='性別',
        choices=(
            (1, '男'),
            (2, '女'),
        ),
        coerce=int # int("1")
    )
    city = core.SelectField(
        label='城市',
        choices=(
            ('bj', '北京'),
            ('sh', '上海'),
        )
    )

    hobby = core.SelectMultipleField(
        label='愛好',
        choices=(
            (1, '籃球'),
            (2, '足球'),
        ),
        coerce=int
    )

    favor = core.SelectMultipleField(
        label='喜愛',
        choices=(
            (1, '籃球'),
            (2, '足球'),
        ),
        widget=widgets.ListWidget(prefix_label=False),
        option_widget=widgets.CheckboxInput(),
        coerce=int,
        default=[1, ]
    )



@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'GET':
        form = RegisterForm()
        return render_template('register.html', form=form)

    form = RegisterForm(formdata=request.form)
    if form.validate():
        print(form.data)
        return redirect('https://www.luffycity.com/home')

    return render_template('register.html', form=form)

import helper
class UserForm(Form):
    city = core.SelectField(
        label='城市',
        choices=(),
        coerce=int
    )
    name = simple.StringField(label='姓名')

    def __init__(self,*args,**kwargs):
        super(UserForm,self).__init__(*args,**kwargs)

        self.city.choices=helper.fetch_all('select id,name from tb1',[],type=None)


@app.route('/user')
def user():
    if request.method == "GET":
        #form = UserForm(data={'name':'alex','city':3})
        form = UserForm()
        return render_template('user.html',form=form)


if __name__ == '__main__':
    app.run()

　　

{{% for  field  in from  %}}
{{ filed }}
{{ field.lable }}
{% endfor %}


默認選中

favor = core.SelectMultipleField(
        label='喜愛',
        choices=(
            (1, '籃球'),
            (2, '足球'),
        ),
        widget=widgets.ListWidget(prefix_label=False),
        option_widget=widgets.CheckboxInput(),
        coerce=int,
        default=[1, ]
    )

顯示錯誤信息
{{filed.error[0] }}