fix(api): support asset library lookup for media guards
Deploy / Build Production Runtime Images (push) Successful in 8m56s
Deploy / Deploy Production (push) Successful in 1m48s
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
Deploy / Deploy Staging (push) Has been cancelled

This commit is contained in:
Xiaoxia AI
2026-06-23 09:44:03 +08:00
parent 136c40360c
commit f7e3a7a5a2
2 changed files with 49 additions and 0 deletions
@@ -8,6 +8,22 @@ class SQLAlchemyAssetLibraryRepository:
def __init__(self, session: Session):
self.session = session
def get(self, library_id: str) -> AssetLibrary | None:
model = self.session.query(AssetLibraryModel).filter(AssetLibraryModel.id == library_id).first()
if model is None:
return None
return AssetLibrary(
id=model.id,
workspace_id=model.workspace_id,
project_id=model.project_id,
name=model.name,
kind=AssetLibraryKind(model.kind),
asset_count=int(model.asset_count or 0),
total_size=int(model.total_size or 0),
created_at=model.created_at,
updated_at=model.updated_at,
)
def list_by_project(self, project_id: str, kind: AssetLibraryKind | None = None) -> list[AssetLibrary]:
query = self.session.query(AssetLibraryModel).filter(AssetLibraryModel.project_id == project_id)
if kind is not None:
@@ -0,0 +1,33 @@
from packages.adapters.sqlalchemy_impl.asset_library_repository import SQLAlchemyAssetLibraryRepository
from packages.adapters.sqlalchemy_impl.models import AssetLibraryModel
from tests.unit.test_workspace_sqlalchemy_repositories import _sqlite_session
def test_asset_library_repository_get_returns_library():
session = _sqlite_session()
session.add(
AssetLibraryModel(
id="lib-1",
workspace_id="workspace-1",
project_id="project-1",
name="素材库",
kind="video",
asset_count=2,
total_size=10,
)
)
session.commit()
library = SQLAlchemyAssetLibraryRepository(session).get("lib-1")
assert library is not None
assert library.id == "lib-1"
assert library.workspace_id == "workspace-1"
assert library.project_id == "project-1"
assert library.kind.value == "video"
def test_asset_library_repository_get_returns_none_for_missing_library():
session = _sqlite_session()
assert SQLAlchemyAssetLibraryRepository(session).get("missing") is None