别再裸 SQL 连金仓了：用 Python + FastAPI 给 KingbaseES 搭一套可上线的 API

# 方式 A：uv（推荐）
uv add fastapi uvicorn "sqlalchemy>=2.0" pydantic-settings ksycopg2

# 方式 B：pip
pip install "fastapi>=0.110" "uvicorn[standard]" "sqlalchemy>=2.0" \
            "pydantic-settings>=2.0" ksycopg2

[project]
name = "kingbase-device-api"
version = "1.0.0"
requires-python = ">=3.9"
dependencies = [
    "fastapi>=0.110",
    "uvicorn[standard]>=0.29",
    "sqlalchemy>=2.0",
    "pydantic-settings>=2.0",
    "ksycopg2>=1.0",
]

[tool.uv]
package = false

KES_DB_HOST=127.0.0.1
KES_DB_PORT=54321
KES_DB_USER=SYSTEM
KES_DB_PASSWORD=your_strong_password
KES_DB_NAME=TEST

# src/app/config.py
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """应用配置，统一从环境变量 / .env 读取。"""

    model_config = SettingsConfigDict(
        env_file=".env", env_prefix="KES_", extra="ignore"
    )

    db_host: str = "127.0.0.1"
    db_port: int = 54321
    db_user: str = "SYSTEM"
    db_password: str = "123456"
    db_name: str = "TEST"
    app_name: str = "kingbase-device-api"
    api_prefix: str = "/api/v1"

    @property
    def database_url(self) -> str:
        """拼接 SQLAlchemy 连接串。

        KingbaseES 的 SQLAlchemy dialect 名为 ``kingbase``，
        官方驱动 ``ksycopg2`` 为其默认 driver，故可省略不写。
        """
        return (
            f"kingbase+ksycopg2://{self.db_user}:{self.db_password}"
            f"@{self.db_host}:{self.db_port}/{self.db_name}"
        )


@lru_cache
def get_settings() -> Settings:
    """返回进程级单例配置，避免每次请求重复实例化。"""
    return Settings()

# src/app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.config import get_settings

settings = get_settings()

# pool_pre_ping=True：借出连接前先发轻量探活包，
# 自动剔除被服务端回收的"半开死连接"，规避偶发 operational error。
engine = create_engine(
    settings.database_url,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    future=True,
)

# sessionmaker 产出数据库会话；autoflush=False 让提交时机更可控。
SessionLocal = sessionmaker(
    bind=engine, autoflush=False, autocommit=False, future=True
)


class Base(DeclarativeBase):
    """所有 ORM 模型的声明基类。"""


