fix: 素材列表默认按created_at倒序排列,新上传的在最前面
CI Build & Deploy Pipeline / Build Staging API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI Build & Deploy Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Production (pull_request) Has been skipped
CI Build & Deploy Pipeline / Production Browser E2E (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 16s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 20s
AI Code Review / AI Code Review (pull_request) Failing after 33s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 57s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 38s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 21s
Auto Merge CI PRs / Auto Merge on CI Green + Approved (pull_request) Successful in 1m13s
Auto Approve CI PRs / Auto Approve on CI Green (pull_request) Successful in 1m14s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m35s

三个列表查询接口全部缺少ORDER BY子句,PostgreSQL按物理存储顺序返回,
导致新上传的素材可能不出现在第一页。

修复:
- find_by_library: 加 ORDER BY created_at DESC
- find_by_project: 加 ORDER BY created_at DESC
- find_by_library_and_file_type: 加 ORDER BY created_at DESC

新增3个单元测试验证排序正确性。
This commit is contained in:
CI Bot
2026-07-18 13:56:56 +08:00
parent c08ef726f7
commit 29d0ee5ba1
2 changed files with 100 additions and 1 deletions
@@ -20,6 +20,7 @@ class SQLAlchemyAssetRepository:
models = (
self.session.query(AssetModel)
.filter(AssetModel.asset_library_id == library_id)
.order_by(AssetModel.created_at.desc())
.offset(skip)
.limit(limit)
.all()
@@ -33,7 +34,12 @@ class SQLAlchemyAssetRepository:
limit: int = 100,
) -> list[Asset]:
models = (
self.session.query(AssetModel).filter(AssetModel.project_id == project_id).offset(skip).limit(limit).all()
self.session.query(AssetModel)
.filter(AssetModel.project_id == project_id)
.order_by(AssetModel.created_at.desc())
.offset(skip)
.limit(limit)
.all()
)
return [self._to_domain(model) for model in models]
@@ -47,6 +53,7 @@ class SQLAlchemyAssetRepository:
models = (
self.session.query(AssetModel)
.filter(AssetModel.asset_library_id == library_id, AssetModel.file_type == file_type)
.order_by(AssetModel.created_at.desc())
.offset(skip)
.limit(limit)
.all()
+92
View File
@@ -0,0 +1,92 @@
"""Unit tests for SQLAlchemyAssetRepository - list ordering (created_at DESC)."""
import sys
from pathlib import Path
from datetime import datetime, timezone, timedelta
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.models import Base, AssetModel
from packages.domain import Asset, AssetStatus
def _repository():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)()
return SQLAlchemyAssetRepository(session), session
def _insert_asset_with_created_at(session, asset_id: str, library_id: str, project_id: str, created_at: datetime, file_type: str = "video"):
"""直接插入指定created_at的素材,绕过create()方法的时间覆盖。"""
model = AssetModel(
id=asset_id,
project_id=project_id,
asset_library_id=library_id,
name=f"{asset_id}.mp4",
file_type=file_type,
file_size=102400,
file_url=f"assets/{asset_id}.mp4",
storage_key=f"assets/{asset_id}.mp4",
status="ready",
classification_status="pending",
uploaded_by_user_id="user-1",
created_at=created_at,
updated_at=created_at,
)
session.add(model)
def test_find_by_library_ordered_by_created_at_desc():
"""find_by_library 返回结果按 created_at 倒序排列,最新的在最前面。"""
repository, session = _repository()
base_time = datetime(2026, 7, 18, 10, 0, 0, tzinfo=timezone.utc)
_insert_asset_with_created_at(session, "asset-oldest", "lib-1", "proj-1", base_time)
_insert_asset_with_created_at(session, "asset-middle", "lib-1", "proj-1", base_time + timedelta(hours=1))
_insert_asset_with_created_at(session, "asset-newest", "lib-1", "proj-1", base_time + timedelta(hours=2))
session.commit()
result = repository.find_by_library("lib-1")
assert len(result) == 3
assert [a.id for a in result] == ["asset-newest", "asset-middle", "asset-oldest"]
def test_find_by_project_ordered_by_created_at_desc():
"""find_by_project 返回结果按 created_at 倒序排列。"""
repository, session = _repository()
base_time = datetime(2026, 7, 18, 10, 0, 0, tzinfo=timezone.utc)
_insert_asset_with_created_at(session, "asset-1", "lib-1", "proj-1", base_time)
_insert_asset_with_created_at(session, "asset-2", "lib-2", "proj-1", base_time + timedelta(hours=1))
_insert_asset_with_created_at(session, "asset-3", "lib-1", "proj-1", base_time + timedelta(hours=2))
session.commit()
result = repository.find_by_project("proj-1")
assert len(result) == 3
assert [a.id for a in result] == ["asset-3", "asset-2", "asset-1"]
def test_find_by_library_and_file_type_ordered_by_created_at_desc():
"""find_by_library_and_file_type 返回结果按 created_at 倒序排列。"""
repository, session = _repository()
base_time = datetime(2026, 7, 18, 10, 0, 0, tzinfo=timezone.utc)
_insert_asset_with_created_at(session, "video-old", "lib-1", "proj-1", base_time, file_type="video")
_insert_asset_with_created_at(session, "video-new", "lib-1", "proj-1", base_time + timedelta(hours=2), file_type="video")
_insert_asset_with_created_at(session, "audio-new", "lib-1", "proj-1", base_time + timedelta(hours=1), file_type="audio")
session.commit()
# 只查video类型,验证排序和过滤同时生效
result = repository.find_by_library_and_file_type("lib-1", "video")
assert len(result) == 2
assert [a.id for a in result] == ["video-new", "video-old"]
# 只查audio类型
audio_result = repository.find_by_library_and_file_type("lib-1", "audio")
assert len(audio_result) == 1
assert audio_result[0].id == "audio-new"