设备管理系统数据库怎么设计？Python模型定义与维护记录关联

在设备管理系统的架构设计中，核心任务是将相对稳定的“物理设备资产台账”与高频动态发生的“维护保养与检修事件”进行解耦与精确绑定。设计该系统的关键在于确立以设备为主体的一对多关系，通过外键约束与事务一致性，保证设备生命周期与其全量维保履历完整追溯。在 Python 技术栈中，利用 SQLAlchemy 2.0 的强类型模型映射（Mapped）与双向关联（relationship），可以高效实现维护工单录入、状态自动流转及高效关联查询。

一、核心业务实体与数据库表结构设计

设备管理系统如果将维护信息直接作为字段塞进设备表中，会迅速遇到字段膨胀与历史覆盖问题。设备是一项资产，具有长期稳定的物理与管理属性；而维护记录是一连串时间序列事件，每次维护包含独立的发生时间、工单状态、执行人、消耗备件和花费金额。因此，必须将“设备主表”与“维护记录表”独立建表，并通过一对多（1:N）外键进行关联。

1. 表设计约束与规范要点

• 主键与唯一键：设备主表使用自增主键 id 作为内部关联标识，同时为业务使用的 asset_code（资产条码/资产编号）建立唯一索引（Unique Constraint），避免因重名导致的数据混乱。
• 外键保护策略：维护记录表中的 device_id 关联设备表的 id。强烈建议在数据库层级使用 ON DELETE RESTRICT，禁止直接物理删除已经存在维保历史的设备。已退役设备应使用状态标记或 is_deleted 逻辑删除。
• 金额精度：维保涉及的耗材与人工成本 cost 字段必须使用定点数 DECIMAL(10, 2)，严禁使用浮点数 FLOAT，防止在累计核算时产生精度漂移。
• 索引设计：维护记录表中除了 device_id 外键必须建索引外，通常还需要按时间范围调阅单台设备的维保记录，因此建议创建 (device_id, start_time) 的复合索引。

二、从需求分析到落地的实施路径

设计并落地一套可靠的设备管理数据层，需要遵循标准化步骤，先从数据边界梳理开始，再到数据库级约束建立，最后通过代码层模型与业务事务进行统一闭环。

三、Python SQLAlchemy 2.0 模型定义实现

现代 Python 开发推荐使用 SQLAlchemy 2.0 规范，通过 Mapped[...] 与 mapped_column(...) 提供完整的静态类型检查支持。以下是设备模型 Device 与维护记录模型 MaintenanceRecord 的完整实现。

from datetime import date, datetime
from decimal import Decimal
from enum import Enum
from typing import List, Optional

from sqlalchemy import (
    BigInteger,
    Date,
    DateTime,
    ForeignKey,
    Index,
    Numeric,
    String,
    Text,
    func
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship
)

# 1. 定义公共基类
class Base(DeclarativeBase):
    pass

# 2. 定义状态枚举，约束业务字段
class DeviceStatus(str, Enum):
    IN_USE = "IN_USE"             # 正常在用
    MAINTAINING = "MAINTAINING"   # 维护检修中
    SCRAPPED = "SCRAPPED"         # 报废下线

class MaintenanceStatus(str, Enum):
    PENDING = "PENDING"           # 待处理/进行中
    COMPLETED = "COMPLETED"       # 维护完成已验收
    CANCELLED = "CANCELLED"       # 已取消

# 3. 设备主表模型
class Device(Base):
    __tablename__ = "devices"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    asset_code: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="资产条形码/资产编码")
    name: Mapped[str] = mapped_column(String(128), nullable=False, comment="设备名称")
    model: Mapped[str] = mapped_column(String(64), nullable=False, default="", comment="规格型号")
    status: Mapped[str] = mapped_column(
        String(32), 
        default=DeviceStatus.IN_USE.value, 
        nullable=False, 
        comment="设备状态"
    )
    location: Mapped[str] = mapped_column(String(128), nullable=False, default="", comment="安装/存放物理位置")
    purchase_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True, comment="设备购入日期")
    is_deleted: Mapped[bool] = mapped_column(default=False, nullable=False, comment="软删除标记")
    created_at: Mapped[datetime] = mapped_column(
        DateTime, 
        server_default=func.now(), 
        nullable=False, 
        comment="录入时间"
    )

    # 建立 1:N 双向关联：指向从表维护记录
    # back_populates 确保对象间属性双向同步
    records: Mapped[List["MaintenanceRecord"]] = relationship(
        back_populates="device",
        order_by="desc(MaintenanceRecord.start_time)"
    )

    def __repr__(self) -> str:
        return f"<Device(id={self.id}, code='{self.asset_code}', status='{self.status}')>"

