flask-sqlAlchemy后端的MVC再探

0.对于MVC陷入了蜜汁执着，继续对mvc这种目录的软件结构持续跟踪。

1.上回说到要研究权限相关内容，搜寻了一番，发现米国在上世纪已经展开了详细研究，并提出了RBAC权限体系，并把RBAC划分为三个层级。百度百科上有相关论述。因此下面我会基于RBAC展开编码。

2.首先定义出角色和用户，角色和用户之间是多对多关系，即，一个用户可能同时拥有多种角色(这种情况，在多数系统是不会出现，多数都是一个用户只能拥有一个角色)；一个角色被多个用户拥有。于是产生映射关系表。user,role,user-role-map.三张表就这么产生。

3.首先创建三个model。然后让sqlAlchemy自动在数据库引擎中生成。上次说到我在犹豫是否把dao层拆分出model层，这次我决定拆出来了，于是我的软件结构，就变成了，model-dao-service-controller-view 算是标准的五层结构。
flask-sqlAlchemy后端的MVC再探

4.在sqlAlchemy官方文档中，建议创建外键，以方便查询。从我往日习惯来说，我不喜欢使用外键这种方式，所以我的model中没有任何外键，只有约束。对于数据库来说，约束，外键，主键，索引，联合查询，等等概念需要一段时间学习，方可不迷茫。
　　4.1 model层内定义下列表模型

from dao.database import Base
from sqlalchemy import ForeignKey,Column, Integer, String, Date,BigInteger,DateTime,Boolean

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True,comment='主键')
    other_id = Column(String(200), unique=True,nullable=False,comment='对外ID')
    username = Column(String(50), unique=True,nullable=False,comment='用户名')
    password = Column(String(200), unique=False,nullable=False,comment='密码')
    status = Column(Integer,nullable=False,default=1,comment='1:启用；2:停用')

    def __repr__(self):
      return "<User(id='%d',otherid='%s',username='%s', password='%s', status='%d' )>" % 
              (self.id, self.other_id, self.username, self.password, self.status)

user.py

from sqlalchemy import ForeignKey,Column, Integer, String, Date,BigInteger,DateTime from sqlalchemy.orm import relationship,backref from dao.database import Base class Role(Base): __tablename__ = 'role' id = Column(Integer, primary_key=True) rolename = Column(String(50), unique=True,nullable=False) desc = Column(String(200), nullable=True) def __repr__(self): return "<role(id='%s', rolename='%s', desc='%s')>" % (self.id, self.rolename, self.desc)

from dao.database import Base from sqlalchemy import ForeignKey,Column, Integer, UniqueConstraint class UserRoleMap(Base): __tablename__ = 'user_role_map' id = Column(Integer, primary_key=True) user_id = Column(Integer, unique=False,nullable=False) role_id = Column(Integer, unique=False,nullable=False) __table_args__ = ( UniqueConstraint("user_id", "role_id"), ) def __repr__(self): return "<role(id='%s', userid='%s', roleid='%s')>" % (self.id, self.user_id, self.role_id)

from .database import db_session from model.user import User import uuid from werkzeug.security import check_password_hash,generate_password_hash from functools import singledispatch # from model.user import User def insert(username,password): #如何处理异常将是需要考虑的 other_id = str(uuid.uuid1()) password = generate_password_hash(username + password) #加盐加密函数，通过随机产生不同salt(盐粒)混入原文，使每次产生的密文不一样。 user = User(other_id=other_id, username=username, password=password, status=1) db_session.add(user) db_session.commit() return user #---------尝试函数重载--------------- @singledispatch def delete(): return False @delete.register(int) def _(id): user = db_session.query(User).get(id) db_session.delete(user) db_session.commit() return True @delete.register(User) def _(user): db_session.delete(user) db_session.commit() return True #---------尝试函数重载--------------- def find(id): return db_session.query(User).filter_by(id=id).one()

from .database import db_session from model.role import Role def insert(rolename,desc): role = Role(rolename=rolename, desc=desc) db_session.add(role) db_session.commit() return role def delete(id): role = db_session.query(Role).get(id) db_session.delete(role) db_session.commit() return True def find(id): return db_session.query(Role).filter_by(id=id).one()

from .database import db_session from model.user_role_map import UserRoleMap def insert(userid,roleid): userRoleMap = UserRoleMap(user_id=userid, role_id=roleid) db_session.add(userRoleMap) db_session.commit() return userRoleMap def delete(id): userRoleMap = db_session.query(UserRoleMap).get(id) db_session.delete(userRoleMap) db_session.commit() return True def showTable(): return UserRoleMap.__table__

