Python MVC架构下的Redis与MySQL混合存储登录验证系统实现

咖啡半糖

发布时间：2026-10-13 | 4人浏览过

原创

本文详细介绍基于Python实现的轻量级登录验证系统，采用MVC架构设计，结合Redis内存数据库的高速查询与MySQL持久化存储的优势。系统涵盖注册、登录、改密及注销四大核心功能，通过Model-View-Controller分层实现业务逻辑与数据访问的解耦，为构建高效安全的用户认证模块提供完整代码参考与架构思路。

Python MVC架构下的Redis与MySQL混合存储登录验证系统实现

本文分享一个使用Python编写的完整登录验证系统实例，供读者参考学习。

小型登录注册验证系统

系统概述与技术选型

一、概述

本系统基于Redis与MySQL数据库构建，旨在通过混合存储方案实现高效的登录注册验证功能，并在此过程中初步掌握MVC框架的应用。

系统主要功能包括：用户登录、新用户注册、密码修改以及账户注销。

数据库选型方面，采用Redis存储用户信息以利用其内存特性实现高速数据查询；同时使用MySQL进行数据存储，凭借其对大容量数据的支持及优秀的表关系管理能力。两者结合使用，兼顾了读写性能与数据持久性，是目前业界流行的数据库搭配方案。

开发语言选用Python。

MVC框架全称为Model View Controller（模型-视图-控制器），这是一种经典的软件设计模式。它通过将业务逻辑、数据模型和界面显示分离来组织代码，将业务逻辑集中处理。这种架构使得在改进界面或定制用户交互时，无需重新编写底层业务逻辑，从而提高了代码的可维护性和扩展性。MVC常用于映射传统的输入、处理与输出功能，构建逻辑清晰的图形化用户界面结构。

代码结构与模块详解

二、代码实现

完整实例代码可通过本站提供的链接下载。

项目GitHub地址：https://github.com/liangdongchang/pyCheckLoginSys.git

1、Init初始化模块

该模块负责服务的初始化工作，具体包括：

① 在MySQL数据库中创建名为“homework”的数据库，并建立“t_usr”用户表；

② 启动Redis服务程序。


'''
@author ldc
'''
import os
import pymysql
'''
初始化服务：
1、在mysql上新建一个数据库“homework”和建表"t_usr"
2、开启redis服务程序
'''
# 建立数据库连接
conn = pymysql.connect(
  host='localhost',
  user='root',
  password="123456",
  port=3306
)
# 获取游标
cursor = conn.cursor()
# 创建数据库
dbname = 'homework'
sql='''
   create database if not EXISTS %s charset=utf8;
  '''%dbname
cursor.execute(sql)
# 使用数据库
cursor.execute('use %s'%dbname)
# 创建表
sql = '''
  create table if not EXISTS t_usr(
     id INTEGER PRIMARY KEY auto_increment,
     username varchar(20) unique not null,
     password varchar(20) not null
    );
'''
cursor.execute(sql)
# 关闭游标与连接
cursor.close()
conn.close()
# 开启redis服务，新建一个启动redisd.bat文件，
#以后开启redis服务就可以直接打开这个文件了
def openRedisd(path):
  rPath = """@echo off
      redis-server %s
      pause"""%path
  with open(r"C:\Users\LDCPC\Desktop\启动redisd.bat","w",encoding="ANSI")
  as f:
   f.write(rPath)
openRedisd(r"D:\ruanjian\redis-64.2.8.2101\redis.windows.conf")
# 打开文件“启动redisd.bat”
os.popen(r"C:\Users\LDCPC\Desktop\启动redisd.bat")

2、View层（视图层）

负责与用户进行交互，接收用户的输入指令并向用户展示操作结果。


'''
@author ldc
'''
from controller import urls
from model.model import User
from utils.dbUtil import RedisUtil
'''
需求：登录注册验证
1、登录
2、注册
3、改密
4、注销
'''
# 主界面接口
def index():
  while True:
    #登录界面
    print("********************************")
    print("*               *")
    print("*  (1) 登录   (2)注册   *")
    print("*  (3) 改密   (4)注销   *")
    print("*      (5)退出      *")
    print("********************************")
    print()
    num = input("请输入功能序号：")
    if num in ['1','2','3','4','5']:
      return num
    else:
      print("输入有误，请重新输入!!!")
# 输入账号与密码
def inputInfo():
  return input("请输入账号和密码(逗号隔开)：").split(',')
