目录

1.BaseEntity

2.t_user

3.DAOBase

4.UserDAO

---

刚开始用Python做Api后端，记录下学习过程

从数据库操作开始，定义实体，定义DAO层基类，及DAO业务类的使用

写的不好，有好的设计建议欢迎大家提出，一起学习进步

1.BaseEntity

这部分是实体的基类，比较简单，就一个使用类名作为表名的方法

from sqlalchemy import Column, Integer, DateTime, modifier
from sqlalchemy.orm import InstrumentedAttribute, properties
from sqlalchemy.sql import func
from sqlalchemy.ext.declarative import as_declarative, declared_attr@as_declarative()
class BaseEntity:@declared_attrdef __tablename__(cls) -> str:# 如果没有指定__tablename__  则默认使用model类名转换表名字return cls.__name__

2.t_user

用户数据表，实体定义，数据库操作的数据依据

这部分我也写了一个自动生成Entity的小应用程序，欢迎大家下载使用

https://download.csdn.net/download/rotion135/85124222

from sqlalchemy import Column, Integer, String,Boolean,DateTime,Double,Float
from Modules.db.BaseEntity import BaseEntity#t_user
class t_user(BaseEntity):ID = Column(String(32), primary_key=True,nullable=False, comment="表ID")UserCode = Column(String(32), primary_key=False,nullable=False, comment="用户编码")UserName = Column(String(128), primary_key=False,nullable=False, comment="用户名称")LoginName = Column(String(128), primary_key=False,nullable=False, comment="登录名")Password = Column(String(255), primary_key=False,nullable=False, comment="登录密码")UserStatus = Column(Integer, primary_key=False,nullable=True, comment="用户状态  默认0:正常，1:禁用 可字典项配置")IsDelete = Column(Integer, primary_key=False,nullable=True, comment="删除为1，未删除为0，默认0")Nick = Column(String(128), primary_key=False,nullable=True, comment="昵称")Title = Column(String(128), primary_key=False,nullable=True, comment="职称")Sex = Column(Integer, primary_key=False,nullable=True, comment="性别 0:男;1:女")Birthday = Column(String(32), primary_key=False,nullable=True, comment="生日")IdentityCard = Column(String(128), primary_key=False,nullable=True, comment="身份证号码")Mobile = Column(String(32), primary_key=False,nullable=True, comment="手机号")Telephone = Column(String(32), primary_key=False,nullable=True, comment="电话号码")QQ = Column(String(32), primary_key=False,nullable=True, comment="QQ号")Email = Column(String(128), primary_key=False,nullable=True, comment="邮箱")Signature = Column(String(255), primary_key=False,nullable=True, comment="个性签名")IsAdmin = Column(Integer, primary_key=False,nullable=True, comment="是否管理员  0-非管理员 1-管理员")CreateTime = Column(DateTime, primary_key=False,nullable=True, comment="创建时间")CreateUserCode = Column(String(32), primary_key=False,nullable=True, comment="创建人的用户编码")ModifyTime = Column(DateTime, primary_key=False,nullable=True, comment="修改时间")ModifyUserCode = Column(String(32), primary_key=False,nullable=True, comment="修改人的用户编码，只保存最后一次修改人")Remark = Column(String, primary_key=False,nullable=True, comment="备注")Token = Column(String(32), primary_key=False,nullable=True, comment="访问令牌")TokenExTime = Column(DateTime, primary_key=False,nullable=True, comment="访问令牌过期时间")RefreshToken = Column(String(32), primary_key=False,nullable=True, comment="刷新令牌")RefreshExTime = Column(DateTime, primary_key=False,nullable=True, comment="刷新令牌过期时间")

3.DAOBase

接下来就是设计DAO层的基类

数据链接 我用了Config.py作为配置文件，在下面一个代码块里边

