BaseService

概述

BaseService 是所有业务 Service 的基类，基于泛型设计，提供完整的 CRUD、Excel 导入导出、数据权限、字段权限、资源自动注册等能力。

泛型定义

class BaseService(Generic[T, CreateSchema, UpdateSchema]):
    """
    T           - SQLAlchemy 模型类（继承 BaseModel）
    CreateSchema - Pydantic 创建 Schema
    UpdateSchema - Pydantic 更新 Schema
    """
    
    # 子类必须定义
    model: ClassVar[Type[DBBaseModel]]
    
    # 可选配置
    RESOURCE_TYPE: ClassVar[Optional[str]] = None          # 资源类型，不定义则自动生成
    RESOURCE_DISPLAY_NAME: ClassVar[Optional[str]] = None  # 资源显示名称
    FIELD_METADATA: ClassVar[Dict] = {}                    # 字段元数据，不定义则自动生成
    excel_columns: ClassVar[Dict[str, str]]                # Excel 列映射
    excel_sheet_name: ClassVar[str]                        # Excel Sheet 名称

自动注册机制

当子类继承 BaseService 并定义了 model 属性时，会自动触发：

1. 生成 RESOURCE_TYPE：根据表名自动生成（去除 core_/sys_ 等前缀）
2. 生成 FIELD_METADATA：从模型列定义自动提取字段元数据
3. 注册到 ResourceRegistry：将 Service 注册到全局资源注册表

def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    if hasattr(cls, 'model') and cls.model is not None:
        # 自动生成资源类型
        if cls.RESOURCE_TYPE is None:
            cls.RESOURCE_TYPE = auto_generate_resource_type(cls.model)
        # 自动生成字段元数据
        if not cls.FIELD_METADATA:
            cls.FIELD_METADATA = auto_generate_field_metadata(cls.model)
        # 注册到资源注册表
        ResourceRegistry.register(
            resource_type=cls.RESOURCE_TYPE,
            service_class=cls,
            display_name=cls.RESOURCE_DISPLAY_NAME
        )

CRUD 方法

create - 创建

@classmethod
async def create(
    cls,
    db: AsyncSession,
    data: CreateSchema,
    auto_commit: bool = True,
    current_user_id: Optional[str] = None
) -> Any:

• 自动填充 sys_creator_id、sys_dept_id（从请求上下文获取）
• auto_commit=False 用于事务场景

get_by_id - 查询单条

@classmethod
async def get_by_id(cls, db: AsyncSession, record_id: str) -> Optional[Any]:

• 自动排除已删除记录（is_deleted=False）

get_list - 分页查询

@classmethod
async def get_list(
    cls,
    db: AsyncSession,
    page: int = 1,
    page_size: int = 20,
    filters: Optional[List[Any]] = None
) -> Tuple[List[Any], int]:

• 返回 (items, total) 元组
• 默认按 sort DESC, sys_create_datetime DESC 排序
• 支持额外过滤条件

update - 更新

@classmethod
async def update(
    cls,
    db: AsyncSession,
    record_id: str,
    data: UpdateSchema,
    auto_commit: bool = True,
    current_user_id: Optional[str] = None
) -> Optional[Any]:

• 使用 exclude_unset=True，仅更新传入的字段
• 自动填充 sys_modifier_id

delete - 删除

@classmethod
async def delete(
    cls,
    db: AsyncSession,
    record_id: str,
    hard: bool = False,
    auto_commit: bool = True
) -> bool:

• hard=False：软删除（设置 is_deleted=True）
• hard=True：物理删除

batch_delete - 批量删除

@classmethod
async def batch_delete(
    cls,
    db: AsyncSession,
    ids: List[str],
    hard: bool = False,
    auto_commit: bool = True
) -> Tuple[int, int]:  # (success_count, fail_count)

数据权限方法

get_list_with_data_scope

带数据权限的分页查询：

@classmethod
async def get_list_with_data_scope(
    cls,
    db: AsyncSession,
    page: int = 1,
    page_size: int = 20,
    filters: Optional[List[Any]] = None,
    data_scope: Optional[Dict] = None,
    dept_field: str = "sys_dept_id",
    user_field: str = "sys_creator_id"
) -> Tuple[List[Any], int]:

数据权限类型：

scope_type	说明	过滤逻辑
all	全部数据	无过滤
self	仅本人	user_field == current_user_id
dept	本部门	dept_field == current_dept_id
dept_and_children	本部门及子部门	dept_field IN (dept_ids)
custom	自定义部门	dept_field IN (custom_dept_ids)

update_with_data_scope / delete_with_data_scope

操作前先检查数据权限，无权限时返回 None / False。

Excel 导入导出

导出

@classmethod
async def export_to_excel(
    cls,
    db: AsyncSession,
    data_converter: Optional[Callable] = None
) -> BytesIO:

导入

@classmethod
async def import_from_excel(
    cls,
    db: AsyncSession,
    file_content: bytes,
    row_processor: Optional[Callable] = None
) -> Tuple[int, int]:  # (success_count, fail_count)

下载模板

@classmethod
def get_import_template(cls) -> BytesIO:

根据 excel_columns 配置自动生成带表头的模板。

唯一性检查

@classmethod
async def check_unique(
    cls,
    db: AsyncSession,
    field: str,
    value: Any,
    exclude_id: Optional[str] = None
) -> bool:

• 检查指定字段值是否唯一
• exclude_id 用于更新时排除自身

使用示例

最小实现

from app.base_service import BaseService
from .model import Customer
from .schema import CustomerCreate, CustomerUpdate

class CustomerService(BaseService[Customer, CustomerCreate, CustomerUpdate]):
    model = Customer

只需一行 model = Customer，即可获得完整的 CRUD、Excel 导入导出、数据权限能力。

带 Excel 配置

class CustomerService(BaseService[Customer, CustomerCreate, CustomerUpdate]):
    model = Customer
    
    excel_columns = {
        "name": "客户名称",
        "phone": "电话",
        "email": "邮箱",
        "level": "等级",
    }
    excel_sheet_name = "客户列表"
    
    @classmethod
    def _export_converter(cls, item):
        return {
            "name": item.name,
            "phone": item.phone or "",
            "email": item.email or "",
            "level": str(item.level),
        }
    
    @classmethod
    def _import_processor(cls, row):
        name = row.get("name")
        if not name:
            return None
        return Customer(
            name=str(name),
            phone=str(row.get("phone") or ""),
            email=str(row.get("email") or ""),
        )

自定义业务逻辑

class CustomerService(BaseService[Customer, CustomerCreate, CustomerUpdate]):
    model = Customer
    RESOURCE_DISPLAY_NAME = "客户"
    
    @classmethod
    async def get_active_customers(cls, db: AsyncSession) -> List[Customer]:
        """自定义查询：获取活跃客户"""
        result = await db.execute(
            select(Customer).where(
                Customer.is_deleted == False,
                Customer.is_active == True
            )
        )
        return list(result.scalars().all())

资源注册表

ResourceRegistry 是全局注册表，存储所有 Service 的资源类型和元数据：

class ResourceRegistry:
    _registry: Dict[str, Dict] = {}
    # 格式: {resource_type: {service, model, display_name, field_metadata, application_id}}
    
    @classmethod
    def register(cls, resource_type, service_class, display_name=None): ...
    
    @classmethod
    def get_service(cls, resource_type) -> Optional[Type]: ...
    
    @classmethod
    def get_all_resources(cls, application_id=None) -> List[Dict]: ...
    
    @classmethod
    def validate(cls, resource_type) -> bool: ...

资源类型生成规则

def auto_generate_resource_type(model_class):
    table_name = model_class.__tablename__
    # 移除前缀: core_user → user, sys_config → config
    for prefix in ['core_', 'sys_', 'app_', 'biz_']:
        if table_name.startswith(prefix):
            return table_name[len(prefix):]
    return table_name

用途

• 数据权限：通过 resource_type 绑定数据权限配置
• 字段权限：通过 field_metadata 提供字段列表
• 权限管理：前端根据注册表显示可配置的资源