if __name__ == '__main__':
  # 连接redis数据库
  RedisUtil.connect()
  while True:
    # 初始化界面
    num = index()
    # 输入账号密码
    username, password = inputInfo()
    # 实例化一个用户类
    user = User(username, password)
    if num == '1':
      urls.login(user) #登录
    elif num == '2':
      urls.regist(user) # 注册
    elif num == '3':
      urls.changePasswd(user) # 改密
    elif num == '4':
      urls.deleteUser(user) # 注销
    else:
      break

3、Controller层（控制器层）

负责处理具体的业务逻辑，并控制整个系统的运行流程。


'''
@author ldc
'''
from model.model import UserDao
# 先查询该用户是否存在数据库中
def exists(user):
  '''先查看Redis缓存中是否有该用户数据'''
  if not UserDao.exists(user.username, 'redis'):
   '''然后在mysql中查询该用户是否存在'''
   if UserDao.exists(user.username, 'mysql'):
     # 若在mysql存在就把该用户写进redis,
     UserDao.redis.set(user.username, user.password)
     return 'mysql'
   else :
     return None
  return 'redis'
'''
# 登录模块
先在redis上验证，验证成功则提示在redis上验证成功
否则到mysql中验证，验证成功则提示在mysql上验证成功
否则提示用户不存在
'''
def login(user):
  print("------------登录界面------------")
  # 查询该用户信息是否存在数据库中
  whereDB = exists(user)
  if whereDB == 'redis':
   # 匹配密码是否正确
   if UserDao.query(user, 'redis') == user.password:
     print("[在redis中查询到该用户]登录成功!!!")
     return 1
   else:
     print("[在redis中查询到该用户] 登录失败,用户名或者密码不正确!!!")
  elif whereDB == 'mysql':
   # 匹配密码是否正确
   if UserDao.query(user, 'mysql'):
     print("[在mysql中查询到该用户] 登录成功!!!")
     return 1
   else:
     print("[在mysql中查询到该用户] 登录失败,用户或者密码不正确!!!")
  else:
   print("[在mysql中查询不到该用户]登录失败,该用户不存在，请注册后再登录!!!")
  return 0
'''
# 注册模块
先在redis上查询账号是否存在，存在则注册失败
否则到mysql上查询，用户存在则注册失败
否则注册成功，把账号写进mysql，写进redis
'''
def regist(user):
  print("------------注册界面------------")
  # 查询该用户信息是否存在数据库中
  whereDB = exists(user)
  if whereDB :
   print("注册失败，该用户已存在!!!")
  else:
   if UserDao.insert(user):
     print("注册成功!!!")
   else:
     print("注册失败!!!")
'''
# 修改密码模块
先在redis上和mysql上查询，用户存在就在mysql上修改该用户密码，
然后把该用户信息重新写进redis中
在mysql中查询不到该用户，就返回该用户不存在，改密失败
'''
def changePasswd(user):
  print("------------改密界面------------")
  # 查询该用户信息是否存在数据库中
  whereDB = exists(user)
  if whereDB:
   user.password = input("请输入新密码：")
   if UserDao.changePasswd(user):
     print("改密成功!!!")
   else:
     print("改密失败!!!")
  else:
   print("用户不存在，改密失败!!!")
'''
# 注销用户模块
先在在redis上和mysql上查询，用户存在就在mysql和redis上删除该用户
在mysql中查询不到该用户，就返回该用户不存在，注销失败
'''
def deleteUser(user):
  print("------------注销界面------------")
  # 查询该用户信息是否存在数据库中
  if login(user):
   if UserDao.deleteUser(user):
     print("注销成功!!!")
     return
  print("注销失败!!!")

4、Model层（模型层）

负责访问数据库，实现业务逻辑与数据访问的分离，从而降低系统维护难度。


'''
@author ldc
'''
from utils.dbUtil import RedisUtil, MySQLUtil
# 用户模型类
class User:
  def __init__(self,username,password):
    self.username = username
    self.password = password
