packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -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()

tests/unit/test_asset_list_ordering.py

@@ -0,0 +1,98 @@
+"""Unit tests for SQLAlchemyAssetRepository - list ordering (created_at DESC)."""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+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 AssetModel, Base
+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"