本文实例讲述了Flask框架通过Flask_login实现用户登录功能。分享给大家供大家参考,具体如下:
通过Flask_Login实现用户验证登录,并通过login_required装饰器来判断用户登录状态来判断是否允许访问视图函数。
运行环境:
python3.5
Flask 0.12.2
Flask_Login 0.4.1
Flask-WTF 0.14.2
PyMySQL 0.8.0
WTForms 2.1
DBUtils 1.2
目录结构:
直接看代码,具体功能有注释
Model/User_model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
#创建一个类,用来通过sql语句查询结果实例化对象用
class User_mod():
def __init__( self ):
self . id = None
self .username = None
self .task_count = None
self .sample_count = None
def todict( self ):
return self .__dict__
#下面这4个方法是flask_login需要的4个验证方式
def is_authenticated( self ):
return True
def is_active( self ):
return True
def is_anonymous( self ):
return False
def get_id( self ):
return self . id
# def __repr__(self):
# return '<User %r>' % self.username
|
templates/login.html
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
<!DOCTYPE html>
< html lang = "en" >
< head >
< meta charset = "UTF-8" >
< title >Title</ title >
</ head >
< body >
< div class = "login-content" >
< form class = "margin-bottom-0" action = "{{ action }}" method = "{{ method }}" id = "{{ formid }}" >
{{ form.hidden_tag() }}
< div class = "form-group m-b-20" >
{{ form.username(class='form-control input-lg',placeholder = "用户名") }}
</ div >
< div class = "form-group m-b-20" >
{{ form.password(class='form-control input-lg',placeholder = "密码") }}
</ div >
< div class = "checkbox m-b-20" >
< label >
{{ form.remember_me() }} 记住我
</ label >
</ div >
< div class = "login-buttons" >
< button type = "submit" class = "btn btn-success btn-block btn-lg" >登 录</ button >
</ div >
</ form >
</ div >
</ body >
</ html >
|
User_dal/dal.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
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 = '192.168.20.195' ,
port = 3306 ,
user = 'root' ,
password = 'youpassword' ,
database = 'mytest' ,
charset = 'utf8'
)
class SQLHelper( object ):
@staticmethod
def fetch_one(sql,args):
conn = POOL.connection() #通过连接池链接数据库
cursor = conn.cursor() #创建游标
cursor.execute(sql, args) #执行sql语句
result = cursor.fetchone() #取的sql查询结果
conn.close() #关闭链接
return result
@staticmethod
def fetch_all( self ,sql,args):
conn = POOL.connection()
cursor = conn.cursor()
cursor.execute(sql, args)
result = cursor.fetchone()
conn.close()
return result
|
User_dal/user_dal.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
from Model import User_model
from User_dal import dal
from User_dal import user_dal
class User_Dal:
persist = None
#通过用户名及密码查询用户对象
@classmethod
def login_auth( cls ,username,password):
print ( 'login_auth' )
result = { 'isAuth' : False }
model = User_model.User_mod() #实例化一个对象,将查询结果逐一添加给对象的属性
sql = "SELECT id,username,sample_count,task_count FROM User WHERE username ='%s' AND password = '%s'" % (username,password)
rows = user_dal.User_Dal.query(sql)
print ( '查询结果>>>' ,rows)
if rows:
result[ 'isAuth' ] = True
model. id = rows[ 0 ]
model.username = rows[ 1 ]
model.sample_count = rows[ 2 ]
model.task_count = rows[ 3 ]
return result,model
#flask_login回调函数执行的,需要通过用户唯一的id找到用户对象
@classmethod
def load_user_byid( cls , id ):
print ( 'load_user_byid' )
sql = "SELECT id,username,sample_count,task_count FROM User WHERE id='%s'" % id
model = User_model.User_mod() #实例化一个对象,将查询结果逐一添加给对象的属性
rows = user_dal.User_Dal.query(sql)
if rows:
result = { 'isAuth' : False }
result[ 'isAuth' ] = True
model. id = rows[ 0 ]
model.username = rows[ 1 ]
model.sample_count = rows[ 2 ]
model.task_count = rows[ 3 ]
return model
#具体执行sql语句的函数
@classmethod
def query( cls ,sql,params = None ):
result = dal.SQLHelper.fetch_one(sql,params)
return result
|
denglu.py flask主运行文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
from flask import Flask,render_template,redirect
from flask_login import LoginManager,login_user,login_required,current_user
from flask_wtf.form import FlaskForm
from wtforms import StringField, PasswordField, BooleanField
from wtforms.validators import Length,DataRequired,Optional
from User_dal import user_dal
app = Flask(__name__)
#项目中设置flask_login
login_manager = LoginManager()
login_manager.init_app(app)
app.config[ 'SECRET_KEY' ] = '234rsdf34523rwsf'
#flask_wtf表单
class LoginForm(FlaskForm):
username = StringField( '账户名:' , validators = [DataRequired(), Length( 1 , 30 )])
password = PasswordField( '密码:' , validators = [DataRequired(), Length( 1 , 64 )])
remember_me = BooleanField( '记住密码' , validators = [Optional()])
@app .route( '/login' ,methods = [ 'GET' , 'POST' ])
def login():
form = LoginForm()
if form.validate_on_submit():
username = form.username.data
password = form.password.data
result = user_dal.User_Dal.login_auth(username,password)
model = result[ 1 ]
if result[ 0 ][ 'isAuth' ]:
login_user(model)
print ( '登陆成功' )
print (current_user.username) #登录成功之后可以用current_user来取该用户的其他属性,这些属性都是sql语句查来并赋值给对象的。
return redirect( '/t' )
else :
print ( '登陆失败' )
return render_template( 'login.html' ,formid = 'loginForm' ,action = '/login' ,method = 'post' ,form = form)
return render_template( 'login.html' ,formid = 'loginForm' ,action = '/login' ,method = 'post' ,form = form)
'''登录函数,首先实例化form对象
然后通过form对象验证post接收到的数据格式是否正确
然后通过login_auth函数,用username与password向数据库查询这个用户,并将状态码以及对象返回
判断状态码,如果正确则将对象传入login_user中,然后就可以跳转到正确页面了'''
@login_manager .user_loader
def load_user( id ):
return user_dal.User_Dal.load_user_byid( id )
'''
load_user是一个flask_login的回调函数,在登陆之后,每访问一个带Login_required装饰的视图函数就要执行一次,
该函数返回一个用户对象,通过id来用sql语句查到的用户数据,然后实例化一个对象,并返回。
'''
#登陆成功跳转的视图函数
@app .route( '/t' )
@login_required
def hello_world():
print ( '登录跳转' )
return 'Hello World!'
#随便写的另一个视图函数
@app .route( '/b' )
@login_required
def hello():
print ( '视图函数b' )
return 'Hello b!'
if __name__ = = '__main__' :
app.run()
|
简单总结一下:
通过flask的form表单验证数据格式
然后通过用户名密码从数据库取用户对象,将sql执行结果赋值给一个实例化的对象
将这个对象传给login_user,
然后成功跳转。
注意要写一个load_user回调函数吗,返回的是通过id取到的数据库并实例化的对象的用户对象。
这个回调函数每次访问带login_required装饰器的视图函数都会被执行。
还有一个就是current_user相当于就是实例化的用户对象,可以取用户的其他属性,注意,其他属性仅限于sql语句查到的字段并添加给实例化对象的属性。
代码比较简单,只是为了实现功能,敬请谅解,如有错误欢迎指出。
希望本文所述对大家Python程序设计有所帮助。
原文链接:https://www.cnblogs.com/ArmoredTitan/p/8985656.html