hospital_performance/docs/backend.md

后端开发指南

技术栈

技术	版本	用途
FastAPI	0.115+	Web框架
SQLAlchemy	2.0+	ORM
Pydantic	2.10+	数据验证
Alembic	1.14+	数据库迁移
python-jose	3.3+	JWT处理
passlib	1.7+	密码哈希
asyncpg	0.30+	PostgreSQL异步驱动
uvicorn	0.32+	ASGI服务器

项目结构

backend/
├── app/
│   ├── api/v1/                 # API路由
│   │   ├── __init__.py
│   │   ├── auth.py             # 认证接口
│   │   ├── staff.py            # 员工接口
│   │   ├── departments.py      # 科室接口
│   │   ├── indicators.py       # 指标接口
│   │   ├── assessments.py      # 考核接口
│   │   ├── salary.py           # 工资接口
│   │   └── stats.py            # 统计接口
│   ├── core/                   # 核心模块
│   │   ├── __init__.py
│   │   ├── config.py           # 配置管理
│   │   ├── database.py         # 数据库连接
│   │   ├── security.py         # 安全工具
│   │   └── init_db.py          # 数据库初始化
│   ├── models/                 # ORM模型
│   │   ├── __init__.py
│   │   └── models.py           # 数据表定义
│   ├── schemas/                # Pydantic模式
│   │   ├── __init__.py
│   │   └── schemas.py          # 请求/响应模式
│   ├── services/               # 业务逻辑
│   │   ├── __init__.py
│   │   ├── staff_service.py
│   │   ├── department_service.py
│   │   ├── indicator_service.py
│   │   ├── assessment_service.py
│   │   ├── salary_service.py
│   │   └── stats_service.py
│   ├── utils/                  # 工具函数
│   ├── __init__.py
│   └── main.py                 # 应用入口
├── alembic/                    # 数据库迁移
│   ├── versions/
│   └── env.py
├── requirements.txt
└── .env                        # 环境配置

开发规范

导入规范

# 1. 标准库
from datetime import datetime
from typing import Optional, List

# 2. 第三方库
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from pydantic import BaseModel, Field, ConfigDict

# 3. 本地模块
from app.core.database import get_db
from app.schemas.schemas import StaffCreate, StaffResponse
from app.services.staff_service import StaffService

命名规范

类型	规范	示例
文件	snake_case	staff_service.py
类	PascalCase	StaffService, StaffCreate
函数/方法	snake_case	get_staff_list()
变量	snake_case	staff_list, department_id
常量	UPPER_SNAKE_CASE	DATABASE_URL
枚举值	UPPER_CASE	ACTIVE, SUBMITTED

路由层规范

# api/v1/staff.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.schemas.schemas import StaffCreate, StaffUpdate, StaffResponse
from app.services.staff_service import StaffService

router = APIRouter(prefix="/staff", tags=["员工管理"])

@router.get("", response_model=dict)
async def get_staff_list(
    name: Optional[str] = Query(None, description="姓名"),
    department_id: Optional[int] = Query(None, description="科室ID"),
    status: Optional[str] = Query(None, description="状态"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db)
):
    """获取员工列表"""
    staff_list, total = await StaffService.get_list(
        db, name, department_id, status, page, page_size
    )
    return {
        "code": 200,
        "message": "success",
        "data": staff_list,
        "total": total,
        "page": page,
        "page_size": page_size
    }

@router.post("", response_model=dict)
async def create_staff(
    data: StaffCreate,
    db: AsyncSession = Depends(get_db)
):
    """创建员工"""
    # 检查工号是否已存在
    existing = await StaffService.get_by_employee_id(db, data.employee_id)
    if existing:
        raise HTTPException(status_code=400, detail="工号已存在")
    
    staff = await StaffService.create(db, data)
    return {"code": 200, "message": "创建成功", "data": staff}

@router.put("/{staff_id}", response_model=dict)
async def update_staff(
    staff_id: int,
    data: StaffUpdate,
    db: AsyncSession = Depends(get_db)
):
    """更新员工"""
    staff = await StaffService.get_by_id(db, staff_id)
    if not staff:
        raise HTTPException(status_code=404, detail="员工不存在")
    
    updated = await StaffService.update(db, staff, data)
    return {"code": 200, "message": "更新成功", "data": updated}

