第 5 章：模块化开发与代码组织

Step 1：定义 Repository 接口

from abc import ABC, abstractmethod
from typing import Optional
from dataclasses import dataclass

@dataclass
class DiskInfo:
    device: str           # /dev/sda
    mount_point: str      # /mnt/data
    fs_type: str          # ext4
    total_gb: float
    used_gb: float
    uuid: str
    model: str

class IDiskRepository(ABC):
    """磁盘信息数据访问接口"""
    @abstractmethod
    async def list_all(self) -> list[DiskInfo]: ...
    @abstractmethod
    async def get_by_device(self, device: str) -> Optional[DiskInfo]: ...
    @abstractmethod
    async def get_mounted(self) -> list[DiskInfo]: ...

class IMountRepository(ABC):
    """挂载操作数据访问接口"""
    @abstractmethod
    async def mount(self, device: str, mount_point: str, fs_type: str) -> bool: ...
    @abstractmethod
    async def unmount(self, mount_point: str, force: bool = False) -> bool: ...
    @abstractmethod
    async def get_fstab_entries(self) -> list[dict]: ...

Step 2：AI 生成业务层

class DiskService:
    """磁盘管理业务层 — 不依赖 FastAPI，不依赖具体数据源"""
    def __init__(self, disk_repo: IDiskRepository, mount_repo: IMountRepository):
        self._disks = disk_repo
        self._mounts = mount_repo

    async def get_dashboard_data(self) -> dict:
        """聚合仪表盘所需的磁盘概览数据"""
        all_disks = await self._disks.list_all()
        mounted = [d for d in all_disks if d.mount_point]
        total = sum(d.total_gb for d in mounted)
        used = sum(d.used_gb for d in mounted)
        return {
            "disk_count": len(all_disks),
            "mounted_count": len(mounted),
            "total_gb": round(total, 2),
            "used_gb": round(used, 2),
            "usage_percent": round((used / total * 100) if total else 0, 1),
            "disks": [self._to_dict(d) for d in all_disks],
        }

Step 3：AI 生成接口层（FastAPI 路由）

from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(prefix="/api/disks", tags=["磁盘管理"])

@router.get("/")
async def list_disks(service: DiskService = Depends(get_disk_service)):
    """获取所有磁盘信息"""
    return await service.get_dashboard_data()

@router.post("/mount")
async def mount_disk(
    device: str, mount_point: str, fs_type: str = "auto",
    service: DiskService = Depends(get_disk_service)
):
    """挂载磁盘"""
    try:
        success = await service.mount_disk(device, mount_point, fs_type)
        if not success:
            raise HTTPException(400, "挂载失败")
        return {"status": "mounted", "device": device}
    except ValueError as e:
        raise HTTPException(400, str(e))

审查要点