4. ORM框架选择与配置：SQLAlchemy基础配置、模型定义、连接池管理

在SECS/GEM报警历史记录系统中，ORM（对象关系映射）框架是连接业务对象与关系数据库的桥梁。本章聚焦于SQLAlchemy——Python生态中最成熟、最灵活的ORM框架，详细讲解其在报警历史存储场景下的基础配置、模型定义与连接池管理策略。

4.1 为什么选择SQLAlchemy

SQLAlchemy在半导体设备数据持久化场景中具备以下核心优势：

4.2 SQLAlchemy基础配置

基础配置包括引擎创建、会话工厂设置与连接参数调优。以下为报警历史记录系统的标准配置模板：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 数据库URL格式：dialect+driver://user:password@host:port/database
DATABASE_URL = "postgresql://alarm_user:alarm_pass@192.168.1.100:5432/alarm_history"

# 创建引擎，配置连接池
engine = create_engine(
    DATABASE_URL,
    pool_size=10,           # 连接池中保持的连接数
    max_overflow=20,        # 池满后额外创建的连接数上限
    pool_pre_ping=True,     # 每次连接使用前检查是否存活
    pool_recycle=3600,      # 连接回收时间（秒），防止数据库端断开
    echo=False              # 生产环境关闭SQL日志
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 声明基类
Base = declarative_base()

4.3 报警历史记录模型定义

基于SECS/GEM标准中的报警对象（Alarm），定义核心ORM模型。模型包含报警ID、设备ID、报警文本、严重等级、发生时间、清除时间等字段。

from sqlalchemy import Column, Integer, String, DateTime, Text, Enum, Index
from sqlalchemy.sql import func
import enum

class AlarmSeverity(enum.Enum):
    CRITICAL = "CRITICAL"
    MAJOR = "MAJOR"
    MINOR = "MINOR"
    WARNING = "WARNING"

class AlarmRecord(Base):
    __tablename__ = "alarm_history"

    id = Column(Integer, primary_key=True, autoincrement=True)
    alarm_id = Column(String(32), nullable=False, comment="SECS/GEM报警ID")
    equipment_id = Column(String(64), nullable=False, comment="设备ID")
    alarm_text = Column(Text, nullable=True, comment="报警描述文本")
    severity = Column(Enum(AlarmSeverity), nullable=False, default=AlarmSeverity.MINOR)
    ceid = Column(Integer, nullable=True, comment="关联的收集事件ID")
    occurred_at = Column(DateTime(timezone=True), server_default=func.now(), comment="报警发生时间")
    cleared_at = Column(DateTime(timezone=True), nullable=True, comment="报警清除时间")
    is_active = Column(Integer, default=1, comment="报警是否仍处于激活状态")
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    # 复合索引：按设备+时间查询报警历史
    __table_args__ = (
        Index("idx_equipment_time", "equipment_id", "occurred_at"),
        Index("idx_alarm_severity", "severity", "occurred_at"),
    )

    def __repr__(self):
        return f"<AlarmRecord(alarm_id={self.alarm_id}, equipment={self.equipment_id}, severity={self.severity})>"

4.4 连接池管理策略

连接池是ORM性能的关键。报警历史记录系统面临高频写入与间歇性批量查询的混合负载，需要精细化的连接池配置。

4.4.1 连接池类型对比

4.4.2 连接池参数调优

针对报警历史记录系统的典型负载特征，推荐以下调优参数：

# 生产环境推荐配置
engine = create_engine(
    DATABASE_URL,
    pool_size=15,               # 基础连接数，匹配报警写入线程数
    max_overflow=30,            # 突发批量查询时允许扩容
    pool_pre_ping=True,         # 连接健康检查
    pool_recycle=1800,          # 30分钟回收一次，适配PG的idle_in_transaction_session_timeout
    pool_use_lifo=False,        # 默认FIFO，避免连接老化
    max_identifier_length=63,   # 兼容PostgreSQL标识符长度限制
    connect_args={
        "connect_timeout": 10,  # 连接超时10秒
        "keepalives": 1,        # 启用TCP keepalive
        "keepalives_idle": 60,  # 60秒无数据开始探测
        "keepalives_interval": 10,
        "keepalives_count": 5
    }
)

4.5 会话管理与依赖注入

在Web应用或后台服务中，正确的会话管理能防止连接泄漏。以下为FastAPI集成示例：

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

app = FastAPI()

# 依赖注入：每个请求获取独立会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/alarms/")
def create_alarm(alarm_data: dict, db: Session = Depends(get_db)):
    alarm = AlarmRecord(**alarm_data)
    db.add(alarm)
    db.commit()
    db.refresh(alarm)
    return alarm

4.6 ORM架构总览

以下SVG图展示了SQLAlchemy在报警历史记录系统中的完整架构层次：

4.7 模型迁移与版本管理

使用Alembic作为SQLAlchemy的迁移工具，管理报警历史记录表结构的版本演进：

# alembic/env.py 配置示例
from alarm_models import Base
target_metadata = Base.metadata

# 生成迁移脚本
# alembic revision --autogenerate -m "add_alarm_clear_index"

# 迁移脚本示例（自动生成）
"""add alarm clear index

Revision ID: 2a1b3c4d5e6f
Revises: 1a2b3c4d5e6f
Create Date: 2025-01-15 14:30:00
"""
def upgrade():
    op.create_index('idx_alarm_clear', 'alarm_history', ['cleared_at'])

def downgrade():
    op.drop_index('idx_alarm_clear')

4.8 性能基准与配置验证

以下为不同连接池配置下的报警写入性能对比（基于10000条报警批量插入测试）：