系统采用操作权限与数据权限分离的设计:
| 类型 | 作用 | 控制层级 |
|---|---|---|
| 操作权限 | 控制用户能否访问某个 API | URL/菜单 |
| 数据权限 | 控制用户能看到哪些数据 | 数据库行级 |
操作权限基于 RBAC(基于角色的访问控制) 模型:
模块:资源:操作,如 module_payment:enterprise:create权限检查依赖类:AuthPermission
# app/core/dependencies.py
class AuthPermission:
def __init__(self, permissions: list[str]):
self.permissions = permissions
使用方式:通过 FastAPI 依赖注入
@EnterpriseRouter.post(
"",
summary="创建企业",
)
async def create_enterprise_controller(
data: EnterpriseCreateSchema,
auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:enterprise:create"]))],
) -> JSONResponse:
...
请求 → get_current_user() 获取用户
→ 获取用户角色
→ 获取角色关联的菜单权限列表
→ 判断所需权限是否在列表中
→ 允许/拒绝
数据权限通过 数据范围过滤 实现,确保用户只能看到自己有权限访问的数据行。
# app/common/enums.py
class PermissionFilterStrategy(str, Enum):
DATA_SCOPE = "data_scope" # 基于数据范围权限(默认)
ROLE_BASED = "role_based" # 基于角色授权(菜单)
DEPT_BASED = "dept_based" # 基于部门关联
SELF_ONLY = "self_only" # 仅本人数据
USER_ROLE = "user_role" # 当前用户绑定的角色
通过 SQLAlchemy Mixin 提供权限字段支撑:
# app/core/base_model.py
class ModelMixin(MappedBase):
"""通用字段混入"""
id: Mapped[int] = mapped_column(Integer, primary_key=True)
uuid: Mapped[str] = mapped_column(String(64), unique=True)
status: Mapped[str] = mapped_column(String(10), default="0") # 通用状态
created_time: Mapped[datetime]
updated_time: Mapped[datetime]
class TenantMixin(MappedBase):
"""租户隔离"""
tenant_id: Mapped[int] = mapped_column(ForeignKey("sys_tenant.id"))
class UserMixin(MappedBase):
"""用户审计字段"""
created_id: Mapped[int | None] = mapped_column(ForeignKey("sys_user.id"))
updated_id: Mapped[int | None] = mapped_column(ForeignKey("sys_user.id"))
# app/core/permission.py
class Permission:
"""数据权限过滤"""
def __init__(self, model: type[DeclarativeBase], auth: AuthSchema):
self.model = model
self.auth = auth
async def filter_query(self, sql: Select) -> Select:
"""根据权限策略过滤查询"""
strategy = getattr(self.model, "__permission_strategy__", PermissionFilterStrategy.DATA_SCOPE)
if strategy == PermissionFilterStrategy.SELF_ONLY:
return self._filter_self_only(sql)
elif strategy == PermissionFilterStrategy.DEPT_BASED:
return self._filter_dept_based(sql)
elif strategy == PermissionFilterStrategy.DATA_SCOPE:
return self._filter_data_scope(sql)
# ... 其他策略
return sql
CRUD 查询
│
▼
__filter_permissions(sql) # base_crud.py
│
▼
Permission(model, auth).filter_query(sql)
│
▼
根据 __permission_strategy__ 选择过滤策略
│
├── DATA_SCOPE: 根据角色 data_scope 字段过滤
├── SELF_ONLY: 仅返回 created_id = 当前用户ID 的数据
├── DEPT_BASED: 根据部门及子部门过滤
└── ...
│
▼
返回带 WHERE 条件的 SQL
# app/core/base_crud.py
class CRUDBase:
async def get(self, id: int, preload: list[str] | None = None) -> Model | None:
sql = select(self.model).where(self.model.id == id)
sql = await self.__filter_permissions(sql) # 自动注入权限过滤
...
async def page(self, offset, limit, search, order_by, out_schema, preload):
sql = select(self.model)
sql = await self.__filter_permissions(sql) # 自动注入权限过滤
...
# 示例:在模型中指定权限策略
class EnterpriseModel(ModelMixin):
__tablename__ = "t_enterprise"
__permission_strategy__ = PermissionFilterStrategy.DATA_SCOPE # 默认
# 字段定义...
# app/api/v1/module_system/auth/schema.py
class AuthSchema(BaseModel):
id: int # 用户ID
username: str # 用户名
tenant_id: int # 租户ID
is_superuser: bool # 是否超级管理员
role_ids: list[int] # 角色ID列表
dept_id: int | None # 部门ID
data_scope: int # 数据范围(1-5)
| data_scope 值 | 说明 | 过滤逻辑 |
|---|---|---|
| 1 | 仅本人 | created_id = 当前用户ID |
| 2 | 本部门 | dept_id = 用户部门ID |
| 3 | 本部门及以下 | dept_id IN (用户部门ID, 子部门...) |
| 4 | 全部数据 | 不过滤 |
| 5 | 自定义 | 按 dept_id_list 过滤 |
@EnterpriseRouter.post(
"",
summary="创建企业",
response_model=ResponseSchema[EnterpriseOutSchema],
)
async def create_enterprise_controller(
data: EnterpriseCreateSchema,
# 1. 操作权限:需要 module_payment:enterprise:create 权限
auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:enterprise:create"]))],
) -> JSONResponse:
# 2. 数据权限:在 CRUD 中自动处理
result = await EnterpriseService.create_enterprise_service(auth=auth, data=data)
return SuccessResponse(data=result)
class SensitiveModel(ModelMixin):
__tablename__ = "t_sensitive"
__permission_strategy__ = PermissionFilterStrategy.SELF_ONLY # 仅本人
...
__permission_strategy__ 可为不同模型配置不同策略is_superuser=True 的用户跳过数据权限过滤