""" 权限管理模型 支持RBAC(基于角色的访问控制) """ from sqlalchemy import Column, String, DateTime, JSON, ForeignKey, Boolean, Integer, Table, func from sqlalchemy.dialects.mysql import CHAR from sqlalchemy.orm import relationship from app.core.database import Base import uuid # 用户角色关联表(多对多) user_roles = Table( 'user_roles', Base.metadata, Column('user_id', CHAR(36), ForeignKey('users.id', ondelete='CASCADE'), primary_key=True), Column('role_id', CHAR(36), ForeignKey('roles.id', ondelete='CASCADE'), primary_key=True) ) # 角色权限关联表(多对多) role_permissions = Table( 'role_permissions', Base.metadata, Column('role_id', CHAR(36), ForeignKey('roles.id', ondelete='CASCADE'), primary_key=True), Column('permission_id', CHAR(36), ForeignKey('permissions.id', ondelete='CASCADE'), primary_key=True) ) class Role(Base): """角色表""" __tablename__ = "roles" id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="角色ID") name = Column(String(50), unique=True, nullable=False, comment="角色名称") description = Column(String(255), comment="角色描述") is_system = Column(Boolean, default=False, comment="是否系统角色(不可删除)") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") # 关系 users = relationship("User", secondary=user_roles, back_populates="roles") permissions = relationship("Permission", secondary=role_permissions, back_populates="roles") def __repr__(self): return f"" class Permission(Base): """权限表""" __tablename__ = "permissions" id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="权限ID") name = Column(String(100), unique=True, nullable=False, comment="权限名称") code = Column(String(100), unique=True, nullable=False, comment="权限代码(如:workflow:create)") resource = Column(String(50), nullable=False, comment="资源类型(如:workflow、agent、execution)") action = Column(String(50), nullable=False, comment="操作类型(如:create、read、update、delete、execute)") description = Column(String(255), comment="权限描述") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") # 关系 roles = relationship("Role", secondary=role_permissions, back_populates="permissions") def __repr__(self): return f"" class WorkflowPermission(Base): """工作流权限表(细粒度权限控制)""" __tablename__ = "workflow_permissions" id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="权限ID") workflow_id = Column(CHAR(36), ForeignKey("workflows.id", ondelete='CASCADE'), nullable=False, comment="工作流ID") user_id = Column(CHAR(36), ForeignKey("users.id", ondelete='CASCADE'), nullable=True, comment="用户ID(null表示所有用户)") role_id = Column(CHAR(36), ForeignKey("roles.id", ondelete='CASCADE'), nullable=True, comment="角色ID(null表示所有角色)") permission_type = Column(String(20), nullable=False, comment="权限类型:read/write/execute/share") granted_by = Column(CHAR(36), ForeignKey("users.id"), nullable=False, comment="授权人ID") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") # 关系 workflow = relationship("Workflow", backref="permissions") user = relationship("User", foreign_keys=[user_id], backref="workflow_permissions") role = relationship("Role", backref="workflow_permissions") grantor = relationship("User", foreign_keys=[granted_by]) def __repr__(self): return f"" class AgentPermission(Base): """Agent权限表(细粒度权限控制)""" __tablename__ = "agent_permissions" id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="权限ID") agent_id = Column(CHAR(36), ForeignKey("agents.id", ondelete='CASCADE'), nullable=False, comment="Agent ID") user_id = Column(CHAR(36), ForeignKey("users.id", ondelete='CASCADE'), nullable=True, comment="用户ID(null表示所有用户)") role_id = Column(CHAR(36), ForeignKey("roles.id", ondelete='CASCADE'), nullable=True, comment="角色ID(null表示所有角色)") permission_type = Column(String(20), nullable=False, comment="权限类型:read/write/execute/deploy") granted_by = Column(CHAR(36), ForeignKey("users.id"), nullable=False, comment="授权人ID") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") # 关系 agent = relationship("Agent", backref="permissions") user = relationship("User", foreign_keys=[user_id], backref="agent_permissions") role = relationship("Role", backref="agent_permissions") grantor = relationship("User", foreign_keys=[granted_by]) def __repr__(self): return f""