feat:[FL-158][ATP模型管理页面一致性优化]

Co-authored-by: multica-agent <github@multica.ai>

api/app/api/v1/atp_assets.py

@@ -59,6 +59,8 @@ def get_atp_asset_list(
     voltage_level: str | None = Query(default=None),
     tower_type: str | None = Query(default=None),
     scene_type: str | None = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
     _: CurrentUser = Depends(require_any_permission("atp.read", "atp.run", "atp.manage")),
     db: Session = Depends(get_db),
 ) -> AtpAssetListResponse:
@@ -69,6 +71,8 @@ def get_atp_asset_list(
         voltage_level=voltage_level,
         tower_type=tower_type,
         scene_type=scene_type,
+        limit=limit,
+        offset=offset,
     )
 
 

api/app/services/atp_asset_service.py

@@ -496,6 +496,7 @@ def serialize_asset(
         voltage_level=item.voltage_level,
         tower_type=item.tower_type,
         scene_type=item.scene_type,
+        arrester_config=item.arrester_config,
         latest_release_no=item.latest_release_no,
         active_release_no=item.active_release_no,
         active_release_id=active_release.id if active_release else None,
@@ -653,6 +654,8 @@ def list_assets(
     voltage_level: str | None,
     tower_type: str | None,
     scene_type: str | None,
+    limit: int = 50,
+    offset: int = 0,
 ) -> AtpAssetListResponse:
     stmt = select(AtpAsset)
     total_stmt = select(func.count()).select_from(AtpAsset)
@@ -681,7 +684,11 @@ def list_assets(
         stmt = stmt.where(AtpAsset.scene_type == scene_type.strip())
         total_stmt = total_stmt.where(AtpAsset.scene_type == scene_type.strip())
 
-    items = db.execute(stmt.order_by(AtpAsset.update_date.desc(), AtpAsset.code.asc())).scalars().all()
+    items = db.execute(
+        stmt.order_by(AtpAsset.update_date.desc(), AtpAsset.code.asc())
+        .limit(limit)
+        .offset(offset)
+    ).scalars().all()
     total = int(db.scalar(total_stmt) or 0)
     asset_ids = [item.id for item in items]
     release_count_map = _load_asset_release_count_map(db, asset_ids)
@@ -729,6 +736,7 @@ def create_asset(db: Session, payload: AtpAssetCreateRequest, *, actor_user_id:
         voltage_level=_normalize_optional_str(payload.voltage_level),
         tower_type=_normalize_optional_str(payload.tower_type),
         scene_type=_normalize_optional_str(payload.scene_type),
+        arrester_config=_normalize_optional_str(payload.arrester_config),
         latest_release_no=0,
         active_release_no=None,
         create_user=actor_user_id,
@@ -771,6 +779,8 @@ def update_asset(
         item.tower_type = _normalize_optional_str(update_data["tower_type"])
     if "scene_type" in update_data:
         item.scene_type = _normalize_optional_str(update_data["scene_type"])
+    if "arrester_config" in update_data:
+        item.arrester_config = _normalize_optional_str(update_data["arrester_config"])
 
     item.update_user = actor_user_id
     item.update_date = utcnow()

api/tests/test_atp_asset_service.py

@@ -177,6 +177,46 @@ def test_create_release_from_archive_requires_asset_dimensions(tmp_path) -> None
         session.close()
 
 
+def test_list_assets_paginates_after_filtering(tmp_path) -> None:
+    testing_session = _build_sessionmaker()
+    session: Session = testing_session()
+    try:
+        _seed_vfs_mount(session, root_dir=tmp_path / "vfs")
+        for index in range(4):
+            created = atp_asset_service.create_asset(
+                session,
+                AtpAssetCreateRequest(
+                    code=f"ATP-ASSET-PAGE-{index}",
+                    name=f"分页模型 {index}",
+                    voltage_level="220",
+                    tower_type="sihuita",
+                    scene_type="fanji",
+                    arrester_config="M123",
+                ),
+                actor_user_id="tester",
+            )
+            assert created is not None
+            assert created.arrester_config == "M123"
+
+        result = atp_asset_service.list_assets(
+            session,
+            keyword="分页模型",
+            status_filter=None,
+            voltage_level=None,
+            tower_type=None,
+            scene_type=None,
+            limit=2,
+            offset=1,
+        )
+
+        assert result.total == 4
+        assert len(result.items) == 2
+        assert [item.code for item in result.items] == ["ATP-ASSET-PAGE-2", "ATP-ASSET-PAGE-1"]
+        assert [item.arrester_config for item in result.items] == ["M123", "M123"]
+    finally:
+        session.close()
+
+
 def test_run_release_dry_run_materializes_directory(tmp_path, monkeypatch) -> None:
     testing_session = _build_sessionmaker()
     monkeypatch.setattr(core_database, "SessionLocal", testing_session)