目录
刚开始用Python做Api后端,记录下学习过程
从数据库操作开始,定义实体,定义DAO层基类,及DAO业务类的使用
写的不好,有好的设计建议欢迎大家提出,一起学习进步
这部分是实体的基类,比较简单,就一个使用类名作为表名的方法
-
- 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_attr
- def __tablename__(cls) -> str:
- # 如果没有指定__tablename__ 则默认使用model类名转换表名字
- return cls.__name__
用户数据表,实体定义,数据库操作的数据依据
这部分我也写了一个自动生成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="刷新令牌过期时间")
-
-
接下来就是设计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 Config
-
-
- class DAOBase():
- def __init__(self,db:Session=None):
- if(db==None):
- self.__InitEngine()
- else :
- self.session=db
-
-
-
- engine:Engine
- session:Session
-
-
- def __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"
-
-
-
-
-
最后,展示用户操作类的视线
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_user
-
- class 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.Success
- else:
- 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.engine.execute(sql)
- count = result.fetchone()[0]
- if(count > 0):
- return BaseResult.Success
- else:
- return BaseResult.Failure
-
- class BaseResult():
- def __init__(self,isSucessed:bool,message:str,mark=0,tag=None,total=0):
- self.IsSucessed=isSucessed
- self.Message=message
- self.Mark=mark
- self.Tag=tag
- self.Total=total
-
-
-
- IsSucessed:bool
- Message:str
- Mark:int
- Tag:any
- Total:int
-
- @staticmethod
- def Success():
- res=BaseResult(True,"Success")
- return res
-
- @staticmethod
- def Failure():
- res=BaseResult(True,"Failure")
- return res