475 lines
15 KiB
Markdown
475 lines
15 KiB
Markdown
# 后端开发指南
|
||
|
||
## 技术栈
|
||
|
||
| 技术 | 版本 | 用途 |
|
||
|------|------|------|
|
||
| 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 # 环境配置
|
||
```
|
||
|
||
## 开发规范
|
||
|
||
### 导入规范
|
||
|
||
```python
|
||
# 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` |
|
||
|
||
### 路由层规范
|
||
|
||
```python
|
||
# 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层规范
|
||
|
||
```python
|
||
# 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规范
|
||
|
||
```python
|
||
# 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模型规范
|
||
|
||
```python
|
||
# 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"),
|
||
)
|
||
```
|
||
|
||
### 错误处理规范
|
||
|
||
```python
|
||
# 使用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="权限不足")
|
||
```
|
||
|
||
## 配置管理
|
||
|
||
```python
|
||
# 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()
|
||
```
|
||
|
||
## 数据库迁移
|
||
|
||
```bash
|
||
# 创建迁移
|
||
alembic revision --autogenerate -m "add new table"
|
||
|
||
# 执行迁移
|
||
alembic upgrade head
|
||
|
||
# 回滚
|
||
alembic downgrade -1
|
||
|
||
# 查看历史
|
||
alembic history
|
||
```
|
||
|
||
## 开发命令
|
||
|
||
```bash
|
||
# 安装依赖
|
||
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
|