# UserDao
# 封装了对User数据的增删改查
# Dao=Database Access Object 数据库访问对象
class UserDao:
  # 创建数据库对象
  redis = RedisUtil()
  mySQL = MySQLUtil('homework','t_usr')
  # 执行数据库查询操作，返回查询结果
  @classmethod
  def query(cls,user,dbType):
    dataDict = {}
    dataDict["username"] = user.username
    dataDict["password"] = user.password
    if dbType == 'redis':
      return cls.redis.get(user.username)
    elif dbType == 'mysql':
      return cls.mySQL.query(dataDict)
  # 执行数据库查询操作，查询用户是否存在，返回查询结果
  @classmethod
  def exists(cls,username,dbType):
    dataDict = {}
    dataDict["username"] = username
    if dbType == 'redis':
      return cls.redis.exists(username)
    elif dbType == 'mysql':
      return cls.mySQL.exists(dataDict)
    else:
      pass
  # 执行数据插入操作，先把用户信息添加进mysql，然后再添加进redis
  @classmethod
  def insert(cls, user):
    dataDict = {}
    dataDict["username"] = user.username
    dataDict["password"] = user.password
    if cls.mySQL.insert(dataDict):
      cls.redis.set(user.username,user.password)
      return 1
    else:
      print("注册失败，服务器繁忙!!!")
      return 0
  # 修改密码
  @classmethod
  def changePasswd(cls, user):
    dataDict = {'changeCol': 'password = %s'%user.password,
     'caluse' : 'username = %s'%user.username}
    if cls.mySQL.update(dataDict):
      cls.redis.set(user.username,user.password)
      return 1
    else:
      print("修改密码失败，服务器繁忙!!!")
      return 0
  # 注销用户
  @classmethod
  def deleteUser(cls, user):
    dataDict = {'username' : user.username}
    if cls.mySQL.delete(dataDict):
      cls.redis.delete(user.username)
      return 1
    else:
      print("修改密码失败，服务器繁忙!!!")
      return 0

5、Utils工具包

封装了数据库的增删改查通用操作，供系统其他模块调用。


'''
@author ldc
'''
import pymysql
import redis as redis
'''
MySQL增删改查操作类
'''
class MySQLUtil:
  def __init__(self,dbName,tableName):
   self.dbName = dbName
   self.tableName = tableName
  # 连接数据库,并生成全局可用的连接对象和查询游标
  def connect(self):
   self.conn = pymysql.connect(
     host='localhost', user='root', password="123456",
     database=self.dbName, port=3306,
   )
   self.cursor = self.conn.cursor()
  # 关闭全局游标，断开全局连接
  def disconnect(self):
   self.cursor.close()
   self.conn.close()
  # 查询用户名是否存在
  def exists(self,dataDict):
   caluse = ''
   for key,value in dataDict.items():
     caluse += key + '="'+ value + '"'
   # print(caluse)
   sql = """
      select * from %s where %s ;
      """ % (self.tableName, caluse)
   return self.execute(sql)
  # 验证用户名和密码是否正确
  def query(self, dataDict):
   # 查询子条件拼接
   caluse = ''
   for key, value in dataDict.items():
     caluse += key + '="' + value + '" and '
   caluse = caluse[:-4]
   # print(caluse)
   sql = """
      select * from %s where %s;
      """% (self.tableName, caluse)
   return self.execute(sql)
  # 添加新用户
  def insert(self, dataDict):
   # sql语句拼接
   columns = ''
   values = ''
   for key, value in dataDict.items():
     columns += key + ','
     values += '"' + value + '",'
   columns = columns[:-1]
   values = values[:-1]
   sql = """
      insert into %s (%s) VALUES (%s);
      """ % (self.tableName, columns,values)
   # print(sql)
   return self.execute(sql)
  # 更新
  def update(self, dataDict):
   # sql语句拼接
   changeCol = dataDict['changeCol'] #要改变值的列名
   caluse = dataDict['caluse'] #要改变值的子条件
   sql = 'update %s set %s where %s'%(self.tableName,changeCol,caluse)
   return self.execute(sql)
  # 删除
  def delete(self, dataDict):
   # sql语句拼接
   caluse = ''
   for key,value in dataDict.items():
     caluse += key + '="' + value + '"'
   sql = """
      delete from %s where %s;
      """ % (self.tableName,caluse)
   # print(sql)
   return self.execute(sql)
  # print(sql)
  # 执行sql语句
  def execute(self, sql):
   self.connect()
   affected = 0
   try:
     affected = self.cursor.execute(sql)
   except BaseException as e:
     print(e)
     affected = 0
   finally:
     self.conn.commit()
     self.disconnect()
     return affected
'''
redis增删改查操作类
'''
class RedisUtil:
  # redis连接
  @classmethod
  def connect(cls):
   cls.client = redis.Redis(
     host='localhost', port=6379,
     db=1, password='123456',
   )
  # 判断键是否存在
  @classmethod
  def exists(cls,key):
   return cls.client.exists(key)
  # 存储键值,
  @classmethod
  def set(cls,key,value):
   # 键值存储在缓存中，保留时间为30秒
   cls.client.setex(key,value,30)
  # 获取键值
  @classmethod
  def get(cls,key):
   res = cls.client.get(key).decode("utf-8")
   return res
  # 删除键值
  def delete(cls, key):
   cls.client.delete(key)