def get_db():
    """FastAPI 依赖：每请求独享一个 Session，用毕必关。

    yield 写法保证即便视图抛异常，finally 仍会执行关闭，
    杜绝数据库连接泄漏。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# src/app/models/device.py
from datetime import date, datetime

from sqlalchemy import Date, DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Device(Base):
    """设备台账表 device_asset。"""

    __tablename__ = "device_asset"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    # 资产编号唯一，避免重复登记；建索引加速按编号检索
    asset_code: Mapped[str] = mapped_column(String(32), unique=True, nullable=False, index=True)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    category: Mapped[str] = mapped_column(String(64), nullable=False)
    # 状态给默认值"在用"，降低入库必填负担
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="在用")
    location: Mapped[str] = mapped_column(String(128), nullable=False)
    purchase_date: Mapped[date] = mapped_column(Date, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    # onupdate 使每次 UPDATE 自动刷新时间戳，无需业务代码手动维护
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

# src/app/schemas/device.py
from datetime import date, datetime
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

# 用 Literal 约束状态枚举，非法值直接被 422 拒绝
StatusLiteral = Literal["在用", "闲置", "维修", "报废"]


class DeviceCreate(BaseModel):
    """创建设备的入参模型。"""

    asset_code: str = Field(..., max_length=32, description="资产编号，唯一")
    name: str = Field(..., max_length=128)
    category: str = Field(..., max_length=64)
    status: StatusLiteral = "在用"
    location: str = Field(..., max_length=128)
    purchase_date: date


class DeviceUpdate(BaseModel):
    """更新模型：全字段可选，仅传需变更项（PATCH 语义）。"""

    name: str | None = Field(None, max_length=128)
    category: str | None = Field(None, max_length=64)
    status: StatusLiteral | None = None
    location: str | None = Field(None, max_length=128)
    purchase_date: date | None = None


class DeviceOut(BaseModel):
    """响应模型。from_attributes=True 使 ORM 对象可直接序列化。"""

    model_config = ConfigDict(from_attributes=True)

    id: int
    asset_code: str
    name: str
    category: str
    status: str
    location: str
    purchase_date: date
    created_at: datetime
    updated_at: datetime

# src/app/repositories/device.py
from typing import Optional, Tuple

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.device import Device
from app.schemas.device import DeviceCreate, DeviceUpdate


def create_device(db: Session, payload: DeviceCreate) -> Device:
    """插入一条设备记录并返回完整对象。"""
    device = Device(**payload.model_dump())
    db.add(device)
    db.commit()
    db.refresh(device)  # 取回自增 id 与数据库侧默认时间戳
    return device


def get_device(db: Session, device_id: int) -> Optional[Device]:
    """按主键取单条；不存在返回 None。"""
    return db.get(Device, device_id)


def list_devices(
    db: Session, *, page: int = 1, size: int = 20, status: Optional[str] = None
) -> Tuple[list[Device], int]:
    """分页列出设备，可选按状态过滤。返回 (数据列表, 总数)。"""
    stmt = select(Device)
    count_stmt = select(func.count()).select_from(Device)
    if status:
        stmt = stmt.where(Device.status == status)
        count_stmt = count_stmt.where(Device.status == status)

    total = db.scalar(count_stmt) or 0
    # 倒序让最新录入置顶；offset/limit 实现游标式分页
    items = db.scalars(
        stmt.order_by(Device.id.desc()).offset((page - 1) * size).limit(size)
    ).all()
    return items, total


def update_device(db: Session, device_id: int, payload: DeviceUpdate) -> Optional[Device]:
    """局部更新；不存在返回 None。"""
    device = db.get(Device, device_id)
    if not device:
        return None
    # exclude_unset=True：仅覆盖前端真实传参的字段，避免把未传字段误写 None
    for key, value in payload.model_dump(exclude_unset=True).items():
        setattr(device, key, value)
    db.commit()
    db.refresh(device)
    return device


def delete_device(db: Session, device_id: int) -> bool:
    """删除设备；返回是否真正命中并删除。"""
    device = db.get(Device, device_id)
    if not device:
        return False
    db.delete(device)
    db.commit()
    return True

# src/app/api/v1/devices.py（片段一：创建）
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.database import get_db
from app.repositories.device import create_device
from app.schemas.device import DeviceCreate, DeviceOut

router = APIRouter(prefix="/devices", tags=["devices"])


@router.post(
    "",
    response_model=DeviceOut,
    status_code=status.HTTP_201_CREATED,
    summary="登记一台新设备",
)
def add_device(payload: DeviceCreate, db: Session = Depends(get_db)) -> DeviceOut:
    try:
        return create_device(db, payload)
    except IntegrityError:
        # asset_code 唯一约束冲突 -> 409 冲突，而非 500 内部错误
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT, detail="资产编号已存在"
        )

# src/app/api/v1/devices.py（片段二：查询单条）
from app.repositories.device import get_device

@router.get("/{device_id}", response_model=DeviceOut, summary="按 ID 查询设备")
def read_device(device_id: int, db: Session = Depends(get_db)) -> DeviceOut:
    device = get_device(db, device_id)
    if not device:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="设备不存在")
    return device

# src/app/api/v1/devices.py（片段三：更新、删除、列表）
from typing import Optional

from fastapi import Query

from app.repositories.device import delete_device, list_devices, update_device
from app.schemas.device import DeviceOut, DeviceUpdate


@router.patch("/{device_id}", response_model=DeviceOut, summary="局部更新设备")
def edit_device(
    device_id: int, payload: DeviceUpdate, db: Session = Depends(get_db)
) -> DeviceOut:
    device = update_device(db, device_id, payload)
    if not device:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="设备不存在")
    return device


@router.delete("/{device_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除设备")
def remove_device(device_id: int, db: Session = Depends(get_db)) -> None:
    if not delete_device(db, device_id):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="设备不存在")


@router.get("", summary="分页列出设备")
def list_device_api(
    page: int = Query(1, ge=1, description="页码"),
    size: int = Query(20, ge=1, le=100, description="每页条数"),
    status_filter: Optional[str] = Query(None, alias="status", description="按状态过滤"),
    db: Session = Depends(get_db),
) -> dict:
    items, total = list_devices(db, page=page, size=size, status=status_filter)
    return {
        "total": total,
        "page": page,
        "size": size,
        "items": [DeviceOut.model_validate(d) for d in items],
    }

# collector.py —— 采集端示例（与 KES API 解耦的独立进程）
import httpx

# 亿牛云隧道代理入口（企业级，出口 IP 自动轮换）
# 格式：http://<用户名>:<密码>@<网关地址>:<端口>
YINIU_PROXY = "http://your_user:your_pass@gateway.ipyiniu.com:埠"

API_ENDPOINT = "http://api.your-intranet.com/api/v1/devices"


async def push_device(record: dict) -> int:
    """将一条采集到的设备信息经亿牛云代理回写至 KES API。

    Args:
        record: 待写入的设备字典，字段需符合 DeviceCreate 契约

    Returns:
        HTTP 状态码（201 表示创建成功）

    Raises:
        httpx.HTTPStatusError: 服务端返回 4xx/5xx 时抛出
    """
    # proxy 参数让本次请求走亿牛云隧道，出口 IP 由服务端轮换
    async with httpx.AsyncClient(proxy=YINIU_PROXY, timeout=10.0) as client:
        resp = await client.post(API_ENDPOINT, json=record)
        resp.raise_for_status()
        return resp.status_code


# 典型采集循环（伪代码）
# for item in crawl_target_site():
#     await push_device(item)   # 经亿牛云代理写入金仓