flask中的CBV,flash,Flask-Session,WTForms - MoudelForm,DBUtils 数据库连接池-白红宇

flask中的CBV,flash,Flask-Session,WTForms - MoudelForm,DBUtils 数据库连接池

阅读量：4353 次

发布时间：2019-06-07

本文共 8840 字，大约阅读时间需要 29 分钟。

CBV和flash

from flask import Flask,request,render_template,redirect,url_for,flash,get_flashed_messagesfrom flask import viewsapp = Flask(__name__)  # type:Flaskapp.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_requestdef asd():    return "123"if __name__ == '__main__':    app.run(debug=True)

Flask-session

from flask import Flask, session, render_template, sessionfrom flask import viewsfrom redis import Redisfrom flask_session import Sessionapp = Flask(__name__)  # type: Flaskapp.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.pyfrom flask import Flask,request,render_templatefrom flask import viewsfrom wtforms.fields import simple,corefrom wtforms import Formfrom wtforms import validators# from wtforms import widgetsclass 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

    
          
          
          Title
      
           {
     { lf.username.label }}{
     { lf.username }}{
     { lf.username.errors.0 }}    
       {
    { lf.password.label }}{
    { lf.password }}{
    { lf.password.errors.0 }}

    
          
          
          Title
      
           {
     % for field in rf %}        
       {
    { field.label }}{
    { field }}{
    { field.errors.0 }}
    {
     % endfor %}

reg.html

DBUtils 数据库连接池

import pymysqlfrom DBUtils.PooledDB import PooledDBPOOL = 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 PooledDBimport pymysqlclass 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 ressqlhelper = 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))

封装成类

转载于:https://www.cnblogs.com/shuai1993/p/10151840.html

你可能感兴趣的文章

mysql的数据存储

查看>>

[转载] Activiti Tenant Id 字段释疑

查看>>

[Java 8] (8) Lambda表达式对递归的优化(上) - 使用尾递归 .

查看>>

SQL Server-聚焦移除Bookmark Lookup、RID Lookup、Key Lookup提高SQL查询性能

查看>>

最小权限的挑战