# # 数据模型配置文件,注意:与数据模型对应的表名称来自配置文件。 # 若使用 生成代码,则这部分表将不会自动生成。 # from sqlalchemy import Column, String, DateTime, LargeBinary, text, BigInteger, Integer, SmallInteger, Text, ForeignKey from sqlalchemy.orm import relationship from paste.core import config from paste.db.basemodel import BaseModel class RbacRuleModel(BaseModel): __tablename__ = config.get_config('rbac.table.rule') __table_args__ = {'comment': '规则'} name = Column(String(64, 'utf8mb4_unicode_ci'), primary_key=True, comment='名称') data = Column(LargeBinary, comment='规则对象') created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='创建时间') updated_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='更新时间') class RbacUserModel(BaseModel): __tablename__ = config.get_config('rbac.table.user') __table_args__ = {'comment': '用户'} id = Column(BigInteger, primary_key=True, comment='系统编号') username = Column(String(255, 'utf8mb4_unicode_ci'), nullable=False, unique=True, comment='用户名') password_hash = Column(String(255, 'utf8mb4_unicode_ci'), nullable=False, comment='密码') password_reset_token = Column(String(255, 'utf8mb4_unicode_ci'), comment='重置标记') auth_key = Column(String(255, 'utf8mb4_unicode_ci'), comment='授权码') status = Column(Integer, nullable=False, server_default=text("'0'"), comment='用户状态') type = Column(String(64, 'utf8mb4_unicode_ci'), nullable=False, server_default=text("'user'"), comment='用户类型') created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='创建时间') updated_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='更新时间') def before_save(self): super().before_save() if self.is_new: self.status = 1 class RbacItemModel(BaseModel): __tablename__ = config.get_config('rbac.table.item') __table_args__ = {'comment': '授权项(角色/权限)'} name = Column(String(64, 'utf8mb4_unicode_ci'), primary_key=True, comment='名称') type = Column(SmallInteger, nullable=False, comment='类型,1角色,2权限') description = Column(Text(collation='utf8mb4_unicode_ci'), comment='描述') rule_name = Column( ForeignKey(f"{config.get_config('rbac.table.rule')}.name", ondelete='SET NULL', onupdate='CASCADE'), index=True, comment='规则名称' ) created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='创建时间') updated_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='更新时间') rbac_rule = relationship(RbacRuleModel) class RbacAssignmentModel(BaseModel): __tablename__ = config.get_config('rbac.table.assignment') __table_args__ = {'comment': '权限分配'} item_name = Column( ForeignKey(f"{config.get_config('rbac.table.item')}.name", ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False, comment='授权项(角色/权限)' ) user_id = Column( ForeignKey(f"{config.get_config('rbac.table.user')}.id", ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False, index=True, comment='用户' ) created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='创建时间') rbac_item = relationship(RbacItemModel) rbac_user = relationship(RbacUserModel) class RbacItemChildModel(BaseModel): __tablename__ = config.get_config('rbac.table.item_child') __table_args__ = {'comment': '授权关系'} parent = Column( ForeignKey(f"{config.get_config('rbac.table.item')}.name", ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False, comment='角色/权限' ) child = Column( ForeignKey(f"{config.get_config('rbac.table.item')}.name", ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False, index=True, comment='授权项(角色/权限)' ) created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment='创建时间') child_item = relationship(RbacItemModel, primaryjoin='RbacItemChildModel.child == RbacItemModel.name') parent_item = relationship(RbacItemModel, primaryjoin='RbacItemChildModel.parent == RbacItemModel.name')