chore: remove unused Asset import from manager.py

Amp-Thread-ID: https://ampcode.com/threads/T-019c24bb-475b-7442-9ff9-8288edea3345
Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
Luke Mino-Altherr 2026-02-03 10:28:20 -08:00
parent 17ad7e393f
commit 70a600baf0
4 changed files with 9 additions and 17 deletions

View File

@ -39,12 +39,12 @@ def pick_best_live_path(states: Sequence[AssetCacheState]) -> str:
def prune_orphaned_assets(session: Session, roots: tuple[str, ...], prefixes_for_root_fn) -> int:
"""Prune cache states outside configured prefixes, then delete orphaned seed assets.
Args:
session: Database session
roots: Tuple of root types to prune
prefixes_for_root_fn: Function to get prefixes for a root type
Returns:
Number of orphaned assets deleted
"""
@ -91,7 +91,7 @@ def fast_db_consistency_pass(
- Optionally return surviving absolute paths
"""
import contextlib
prefixes = prefixes_for_root_fn(root)
if not prefixes:
return set() if collect_existing_paths else None

View File

@ -1,7 +1,7 @@
import pytest
from sqlalchemy.orm import Session
from app.assets.database.models import Asset, AssetInfo, AssetInfoMeta, AssetInfoTag, Tag
from app.assets.database.models import Asset, AssetInfo
from app.assets.database.queries import (
asset_info_exists_for_asset_id,
get_asset_info_by_id,
@ -116,7 +116,7 @@ class TestListAssetInfosPage:
def test_include_tags_filter(self, session: Session):
asset = _make_asset(session, "hash1")
info1 = _make_asset_info(session, asset, name="tagged")
info2 = _make_asset_info(session, asset, name="untagged")
_make_asset_info(session, asset, name="untagged")
ensure_tags_exist(session, ["wanted"])
add_tags_to_asset_info(session, asset_info_id=info1.id, tags=["wanted"])
session.commit()
@ -127,10 +127,10 @@ class TestListAssetInfosPage:
def test_exclude_tags_filter(self, session: Session):
asset = _make_asset(session, "hash1")
info1 = _make_asset_info(session, asset, name="keep")
info2 = _make_asset_info(session, asset, name="exclude")
_make_asset_info(session, asset, name="keep")
info_exclude = _make_asset_info(session, asset, name="exclude")
ensure_tags_exist(session, ["bad"])
add_tags_to_asset_info(session, asset_info_id=info2.id, tags=["bad"])
add_tags_to_asset_info(session, asset_info_id=info_exclude.id, tags=["bad"])
session.commit()
infos, _, total = list_asset_infos_page(session, exclude_tags=["bad"])

View File

@ -1,17 +1,11 @@
"""Tests for cache_state query functions."""
import os
import tempfile
from unittest.mock import patch
import pytest
from sqlalchemy.orm import Session
from app.assets.database.models import Asset, AssetCacheState, AssetInfo
from app.assets.database.models import Asset, AssetCacheState
from app.assets.database.queries import (
list_cache_states_by_asset_id,
pick_best_live_path,
)
from app.assets.helpers import utcnow
def _make_asset(session: Session, hash_val: str | None = None, size: int = 1024) -> Asset:

View File

@ -1,6 +1,4 @@
"""Tests for metadata filtering logic in asset_info queries."""
import pytest
from decimal import Decimal
from sqlalchemy.orm import Session
from app.assets.database.models import Asset, AssetInfo, AssetInfoMeta