@router.delete("/{staff_id}")
async def delete_staff(
    staff_id: int,
    db: AsyncSession = Depends(get_db)
):
    """删除员工"""
    staff = await StaffService.get_by_id(db, staff_id)
    if not staff:
        raise HTTPException(status_code=404, detail="员工不存在")
    
    await StaffService.delete(db, staff)
    return {"code": 200, "message": "删除成功"}

Service层规范

# services/staff_service.py
from typing import Optional, List, Tuple
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, or_

from app.models.models import Staff, Department
from app.schemas.schemas import StaffCreate, StaffUpdate


class StaffService:
    """员工服务"""
    
    @staticmethod
    async def get_list(
        db: AsyncSession,
        name: Optional[str] = None,
        department_id: Optional[int] = None,
        status: Optional[str] = None,
        page: int = 1,
        page_size: int = 20
    ) -> Tuple[List[Staff], int]:
        """获取员工列表"""
        query = select(Staff).join(Department)
        
        # 条件过滤
        if name:
            query = query.where(Staff.name.ilike(f"%{name}%"))
        if department_id:
            query = query.where(Staff.department_id == department_id)
        if status:
            query = query.where(Staff.status == status)
        
        # 统计总数
        count_query = select(func.count()).select_from(query.subquery())
        total = await db.scalar(count_query)
        
        # 分页
        query = query.offset((page - 1) * page_size).limit(page_size)
        result = await db.execute(query)
        staff_list = result.scalars().all()
        
        return staff_list, total
    
    @staticmethod
    async def get_by_id(db: AsyncSession, staff_id: int) -> Optional[Staff]:
        """根据ID获取员工"""
        result = await db.execute(
            select(Staff).where(Staff.id == staff_id)
        )
        return result.scalar_one_or_none()
    
    @staticmethod
    async def get_by_employee_id(db: AsyncSession, employee_id: str) -> Optional[Staff]:
        """根据工号获取员工"""
        result = await db.execute(
            select(Staff).where(Staff.employee_id == employee_id)
        )
        return result.scalar_one_or_none()
    
    @staticmethod
    async def create(db: AsyncSession, data: StaffCreate) -> Staff:
        """创建员工"""
        staff = Staff(**data.model_dump())
        db.add(staff)
        await db.commit()
        await db.refresh(staff)
        return staff
    
    @staticmethod
    async def update(db: AsyncSession, staff: Staff, data: StaffUpdate) -> Staff:
        """更新员工"""
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(staff, key, value)
        await db.commit()
        await db.refresh(staff)
        return staff
    
    @staticmethod
    async def delete(db: AsyncSession, staff: Staff) -> None:
        """删除员工"""
        await db.delete(staff)
        await db.commit()

Pydantic Schema规范

# schemas/schemas.py
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, ConfigDict
from enum import Enum


class StaffStatus(str, Enum):
    ACTIVE = "active"
    LEAVE = "leave"
    RESIGNED = "resigned"
    RETIRED = "retired"


# 基础模式
class StaffBase(BaseModel):
    """员工基础模式"""
    employee_id: str = Field(..., max_length=20, description="工号")
    name: str = Field(..., max_length=50, description="姓名")
    department_id: int = Field(..., description="所属科室ID")
    position: str = Field(..., max_length=50, description="职位")
    title: Optional[str] = Field(None, max_length=50, description="职称")
    phone: Optional[str] = Field(None, max_length=20, description="电话")
    email: Optional[str] = Field(None, max_length=100, description="邮箱")
    base_salary: float = Field(default=0, ge=0, description="基本工资")
    performance_ratio: float = Field(default=1.0, ge=0, le=5, description="绩效系数")


# 创建模式
class StaffCreate(StaffBase):
    """创建员工"""
    status: StaffStatus = Field(default=StaffStatus.ACTIVE)
    hire_date: Optional[datetime] = None


