ORM框架SQLAlchemy与权限管理系统的数据库设计
Posted 不辣不辣
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ORM框架SQLAlchemy与权限管理系统的数据库设计相关的知识，希望对你有一定的参考价值。
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用对象关系映射进行数据库操作，即：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
执行流程：
1、使用者通过ORM对象提交命令
2、将命令交给SQLAlchemy Core（Schema/Types SQL Expression Language）转换成SQL
3、使用 Engine/ConnectionPooling/Dialect 进行数据库操作
4、匹配使用者事先配置好的egine
5、egine从连接池中取出一个链接
6、基于该链接通过Dialect调用DB API，将SQL转交给它去执行
1、关于DB API
SQLAlchemy本身无法操作数据库，其必须依赖pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作。
1、mysql-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
2、pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
3、MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
4、cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
2、创建表操作
在ORM框架下的Alchemy中，类==>表table 对象==>数据行column ；
四张表:业务线,服务,用户,角色，利用ORM创建出它们，并建立好它们直接的关系
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,DateTime,Enum,ForeignKey,UniqueConstraint,ForeignKeyConstraint,Index from sqlalchemy.orm import sessionmaker egine=create_engine(\'mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8\',max_overflow=5) Base=declarative_base() #创建单表:业务线 class Business(Base): __tablename__=\'business\' id=Column(Integer,primary_key=True,autoincrement=True) bname=Column(String(32),nullable=False,index=True) #多对一:多个服务可以属于一个业务线,多个业务线不能包含同一个服务 class Service(Base): __tablename__=\'service\' id=Column(Integer,primary_key=True,autoincrement=True) sname=Column(String(32),nullable=False,index=True) ip=Column(String(15),nullable=False) port=Column(Integer,nullable=False) business_id=Column(Integer,ForeignKey(\'business.id\')) __table_args__=( UniqueConstraint(ip,port,name=\'uix_ip_port\'), Index(\'ix_id_sname\',id,sname) ) #一对一:一种角色只能管理一条业务线,一条业务线只能被一种角色管理 class Role(Base): __tablename__=\'role\' id=Column(Integer,primary_key=True,autoincrement=True) rname=Column(String(32),nullable=False,index=True) priv=Column(String(64),nullable=False) business_id=Column(Integer,ForeignKey(\'business.id\'),unique=True) #多对多:多个用户可以是同一个role,多个role可以包含同一个用户 class Users(Base): __tablename__=\'users\' id=Column(Integer,primary_key=True,autoincrement=True) uname=Column(String(32),nullable=False,index=True) class Users2Role(Base): __tablename__=\'users2role\' id=Column(Integer,primary_key=True,autoincrement=True) uid=Column(Integer,ForeignKey(\'users.id\')) rid=Column(Integer,ForeignKey(\'role.id\')) __table_args__=( UniqueConstraint(uid,rid,name=\'uix_uid_rid\'), ) def init_db(): Base.metadata.create_all(egine) def drop_db(): Base.metadata.drop_all(egine) if __name__ == \'__main__\': init_db()
3、对于表的增删改查
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker egine=create_engine(\'mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8\',max_overflow=5) Base=declarative_base() class Dep(Base): __tablename__=\'dep\' id=Column(Integer,primary_key=True,autoincrement=True) dname=Column(String(64),nullable=False,index=True) class Emp(Base): __tablename__=\'emp\' id=Column(Integer,primary_key=True,autoincrement=True) ename=Column(String(32),nullable=False,index=True) dep_id=Column(Integer,ForeignKey(\'dep.id\')) def init_db(): Base.metadata.create_all(egine) def drop_db(): Base.metadata.drop_all(egine) drop_db() init_db() Session=sessionmaker(bind=egine) session=Session()
【增】
有两种row增加的方式：单行增加session.add( )；多行增加session.addall( )
#增 row_obj=Dep(dname=\'销售\') #按关键字传参,无需指定id,因其是自增长的 session.add(row_obj) session.add_all([ Dep(dname=\'技术\'), Dep(dname=\'运营\'), Dep(dname=\'人事\'), ]) session.commit()
【删】
session.query(Dep).filter(Dep.id > 3).delete()
session.commit()
【改】
#改 session.query(Dep).filter(Dep.id > 0).update({\'dname\':\'哇哈哈\'}) session.query(Dep).filter(Dep.id>0).update ({\'dname\':Dep.dname+\'_SB\'},synchronize_session=False) session.query(Dep).filter(Dep.id>0).update ({\'id\':Dep.id*100},synchronize_session=\'evaluate\') session.commit()
使用Alchemy 进行增、删、改的操作时最后都需要进行向数据库提交数据：session.commit( )
【查】
#查所有,取所有字段 res=session.query(Dep).all() #for row in res:print(row.id,row.dname) #查所有,取指定字段 res=session.query(Dep.dname).order_by(Dep.id).all() #for row in res:print(row.dname) res=session.query(Dep.dname).first() print(res) # (\'哇哈哈_SB\',) #过滤查 res=session.query(Dep).filter(Dep.id > 1,Dep.id <1000) #逗号分隔,默认为and print([(row.id,row.dname) for row in res])
4、子查询
个人感觉使用连表操作能很大程度代替子查询的工作量，而且Alchemy实现起来也很方便！
详细内容参考海峰老师博客，链接：http://www.cnblogs.com/linhaifeng/articles/7560153.html
5、权限管理系统的设计
1. 登陆、注册、找回密码
2. 权限管理
3. 角色管理
4. 角色分配权限
5. 动态显示当前登陆用户权限菜单
1 from sqlalchemy import create_engine 2 from sqlalchemy.ext.declarative import declarative_base 3 from sqlalchemy import Column,Integer,ForeignKey,CHAR,VARCHAR,UniqueConstraint 4 from sqlalchemy.orm import sessionmaker 5 #建立链接 6 engine = create_engine("mysql+pymysql://root:761379545@127.0.0.1:3306/test_db?charset=utf8", max_overflow=5) 7 Base = declarative_base() 8 Session = sessionmaker(bind=engine) 9 session = Session() 10 11 #创建用户信息表 12 class User(Base): 13 __tablename__ = \'user\' 14 uid = Column(Integer,primary_key=True,autoincrement=True) 15 name = Column(CHAR(20),nullable=False) 16 pwd = Column(CHAR(20),nullable=False) 17 roll_id = Column(Integer,ForeignKey(\'roll.rid\')) 18 #__table_args__的参数必须是元组的形式 19 __table_args__ = ( 20 UniqueConstraint(name,pwd,name=\'uni_np\'), #用户名和密码不能重复！ 21 ) 22 23 #创建权限表 24 class Authority(Base): 25 __tablename__ =\'authority\' 26 aid = Column(Integer,primary_key=True,autoincrement=True) 27 auth = Column(VARCHAR(255),nullable=True) 28 29 #创建角色表 30 class Roll(Base): 31 __tablename__ =\'roll\' 32 rid = Column(Integer,primary_key=True,autoincrement=True) 33 dep = Column(VARCHAR(255),nullable=False) 34 35 #角色权限关系表 36 class Ro_au(Base): 37 __tablename__ =\'ro_au\' 38 id = Column(Integer,primary_key=True,autoincrement=True) 39 rid = Column(Integer,ForeignKey(\'roll.rid\')) 40 aid = Column(Integer,ForeignKey(\'authority.aid\')) 41 42 #创建原始关系表 43 def original_table(): 44 session.add_all([ 45 Roll(dep=\'项目经理\'), 46 Roll(dep=\'技术部\'), 47 Roll(dep=\'运营部\'), 48 Roll(dep=\'人事部\'), 49 Roll(dep=\'后勤部\'), 50 Roll(dep=\'普通员工\'), 51 Authority(auth=\'查看个人信息\'), 52 Authority(auth=\'更改用户信息\'), 53 Authority(auth=\'删除用户\'), 54 Authority(auth=\'设定项目整体方案\'), 55 Authority(auth=\'查看项目进度\'), 56 Authority(auth=\'查看项目经费\'), 57 Authority(auth=\'查看项目计划\'), 58 Authority(auth=\'项目技术方案\'), 59 Authority(auth=\'项目支出管理\'), 60 Authority(auth=\'项目人事安排\'), 61 Authority(auth=\'项目后勤保障\'), 62 ]) 63 session.commit() 64 session.add_all([ 65 User(name=\'alex\',pwd=\'3714\',roll_id=1), 66 User(name=\'egon\',pwd=\'123456\',roll_id=2), 67 User(name=\'sl\',pwd=\'761379545\',roll_id=2), 68 User(name=\'eric\',pwd=\'123456\',roll_id=3), 69 User(name=\'mick\',pwd=\'123456\',roll_id=4), 70 User(name=\'john\',pwd=\'123456\',roll_id=5), 71 User(name=\'kaka\',pwd=\'123456\',roll_id=6), 72 ]) 73 session.add_all([ 74 Ro_au(rid=1,aid=1),Ro_au(rid=1,aid=5),Ro_au(rid=1,aid=6), 75 Ro_au(rid=2, aid=1),Ro_au(rid=2, aid=4),Ro_au(rid=2, aid=8), 76 Ro_au(rid=3, aid=1),Ro_au(rid=3, aid=5),Ro_au(rid=3, aid=6),Ro_au(rid=3, aid=7), 77 Ro_au(rid=4, aid=1),Ro_au(rid=4, aid=2),Ro_au(rid=4, aid=3), 78 Ro_au(rid=5, aid=1),Ro_au(rid=5, aid=9),Ro_au(rid=5, aid=11), 79 Ro_au(rid=6, aid=1), 80 ]) 81 session.commit() 82 83 #创建数据库所有表 84 def creat_db(): 85 Base.metadata.create_all(engine) 86 87 #删除数据库所用表 88 def drop_db(): 89 Base.metadata.drop_all(engine) 90 91 #显示个人权限 92 def show_auth(user_name): 93 user_roll_id = session.query(User.roll_id).filter(User.name == user_name).subquery() 94 user_dep_list = session.query(Roll.dep).filter(Roll.rid == user_roll_id).all() 95 for row in user_dep_list: 96 user_dep = row[0] 97 print(\'您当前所属部门名称：\', user_dep) 98 print(\'\\33[33m您具有的权限如下:\\33[37m\') 99 auth_list = session.query(Authority.auth).join(Ro_au).filter(Ro_au.rid == user_roll_id).all() 100 count = 1 101 for auth in auth_list: 102 print(count, auth[0]) 103 count += 1 104 105 #显示所有部门信息 106 def show_department(): 107 print(\'公司现有部门如下：\') 108 dep_list = session.query(Roll).all() 109 for dep in dep_list: 110 print(dep[0],dep[1]) 111 112 113 def login(): 114 while True: 115 user_name = input(\'请输入用户名【返回B】：\\n>>>\').strip() 116 if user_name.lower() == \'b\': 117 break 118 user_pwd = input(\'请输入用户密码【返回B】：\\n>>>\').strip() 119 if user_pwd.lower() == \'b\': 120 break 121 user_list = session.query(User.name,User.pwd).all() 122 123 if (user_name,user_pwd) in user_list: 124 print(\'登陆成功！\') 125 choice = input(\'【1、进入个人权限界面】\' 126 \'\\n【2、查看公司具有的所有部门】\' 127 \'\\n【B、返回主界面】\' 128 \'\\n>>>\').strip() 129 if choice == \'1\': 130 show_auth(user_name) 131 continue 132 if choice == \'2\': 133 show_department() 134 continue 135 if choice.lower() == \'b\': 136 break 137 else: 138 print(\'用户名或者密码错误！\') 139 continue 140 141 #用户注册 142 def register(): 143 while True: 144 user_name = input(\'请输入用户名【返回B】：\\n>>>\').strip() 145 if not user_name:continue 146 if user_name.lower() == \'b\': 147 break 148 name_list = session.query(User.name).all() 149 if (user_name,) in name_list: 150 print(\'该用户名已经存在！\') 151 break 152 else: 153 user_pwd = input(\'请输入用户密码：\\n>>>\').strip() 154 if not user_pwd:continue 155 dep_list = session.query(Roll.rid,Roll.dep).all() 156 for row in dep_list: 157 print(row.rid,row.dep) 158 user_roll = input(\'请输入所属部门序号：\\n>>>\').strip() 159 if not user_roll: 160 print(\'部门序号不能为空，请重新开始！\') 161 continue 162 if user_roll.isdigit(): 163 session.add(User(name=user_name,pwd=user_pwd,roll_id=int(user_roll))) #写入数据库 164 session.commit() 165 print(\'新的员工信息已经注册成功！\') 166 break 167 else: 168 print(\'您选择的部门有误，请重新选择！\') 169 170 #找回密码 171 def find_pwd(): 172 while True: 173 user_name =input(\'请输入用户名【返回B】：\').strip() 174 if not user_name:continue 175 if user_name.lower() == \'b\': 176 break 177 name_list =session.query(User.name).all() 178 if (user_name,) in name_list: 179 pwd_list =session.query(User.pwd).filter(User.name==user_name).all() 180 print(\'您的密码为：\') 181 for pwd in pwd_list: 182 print(pwd[0]) 183 else: 184 print(\'您输入的用户名不存在！\') 185 186 if __name__ == \'__main__\': 187 while True: 188 creat_tab_choice = input(\'是否要在数据库中创建原始关系表？第一次请选择是！\' 189 \'**此操作会删除已添加的数据！**【是Y否N】：\\n>>>\').strip() 190 if creat_tab_choice.lower() == \'y\': 191 drop_db() 192 creat_db() 193 original_table() 194 print(\'基本权限管理原始表已创建！\') 195 elif creat_tab_choice.lower() == \'n\': 196 pass 197 else: 198 print(\'您的操作有误，请重新选择！\') 199 continue 200 while True: 201 print(\'欢迎进入权限管理系统！\'.center(21, \'*\')) 202 choice = input(\'登录L\\t注册R\\t找回密码F\\t退出Q:\\n>>>\').strip() 203 if choice.lower() == \'q\': 204 print(Python ORM框架之SQLALchemy