flask中的CBV,flash,Flask-Session,WTForms - MoudelForm,DBUtils 資料庫連線池
阿新 • • 發佈：2018-12-20
CBV和flash
from flask import Flask,request,render_template,redirect,url_for,flash,get_flashed_messages from flask import views app = Flask(__name__) # type:Flask app.secret_key = "123" @app.route("/") def index(): # flash("6666","tag") # flash("7777","tags") return "123" class LoginClass(views.MethodView):def get(self): # print(get_flashed_messages("tag")) # print(get_flashed_messages()) print(url_for("login")) return render_template("login.html") def post(self): return "123" # self.add_url_rule(rule, endpoint, f, **options) app.add_url_rule("/login",view_func=LoginClass.as_view("login")) @app.before_first_request def asd(): return "123" if __name__ == '__main__': app.run(debug=True)
Flask-session
from flask import Flask, session, render_template, session from flask import views from redis import Redis from flask_session importSession app = Flask(__name__) # type: Flask app.config['SESSION_TYPE'] = 'redis' app.config['SESSION_REDIS'] = Redis('127.0.0.1', 6379, db=8) Session(app) @app.route('/') def index(): session['user'] = 'qwe' # 1a86508c - f214 - 4382 - bfac - e837aaf5b1ed # 將這個作為key 存在了redis中,可以通過get session:1a86508c-f214-4382-bfac-e837aaf5b1ed return '>>>index' class LoginClass(views.MethodView): def get(self): ll = session.get('user') print(ll) return render_template('login.html') def post(self): return '123' app.add_url_rule('/login', view_func=LoginClass.as_view('login')) if __name__ == '__main__': app.run(debug=True)
WTForms
app.py from flask import Flask,request,render_template from flask import views from wtforms.fields import simple,core from wtforms import Form from wtforms import validators # from wtforms import widgets class LoginForm(Form): username = simple.StringField( label="使用者名稱", validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=3,max=5,message="不能小於3位，不能大於5位") ], # widget=widgets.TextInput(), render_kw={"class":"my_username"} ) password = simple.PasswordField( label="密碼", validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=6, max=6, message="密碼必須為6位"), validators.Regexp(regex="\d+", message="密碼必須位數字"), ], # widget=widgets.TextInput(), render_kw={"class": "my_password"} ) class RegForm(Form): username = simple.StringField( label="使用者名稱", validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=3, max=5, message="不能小於3位，不能大於5位") ], # widget=widgets.TextInput(), render_kw={"class": "my_username"} ) nickname = simple.StringField( label="暱稱", validators=[ validators.DataRequired(message="不能為空"), ], # widget=widgets.TextInput(), render_kw={"class": "my_username"} ) password = simple.PasswordField( label="密碼", validators=[ validators.DataRequired(message="不能為空"), validators.Length(min=6, max=6, message="密碼必須為6位"), validators.Regexp(regex="\d+", message="密碼必須位數字"), ], # widget=widgets.TextInput(), render_kw={"class": "my_password"} ) repassword = simple.PasswordField( label="重複密碼", validators=[ validators.EqualTo(fieldname="password",message="兩次密碼不一致") ] ) email = simple.StringField( label="暱稱", validators=[ validators.Email(message="格式不正確"), ], # widget=widgets.TextInput(), render_kw={"class": "my_username"} ) gender = core.RadioField( label="性別", coerce=int, choices=( (1,"女"), (2,"男") ), default=1 ) hobby = core.SelectMultipleField( label="愛好", coerce=int, choices=( (1, "小姐姐"), (2, "小蘿莉"), (3, "小哥哥"), (4, "小正太"), (5, "阿姨"), (6, "大叔"), ), default=(1,2,5) ) submit = simple.SubmitField( label="提交" ) app = Flask(__name__) # type:Flask @app.route("/") def index(): # 87fd0ba1-6c36-44be-bc36-2a95b764e122 return "123" class LoginClass(views.MethodView): def get(self): lf = LoginForm() return render_template("login.html",lf=lf) def post(self): lf = LoginForm(request.form) if lf.validate(): return "123" else: return render_template("login.html", lf=lf) app.add_url_rule("/login",view_func=LoginClass.as_view("login")) class RegClass(views.MethodView): def get(self): rf = RegForm() return render_template("reg.html", rf=rf) def post(self): rf = RegForm(request.form) if rf.validate(): return "123" else: return render_template("reg.html", rf=rf) app.add_url_rule("/reg", view_func=RegClass.as_view("reg")) if __name__ == '__main__': app.run(debug=True)app.py
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta http-equiv="content-Type" charset="UTF-8"> <meta http-equiv="x-ua-compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>Title</title> </head> <body> <form action="" method="post" novalidate> {{ lf.username.label }}{{ lf.username }}{{ lf.username.errors.0 }} <p>{{ lf.password.label }}{{ lf.password }}{{ lf.password.errors.0 }}</p> <input type="submit" value="提交"> </form> </body> </html>login.html
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta http-equiv="content-Type" charset="UTF-8"> <meta http-equiv="x-ua-compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>Title</title> </head> <body> <form action="" method="post"> {% for field in rf %} <p>{{ field.label }}{{ field }}{{ field.errors.0 }}</p> {% endfor %} </form> </body> </html>reg.html
DBUtils 資料庫連線池
import pymysql from DBUtils.PooledDB import PooledDB 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='', database='db7', charset='utf8' ) def func(): conn = POOL.connection() cursor = conn.cursor(pymysql.cursors.DictCursor) cursor.execute('select * from userinfo') 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，所有值無論設定為多少，_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=host, port=int(port), user=dbuser, password=password, database=database, 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() conn.close() return result def insert_one(self,sql,args): conn,cursor = self.create_conn_cursor() res = cursor.execute(sql,args) conn.commit() 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", "", "db7") # res = sqlhelper.fetch_all("select * from userinfo where id=%s",(5,)) # print(res) # res = sqlhelper.insert_one("insert into userinfo VALUES (%s,%s,%s)",(7,"jinwangba","66666")) # print(res) # sqlhelper.update("update user SET name=%s WHERE id=%s",("yinwangba",1))封裝成類