# 更新模式
class StaffUpdate(BaseModel):
    """更新员工"""
    name: Optional[str] = Field(None, max_length=50)
    department_id: Optional[int] = None
    position: Optional[str] = Field(None, max_length=50)
    title: Optional[str] = Field(None, max_length=50)
    phone: Optional[str] = Field(None, max_length=20)
    email: Optional[str] = Field(None, max_length=100)
    base_salary: Optional[float] = Field(None, ge=0)
    performance_ratio: Optional[float] = Field(None, ge=0, le=5)
    status: Optional[StaffStatus] = None


# 响应模式
class StaffResponse(StaffBase):
    """员工响应"""
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    status: StaffStatus
    hire_date: Optional[datetime]
    created_at: datetime
    updated_at: datetime
    department_name: Optional[str] = None

ORM模型规范

# models/models.py
from datetime import datetime
from typing import Optional, List
from sqlalchemy import String, Integer, Numeric, Boolean, DateTime, ForeignKey, Enum as SQLEnum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from enum import Enum

from app.core.database import Base


class StaffStatus(Enum):
    ACTIVE = "active"
    LEAVE = "leave"
    RESIGNED = "resigned"
    RETIRED = "retired"


class Staff(Base):
    """员工信息表"""
    __tablename__ = "staff"
    
    # 使用Mapped类型注解
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    employee_id: Mapped[str] = mapped_column(String(20), unique=True, nullable=False, comment="工号")
    name: Mapped[str] = mapped_column(String(50), nullable=False, comment="姓名")
    department_id: Mapped[int] = mapped_column(ForeignKey("departments.id"), nullable=False, comment="所属科室")
    position: Mapped[str] = mapped_column(String(50), nullable=False, comment="职位")
    title: Mapped[Optional[str]] = mapped_column(String(50), nullable=True, comment="职称")
    phone: Mapped[Optional[str]] = mapped_column(String(20), nullable=True, comment="联系电话")
    email: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, comment="邮箱")
    base_salary: Mapped[float] = mapped_column(Numeric(10, 2), default=0, comment="基本工资")
    performance_ratio: Mapped[float] = mapped_column(Numeric(5, 2), default=1.0, comment="绩效系数")
    status: Mapped[StaffStatus] = mapped_column(SQLEnum(StaffStatus), default=StaffStatus.ACTIVE, comment="状态")
    hire_date: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, comment="入职日期")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, comment="创建时间")
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, comment="更新时间")
    
    # 关系
    department: Mapped["Department"] = relationship("Department", back_populates="staff")
    
    # 索引
    __table_args__ = (
        Index("idx_staff_dept", "department_id"),
        Index("idx_staff_status", "status"),
    )

错误处理规范

# 使用HTTPException返回错误
from fastapi import HTTPException

# 404错误
if not staff:
    raise HTTPException(status_code=404, detail="员工不存在")

# 400错误（业务规则）
if existing:
    raise HTTPException(status_code=400, detail="工号已存在")

# 403错误（权限）
if current_user.role != "admin":
    raise HTTPException(status_code=403, detail="权限不足")

配置管理

# core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """系统配置"""
    
    # 应用配置
    APP_NAME: str = "医院绩效考核管理系统"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = True
    API_PREFIX: str = "/api/v1"
    
    # 数据库配置
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/hospital_performance"
    DATABASE_POOL_SIZE: int = 20
    DATABASE_MAX_OVERFLOW: int = 10
    
    # JWT配置
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 480  # 8小时
    
    # 跨域配置
    CORS_ORIGINS: list[str] = ["http://localhost:3000", "http://localhost:5173"]
    
    # 分页配置
    DEFAULT_PAGE_SIZE: int = 20
    MAX_PAGE_SIZE: int = 100
    
    class Config:
        env_file = ".env"
        case_sensitive = True


@lru_cache()
def get_settings() -> Settings:
    return Settings()


settings = get_settings()

数据库迁移

# 创建迁移
alembic revision --autogenerate -m "add new table"

# 执行迁移
alembic upgrade head

# 回滚
alembic downgrade -1

# 查看历史
alembic history

开发命令

# 安装依赖
pip install -r requirements.txt

# 启动开发服务器
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 运行测试
pytest

# 运行特定测试
pytest tests/test_staff.py -v

API文档

启动服务后访问：

• Swagger UI: http://localhost:8000/api/v1/docs
• ReDoc: http://localhost:8000/api/v1/redoc