1. 程式人生 > >flask的CBV,flash,Flask-Session,及WTForms-MoudelForm

flask的CBV,flash,Flask-Session,及WTForms-MoudelForm

1,CBV:

  • from flask import vews
  • class LoginView(views.MethodView):
    • def get(self):
      • return "雪雪其實也很好,"
    • def post(self):
      • return "聽說雪人很要走了,那些等有機會,等有時間,都覺的遙遠了"
  • app.add_url_rule("/login", endpoint=None, view_func=LoginView.as_view("login"))

2,flask:

  • from flask import flask,get_flash_messages
  • flash("雪雪", "tag")   # 快閃記憶體
  • get_flash_messages("tag")

3,Flask-Session

  • from flask_session import Session
  • from flask import session
  • app.config["SESSION_TYPE"] = "redis"
  • app.config["SESSION_REDIS"] = Redis("127.0.0.1", 6379, db=7)
  • Session(app)

4,WTForms - MoudelForm

  • from wtfroms.field import simple,core
  • from wtforms import Form,validators
  • class LoginForm(Form):
    • username = simple.StringField(label= "使用者名稱")
  • loginForm = LoginForm()
  • render_template("login.html", loginForm=loginForm)
  • {{ loginForm.username }}
  • {{ loginForm.password }}
  • {{ loginForm.username.errors.0 }}
  • loginForm = LoginForm(request.forms)
  • if not loginForm.validata():
    • render_template("login.html", loginForm)

具體程式碼:

CBV的寫法:

from flask import Flask, flash, get_flashed_messages,render_template
from flask import views
# 例項化Flask物件
app = Flask(__name__)  # type: Flask
app.secret_key = "miss"


@app.route("/")
def index():
    # 這個flash取完值就沒有了,一次性的事
    flash("xuexue", "小老弟") flash("xiaoxue") return "我的頭可不是麵糰捏的!!!" # CBV的寫法 class LoginClass(views.MethodView): # get請求的方法 def get(self): print(get_flashed_messages("小老弟")) print(get_flashed_messages()) return render_template("login.html") # 提交資料的方法 def post(self): return "來了, 老弟~~~" # 註冊路由 app.add_url_rule("/login", view_func=LoginClass.as_view("login")) @app.before_first_request def asd(): # 在這根據flash判斷是不是第一次請求 # 第一次請求,就把flash的值取走, print("bf1") return "不知道該寫什麼了..." if __name__ == '__main__': app.run(debug=True)

APP-flask-Session

from flask import Flask,render_template, redirect, session
from flask import views
from redis import Redis  # 這個必須寫,重新封裝Redis的時候用
from flask_session import Session  # 用了這個Session就不用指定secret_key = xxxx
# 例項化一個 Flask物件
app = Flask(__name__)  # type:Flask
app.config["SESSION_TYPE"] = "redis"
app.config["SESSION_REDIS"] = Redis("127.0.0.1", 6379, db=7)  # db表示能開幾個服務
# 例項化一個Session的物件把Flask數理化的物件傳進去,就不用指定secret_key=xxx了
Session(app)

@app.route("/")
def index(): session["user"] = "雪人" return "雪人哈哈哈哈哈~~~~" class LoginClass(views.MethodView): def get(self): print(session.get("user")) # 7b97e0ef-2a14-4725-842e-848fde4d2f15 # 和uuid的值特別的像 return render_template("login.html") def post(self): return "嘖嘖嘖..." app.add_url_rule("/login", view_func=LoginClass.as_view("login")) if __name__ == '__main__': app.run(debug=True)

app的檢視函式

from flask import Flask,render_template, request
from flask import views
from wtforms.fields import simple, core
from wtforms import Form
from wtforms import validators
from wtforms import widgets # CBV的寫法 # 指定登入的類繼承Form class LoginForm(Form): # 定義校驗的欄位 username = simple.StringField( label="使用者名稱", # 要校驗該欄位的內容 validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=5, max=10, message="使用者名稱不能小於5個字元求不能大於10個字元") ], # widget=widgets.TextInput render_kw={"class": "my_username"} # 在標籤上加一個樣式類  ) password = simple.PasswordField( label="密碼", # 要校驗該欄位的內容 validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=6,message="密碼最少要6位"), validators.Length(max=12, message="密碼至多12位"), validators.Regexp(regex="\d+", message="密碼必須是純數字"), ], render_kw={"class": "my_password"} ) # 註冊的Form class RegisterForm(Form): username = simple.StringField( label="使用者名稱", validators=[ validators.DataRequired(message="使用者名稱不能為空"), ], render_kw={"class": "my_username"} ) nickname = simple.StringField( label="暱稱", validators=[ validators.DataRequired(message="暱稱不能為空") ], render_kw={"class": "my_username"} ) password = simple.PasswordField( label="密碼", validators=[ validators.DataRequired(message="密碼不能為空"), validators.Length(min=6, max=12, message="密碼大於6小於12"), validators.Regexp(regex="\d+", message="密碼必須是數字"), ], render_kw={"class": "my_password"} ) re_password = simple.PasswordField( label="重複密碼", validators=[ validators.EqualTo(fieldname="password",message="兩次密碼不一致"), ], ) email = simple.StringField( label="郵箱", validators=[ validators.Email(message="格式不正確"), ], render_kw={"class": "my_email"} ) gender = core.RadioField( label="性別", coerce=int, # 拿到的是性別在資料庫中的前邊數字 choices=( (1, "女"), (2, "男"), ), default=1 # 設定預設值為1  ) hobby = core.SelectMultipleField( label="愛好", coerce=int, choices=( (1, "烤串"), (2, "烤麵包片"), (3, "烤魚"), (4, "pizza"), (5, "蠍王府"), ), default=(3,5) ) # 例項化一個Flask的物件 app = Flask(__name__) # type:Flask  @app.route("/") def index(): return "我的頭可不是麵糰捏的" # 登入頁面 class LoginView(views.MethodView): def get(self): loginForm = LoginForm() return render_template("login.html", loginForm=loginForm) def post(self): # 將獲取到的資料放到LoginForm中進行 loginForm = LoginForm(request.form) # 通過校驗的資料 if loginForm.validate(): return "登入成功" else: return render_template("login.html", loginForm=loginForm) app.add_url_rule("/login", view_func=LoginView.as_view("login")) # 註冊的業務邏輯函式 class RegisterView(views.MethodView): def get(self): # 例項化註冊的校驗類 regForm = RegisterForm() if regForm.validate(): return "註冊通過了校驗" else: return render_template("register.html", regForm=regForm) def post(self): regForm = RegisterForm(request.form) if regForm.validate(): return "通過校驗的資料" else: return render_template("register.html", regForm=regForm) app.add_url_rule("/register", view_func=RegisterView.as_view("register")) if __name__ == '__main__': app.run(debug=True)