# 4. 维护记录表模型
class MaintenanceRecord(Base):
    __tablename__ = "maintenance_records"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    
    # 外键关联到 devices.id，并在数据库层设置禁止物理删除父项
    device_id: Mapped[int] = mapped_column(
        BigInteger, 
        ForeignKey("devices.id", ondelete="RESTRICT"), 
        nullable=False,
        index=True,
        comment="关联的设备主键"
    )
    
    record_type: Mapped[str] = mapped_column(String(32), nullable=False, comment="维护类型: 保养/报修/巡检")
    description: Mapped[str] = mapped_column(Text, nullable=False, comment="维护描述或故障内容")
    cost: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False, default=Decimal("0.00"), comment="维护费用")
    operator: Mapped[str] = mapped_column(String(64), nullable=False, default="", comment="责任人/工程师")
    status: Mapped[str] = mapped_column(
        String(32), 
        default=MaintenanceStatus.PENDING.value, 
        nullable=False, 
        comment="工单状态"
    )
    start_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.now, comment="开始维护时间")
    finish_time: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, comment="完成验收时间")

    # 反向关联回设备主表
    device: Mapped["Device"] = relationship(back_populates="records")

    # 创建联合索引，加速按设备及时间段回溯
    __table_args__ = (
        Index("idx_device_time", "device_id", "start_time"),
    )

    def __repr__(self) -> str:
        return f"<MaintenanceRecord(id={self.id}, device_id={self.device_id}, status='{self.status}')>"

四、维护生命周期与设备状态联动机制

维护记录不能孤立存在，必须与主设备的状态机紧密联动。当设备报修或录入保养工单时，设备应自动标记为“维护中”，以防产线重复调度；当维保工程师确认完成并由管理员验收后，工单状态更新为“已完成”，主设备状态同步恢复为“在用”。

1. 业务操作事务实现（新增工单与完成验收）

在 Python 代码中，通过 Session 事务确保维护记录写入与设备状态更新具有原子性，任何一步异常均回滚，避免出现“工单已写入但设备仍显示在用”的状态脱节。

from datetime import datetime
from decimal import Decimal
from sqlalchemy.orm import Session
from sqlalchemy import select

def create_maintenance_order(
    session: Session, 
    device_id: int, 
    record_type: str, 
    description: str, 
    operator: str
) -> MaintenanceRecord:
    """
    新建维护保养工单，并将设备状态变更为 MAINTAINING
    """
    with session.begin():
        # 1. 查找目标设备
        stmt = select(Device).where(Device.id == device_id).with_for_update()
        device = session.scalar(stmt)
        if not device:
            raise ValueError(f"设备 ID {device_id} 不存在")
            
        if device.status == DeviceStatus.MAINTAINING.value:
            raise RuntimeError(f"设备 {device.asset_code} 正在维保中，不能重复创建工单")
        if device.status == DeviceStatus.SCRAPPED.value:
            raise RuntimeError(f"设备 {device.asset_code} 已报废，不可录入维护工单")

        # 2. 创建维保从表记录
        record = MaintenanceRecord(
            device_id=device.id,
            record_type=record_type,
            description=description,
            operator=operator,
            status=MaintenanceStatus.PENDING.value,
            start_time=datetime.now()
        )
        session.add(record)

        # 3. 联动修改主设备状态
        device.status = DeviceStatus.MAINTAINING.value

    return record

def complete_maintenance_order(
    session: Session, 
    record_id: int, 
    cost: Decimal
) -> None:
    """
    完成维护工单，登记费用并恢复设备为正常在用
    """
    with session.begin():
        stmt = select(MaintenanceRecord).where(MaintenanceRecord.id == record_id).with_for_update()
        record = session.scalar(stmt)
        if not record:
            raise ValueError(f"维护工单 ID {record_id} 不存在")
            
        if record.status == MaintenanceStatus.COMPLETED.value:
            return  # 已完成无需重复处理

        # 更新工单
        record.status = MaintenanceStatus.COMPLETED.value
        record.finish_time = datetime.now()
        record.cost = cost

        # 联动将设备状态恢复为在用
        record.device.status = DeviceStatus.IN_USE.value

2. 关联查询与性能优化（避免 N+1 陷阱）

在展示设备列表及其最新维保记录时，如果不加注意，直接在模板或循环中访问 device.records，会导致针对每台设备各执行一次 SQL 查询，产生典型的 N+1 性能瓶颈。在 SQLAlchemy 中应使用 selectinload 或 joinedload 进行预加载。

from sqlalchemy.orm import selectinload

def get_devices_with_maintenance_history(session: Session, limit: int = 50):
    """
    预加载设备及其所有的维护历史记录，仅产生 2 次高效查询
    """
    stmt = (
        select(Device)
        .where(Device.is_deleted == False)
        .options(selectinload(Device.records))
        .order_by(Device.id.desc())
        .limit(limit)
    )
    devices = session.scalars(stmt).all()
    
    # 此时访问 device.records 为纯内存访问，零额外 SQL
    for d in devices:
        print(f"设备 {d.asset_code} ({d.name}) | 历史维保次数: {len(d.records)}")
        for rec in d.records[:3]:  # 查看最近 3 次
            print(f"  - [{rec.start_time.strftime('%Y-%m-%d')}] {rec.record_type}: {rec.description} (￥{rec.cost})")

    return devices

五、设计总结与落地核查清单

为了保障系统在长期稳定运行中既能保持高并发查询性能，又能严格保证历史数据的安全完整性，以下三层设计检查清单涵盖了从表结构、模型定义到业务事务的核心关键项。