diff --git a/app/api/function_list.py b/app/api/function_list.py new file mode 100644 index 0000000..9825c49 --- /dev/null +++ b/app/api/function_list.py @@ -0,0 +1,100 @@ +from fastapi import APIRouter, Depends, status +from sqlalchemy.orm import Session +from ..core.database import get_db +from ..core.response_code import ResponseCode +from ..schemas.function_list import ( + FunctionListCreate, FunctionListRequest, FunctionGetRequest, + FunctionUpdateRequest, FunctionDeleteRequest, + FunctionApiResponse, FunctionListApiResponse +) +from ..services.function_list import FunctionListService + +router = APIRouter(prefix="/functions", tags=["铁科平台功能管理"]) + + +@router.post("/create", response_model=FunctionApiResponse, status_code=status.HTTP_201_CREATED) +def create_function(func_data: FunctionListCreate, db: Session = Depends(get_db)): + """创建功能""" + # 检查功能名称是否已存在 + existing = FunctionListService.get_function_by_name(db, func_data.function_name) + if existing: + return FunctionApiResponse( + code=ResponseCode.DATA_EXISTS, + message="功能名称已存在", + data=None + ) + + func_response = FunctionListService.create_function(db, func_data) + return FunctionApiResponse( + code=ResponseCode.SUCCESS, + message="功能创建成功", + data=func_response.model_dump() + ) + + +@router.post("/list", response_model=FunctionListApiResponse) +def get_functions(request: FunctionListRequest, db: Session = Depends(get_db)): + """获取功能列表""" + functions = FunctionListService.get_functions(db, skip=request.skip, limit=request.limit) + return FunctionListApiResponse( + code=ResponseCode.SUCCESS, + message="查询成功", + total=len(functions), + data=functions + ) + + +@router.post("/get", response_model=FunctionListApiResponse) +def get_function(request: FunctionGetRequest, db: Session = Depends(get_db)): + """根据条件查询功能""" + functions = FunctionListService.search_functions( + db, + func_id=request.id, + function_name=request.function_name + ) + if not functions: + return FunctionListApiResponse( + code=ResponseCode.DATA_NOT_FOUND, + message="功能不存在", + total=0, + data=[] + ) + return FunctionListApiResponse( + code=ResponseCode.SUCCESS, + message="查询成功", + total=len(functions), + data=functions + ) + + +@router.post("/update", response_model=FunctionApiResponse) +def update_function(request: FunctionUpdateRequest, db: Session = Depends(get_db)): + """更新功能""" + func = FunctionListService.update_function(db, request.id, request.function_data) + if not func: + return FunctionApiResponse( + code=ResponseCode.DATA_NOT_FOUND, + message="功能不存在", + data=None + ) + return FunctionApiResponse( + code=ResponseCode.SUCCESS, + message="功能更新成功", + data=func.model_dump() + ) + + +@router.post("/delete", response_model=FunctionApiResponse) +def delete_function(request: FunctionDeleteRequest, db: Session = Depends(get_db)): + """删除功能""" + if not FunctionListService.delete_function(db, request.id): + return FunctionApiResponse( + code=ResponseCode.DATA_NOT_FOUND, + message="功能不存在", + data=None + ) + return FunctionApiResponse( + code=ResponseCode.SUCCESS, + message="功能删除成功", + data=None + ) diff --git a/app/main.py b/app/main.py index 7b99340..2fe1c27 100644 --- a/app/main.py +++ b/app/main.py @@ -16,6 +16,7 @@ from .api.daily import router as daily_router from .api.section_data import router as section_data_router from .api.level_data import router as level_data_router from .api.checkpoint import router as checkpoint_router +from .api.function_list import router as function_list_router from .utils.scheduler import task_scheduler # 初始化日志系统 @@ -98,6 +99,7 @@ app.include_router(daily_router, prefix="/api") app.include_router(section_data_router, prefix="/api") app.include_router(level_data_router, prefix="/api") app.include_router(checkpoint_router, prefix="/api") +app.include_router(function_list_router, prefix="/api") # app.include_router(test_router, prefix="/api") # 根路径 diff --git a/app/models/function_list.py b/app/models/function_list.py new file mode 100644 index 0000000..db89338 --- /dev/null +++ b/app/models/function_list.py @@ -0,0 +1,18 @@ +from sqlalchemy import Column, BigInteger, String +from ..core.database import Base + + +class FunctionList(Base): + """铁科平台功能列表""" + __tablename__ = "function_list" + + id = Column(BigInteger, primary_key=True, index=True, autoincrement=True) + function_name = Column(String(255), nullable=False, comment="功能名称") + description = Column(String(1000), nullable=True, comment="描述") + + def to_dict(self): + """将模型实例转换为字典""" + return { + column.name: getattr(self, column.name) + for column in self.__table__.columns + } diff --git a/app/schemas/function_list.py b/app/schemas/function_list.py new file mode 100644 index 0000000..5f9cf21 --- /dev/null +++ b/app/schemas/function_list.py @@ -0,0 +1,74 @@ +from pydantic import BaseModel, ConfigDict +from typing import Optional, List, Any + + +class FunctionListBase(BaseModel): + """功能基础模型""" + function_name: str + description: Optional[str] = None + + +class FunctionListCreate(FunctionListBase): + """创建功能请求""" + pass + + +class FunctionListUpdate(BaseModel): + """更新功能请求""" + function_name: Optional[str] = None + description: Optional[str] = None + + +class FunctionListResponse(FunctionListBase): + """功能响应模型""" + id: int + + model_config = ConfigDict(from_attributes=True) + + @classmethod + def from_orm_function(cls, func): + """从ORM对象创建响应对象""" + return cls( + id=func.id, + function_name=func.function_name, + description=func.description + ) + + +class FunctionListRequest(BaseModel): + """列表查询请求""" + skip: Optional[int] = 0 + limit: Optional[int] = 100 + + +class FunctionGetRequest(BaseModel): + """条件查询请求""" + id: Optional[int] = None + function_name: Optional[str] = None + + +class FunctionUpdateRequest(BaseModel): + """更新请求""" + id: int + function_data: FunctionListUpdate + + +class FunctionDeleteRequest(BaseModel): + """删除请求""" + id: int + + +# 统一响应格式 +class FunctionApiResponse(BaseModel): + """功能API统一响应格式""" + code: int = 0 + message: str + data: Optional[Any] = None + + +class FunctionListApiResponse(BaseModel): + """功能列表响应格式""" + code: int = 0 + message: str + total: int + data: List[FunctionListResponse] = [] diff --git a/app/services/function_list.py b/app/services/function_list.py new file mode 100644 index 0000000..2f49a2d --- /dev/null +++ b/app/services/function_list.py @@ -0,0 +1,80 @@ +from sqlalchemy.orm import Session +from ..models.function_list import FunctionList +from ..schemas.function_list import ( + FunctionListCreate, FunctionListUpdate, FunctionListResponse +) +from typing import List, Optional +import logging + +logger = logging.getLogger(__name__) + + +class FunctionListService: + """铁科平台功能服务""" + + @staticmethod + def create_function(db: Session, func_data: FunctionListCreate) -> FunctionListResponse: + """创建功能""" + db_func = FunctionList(**func_data.model_dump()) + db.add(db_func) + db.commit() + db.refresh(db_func) + return FunctionListResponse.from_orm_function(db_func) + + @staticmethod + def get_function_by_id(db: Session, func_id: int) -> Optional[FunctionList]: + """根据ID获取功能""" + return db.query(FunctionList).filter(FunctionList.id == func_id).first() + + @staticmethod + def get_function_by_name(db: Session, name: str) -> Optional[FunctionList]: + """根据名称获取功能""" + return db.query(FunctionList).filter(FunctionList.function_name == name).first() + + @staticmethod + def get_functions(db: Session, skip: int = 0, limit: int = 100) -> List[FunctionListResponse]: + """获取功能列表""" + functions = db.query(FunctionList).offset(skip).limit(limit).all() + return [FunctionListResponse.from_orm_function(f) for f in functions] + + @staticmethod + def search_functions( + db: Session, + func_id: Optional[int] = None, + function_name: Optional[str] = None + ) -> List[FunctionListResponse]: + """根据条件搜索功能""" + query = db.query(FunctionList) + + if func_id is not None: + query = query.filter(FunctionList.id == func_id) + if function_name is not None: + query = query.filter(FunctionList.function_name.like(f"%{function_name}%")) + + functions = query.all() + return [FunctionListResponse.from_orm_function(f) for f in functions] + + @staticmethod + def update_function( + db: Session, func_id: int, func_data: FunctionListUpdate + ) -> Optional[FunctionListResponse]: + """更新功能""" + db_func = db.query(FunctionList).filter(FunctionList.id == func_id).first() + if db_func: + update_data = func_data.model_dump(exclude_unset=True) + for field, value in update_data.items(): + setattr(db_func, field, value) + db.commit() + db.refresh(db_func) + return FunctionListResponse.from_orm_function(db_func) + return None + + @staticmethod + def delete_function(db: Session, func_id: int) -> bool: + """删除功能""" + db_func = db.query(FunctionList).filter(FunctionList.id == func_id).first() + if db_func: + db.delete(db_func) + db.commit() + return True + return False