from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.ext.declarative import declarative_base # engine = create_engine("mysql+mysqlconnector://root:123@localhost:3306/hello_login", encoding="utf-8",echo=True) engine = create_engine('mssql+pymssql://sa:123@localhost:1433/hello_Login', encoding="utf-8",echo=True) db_session = scoped_session(sessionmaker(autocommit=False,autoflush=False,bind=engine)) Base = declarative_base() Base.query = db_session.query_property() # @on_request_end # def remove_session(req): # db_session.remove() def init_db(): import model Base.metadata.create_all(bind=engine)

# from dao import user from flask import jsonify import dao.user import dao.role import dao.user_role_map import dao.access_control def insertUser(): try: user = dao.user.insert(username='test1', password='123') returnData = {'code': 0, 'msg': 'success', 'data': 'insert ' + str(user)} except Exception as ex: returnData = {'code': 1, 'msg': 'failer', 'data': '%s' %ex} return jsonify(returnData),200 def deleteUser(): user = dao.user.find(25) dao.user.delete(user) returnData = {'code': 0, 'msg': 'success', 'data': 'delete success'} return jsonify(returnData),200 def insertRole(): role = dao.role.insert(rolename='系统管理员', desc='具有最高权限的角色') returnData = {'code': 0, 'msg': 'success', 'data': 'insert '+ str(role)} return jsonify(returnData),200 def deleteRole(): dao.role.delete(1) returnData = {'code': 0, 'msg': 'success', 'data': 'delete success'} return jsonify(returnData),200 def bindUserAndRole(): userRoleMap = dao.user_role_map.insert(26,2) returnData = {'code': 0, 'msg': 'success', 'data': 'insert '+ str(userRoleMap)} return jsonify(returnData),200 def table(): info = dao.user_role_map.showTable() return info def findUserRole(): dao.access_control.findUserRole() return 'nihao'

from flask import Blueprint,jsonify import service.access_control as accessControl bp = Blueprint('user_page',__name__) @bp.route('/user/insert') def insertUser(): return accessControl.insertUser() @bp.route('/user/delete') def deleteUser(): return accessControl.deleteUser() @bp.route('/role/insert') def insertRole(): return accessControl.insertRole() @bp.route('/role/delete') def deleteRole(): return accessControl.deleteRole() @bp.route('/binduserandrole') def bindUserAndRole(): return accessControl.bindUserAndRole() @bp.route('/tableinfo') def tableInfo(): print(accessControl.table()) return str(accessControl.table()) @bp.route('/finduserrole') def findUserRole(): accessControl.findUserRole() return 'aslkdf'

import os import importlib def get_modules(package="."): """ 获取包名下所有非__init__的模块名 """ modules = [] files = os.listdir(package) for file in files: if not file.startswith("__"): name, ext = os.path.splitext(file) modules.append("." + name) return modules def init_app(app): with app.app_context(): # from .home import bp,helloworld # app.register_blueprint(bp) # app.add_url_rule("/", view_func=helloworld) # from .login import bp # app.register_blueprint(bp) # from .testdb import bp # app.register_blueprint(bp) # from .usermanage import bp # app.register_blueprint(bp) # from .access_control import bp # app.register_blueprint(bp) modules = get_modules('hellologin/controller') for module in modules: module = importlib.import_module(module, 'hellologin.controller') app.register_blueprint(module.bp)

from .database import db_session from model.role import Role from model.user import User from model.user_role_map import UserRoleMap from sqlalchemy import and_ def findUserRole0(): dataTable = db_session.query(User,Role). filter(User.id==UserRoleMap.user_id). filter(Role.id==UserRoleMap.role_id).all() print(dataTable) return True def findUserRole(): dataTable = db_session.query(User.id,User.username,Role.rolename).select_from(User). outerjoin(UserRoleMap,and_(User.id==UserRoleMap.user_id,User.id==26)). outerjoin(Role,Role.id==UserRoleMap.role_id).filter(User.id==26).all() print(dataTable) ''' SELECT [user].id AS user_id, [user].username AS user_username, role.rolename AS role_rolename FROM [user] LEFT OUTER JOIN [user_role_map] ON [user].id = [user_role_map].user_id AND [user].id = 26 LEFT OUTER JOIN [role] ON [role].id = user_role_map.role_id WHERE [user].id = 26 ''' return True

flask-sqlAlchemy后端的MVC再探

相关推荐