5,DBUtils資料連線池

import pymysql
from DBUtils.PooledDB import PooledDB
# 例項化pooleDB物件
POOL = PooledDB(
    creator=pymysql,  # 使用連線資料庫的模組
    maxconnections=6,  # 連線池允許的最大連線數, 0和None表示不限制連線數
    mincached=2,  # 初始化時,連線池中至少建立的空閒的連線,0表示不建立
    maxcached=5, # 連線池中最多閒置的連線, 0和None不限制
    maxshared=3,  # 連線池中最多共享的連線數量, 0和None表示全部共享.ps:無用,因為pymysql和mysqldb等模組的threadsafety都為1, 所有值無論設定為多少,maxacahed永遠為0,所以永遠是共享所有連線都共享
    blocking=True,  # 連線池中如果沒有可用連線,是否阻塞等待.True,等待,False,不等待然後報錯
    maxusage=None,  # 一個連線最多被重複使用的次數, None表示無限制
    setsession=[],  # 開始會話前執行的命令列表. 如["set datestyle to ...", "set time zone ..."]
    ping=0,  # ping Mysql服務端,檢查是否服務可用. 如: 0 = None = nerver, 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="test", charset="utf8" ) def func(): # 在連線池中連線 conn = POOL.connection() # 油表使用闢謠mysql cursor = conn.cursor(pymysql.cursors.DictCursor) # 從表中查詢資料 cursor.execute('select * from xueren ') result = cursor.fetchall() print(result) conn.close() func()
from DBUtils.PooledDB import PooledDB
import pymysql

class MySQLhelper(object):
    def __init__(self, host, port, dbuser, password, database):
        self.pool = PooledDB(
            creator=pymysql,  # 使用連線資料庫的模組
            maxconnections=6,  # 連線池允許的最大連線數, 0和None表示不限制連線數
            mincached=2,  # 初始化時,連線池中至少建立的空閒的連線,0表示不建立
            maxcached=5,  # 連線池中最多閒置的連線, 0和None不限制
            maxshared=3, # 連線池中最多共享的連線數量, 0和None表示全部共享.ps:無用,因為pymysql和mysqldb等模組的threadsafety都為1, 所有值無論設定為多少,maxacahed永遠為0,所以永遠是共享所有連線都共享 blocking=True, # 連線池中如果沒有可用連線,是否阻塞等待.True,等待,False,不等待然後報錯 maxusage=None, # 一個連線最多被重複使用的次數, None表示無限制 setsession=[], # 開始會話前執行的命令列表. 如["set datestyle to ...", "set time zone ..."] ping=0, # ping Mysql服務端,檢查是否服務可用. 如: 0 = None = nerver, 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="test", charset="utf8" ) # 建立連線油表 def create_conn_cursor(self): conn = self.pool.connection() cursor = conn.cursor(pymysql.cursors.DictCursor) return conn, cursor # 獲取所有的資料(從資料庫中) def fetch_all(self, sql, args): conn, cursor = self.create_conn_cursor() cursor.execute(sql, args) result = cursor.fetchall() cursor.close( ) # 插入資料 def insert_one(self, sql, args): conn, cursor = self.create_conn_cursor() res = cursor.execute(sql, args) conn.commit() print(res) conn.close() return res # 更新資料 def update(self, sql, args): conn,cursor = self.create_conn_cursor() res = cursor.execute(sql, args) conn.commit() print(res) conn.close() return res sqlhelper = MySQLhelper("127.0.0.1", 3306, "root", "", "test") # 查詢表裡所有的資料 res = sqlhelper.fetch_all("select * from xueren where id=%s", (1,)) print(res) # 插入資料 # res = sqlhelper.insert_one("insert into xueren VALUES (%s,%s %s)", (1, "miss", "really")) # print(res) # 更新資料 # res = sqlhelper.update("update user SET name=%s WHERE id=%s", ("哈哈哈"))