from sqlalchemy import create_engine, Column, Integer, String,Engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,Session
from Modules.SysFrame import Configclass DAOBase():def __init__(self,db:Session=None):if(db==None):self.__InitEngine()else :self.session=dbengine:Enginesession:Sessiondef __InitEngine(self):# 创建一个引擎来连接到 MySQL 数据库self.engine = create_engine(Config.MySQLConnection)# 创建一个会话类sm = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)self.session = sm()

cors_allowOrigins="*"
#后台自定义的请求头参数
cors_allowHeaders="Origin,Accept,Accept-Language,Authorization,Content-Type,UserCode,Timestamp,Nonce,Signature"
#请求后台允许的方式
cors_allowMethods="GET, POST, PUT, DELETE, OPTIONS"
#登录限制时间，时间内登录次数超过设置值，则锁定
LoginLimitMinutes=60
#登录限制时间内的登陆错误次数限制
LoginLimitCount=5#MySQL数据库连接字符串
# 格式为：'数据库类型+数据库驱动名称://用户名:密码@主机地址:端口号/数据库名称'
MySQLConnection="mysql+mysqlconnector://root:123qwe@127.0.0.1:3306/platform"

4.UserDAO

最后，展示用户操作类的视线

BaseResult 是我自己定义的一个结果类，也放在下面代码块了

from datetime import datetime
from sqlalchemy import func,text
from sqlalchemy.orm import sessionmaker
from Modules.Models.BaseResult import BaseResult
from Modules.db.DAOBase import DAOBase
from Modules.db.Entity.t_user import t_userclass UserDAO(DAOBase):def __init__(self, db: sessionmaker = None):super().__init__(db)# 根据用户编码获取用户数据        def GetUserByUserCode(self,userCode:str)->t_user:sql =  self.session.query(t_user).filter(t_user.UserCode==userCode).first()return sql;# 根据登录名获取用户数据def  GetUserByLogin(self,loginName:str)->t_user:sql = self.session.query(t_user).filter(t_user.LoginName==loginName).first()return sql;# 新增前检测 true-正常，可新增 false-已存在相同数据def  CheckBeforeInsert(self,userCode:str)->BaseResult:row_count = self.session.query(func.count(t_user.ID)).filter(t_user.UserCode==userCode).scalar()if(row_count <= 0):return BaseResult.Successelse:return BaseResult.Failure# 新增用户def Insert(self,entity:t_user)->BaseResult:self.session.add(entity)self.session.commit()return BaseResult.Success# 更新用户def Update(self,entity:t_user,userCode:str)->BaseResult:user = self.session.query(t_user).filter(t_user.UserCode == entity.UserCode).first()user.UserName = entity.UserName;user.Birthday = entity.Birthday;user.IdentityCard = entity.IdentityCard;user.Email = entity.Email;user.Mobile = entity.Mobile;user.Nick = entity.Nick;user.QQ = entity.QQ;user.Sex = entity.Sex;user.Signature = entity.Signature;user.Telephone = entity.Telephone;user.Title = entity.Title;user.ModifyTime = datetime.now();user.ModifyUserCode = userCode;self.session.commit()return BaseResult.Success# 判断用户是否有此权限def JudgeUserPermit(self,type:str, method:str, userCode:str)->BaseResult:sql = text(f"select COUNT(1) FROM t_permission  WHERE ResType='{type}' and ResMethod='{method}'  and PermitCode in (SELECT PermitCode FROM t_rolepermission WHERE RoleCode in (SELECT RoleCode FROM t_userrole WHERE UserCode='{userCode}'))");result = self.session.execute(sql)count = result.fetchone()[0]if(count > 0):return BaseResult.Successelse:return BaseResult.Failure   

class BaseResult():def __init__(self,isSucessed:bool,message:str,mark=0,tag=None,total=0):self.IsSucessed=isSucessedself.Message=messageself.Mark=markself.Tag=tagself.Total=totalIsSucessed:boolMessage:strMark:intTag:anyTotal:int@staticmethoddef Success():res=BaseResult(True,"Success")return res@staticmethoddef Failure():res=BaseResult(True,"Failure")return res