From 322d152e397f11b98597e5a57c5c4be89e3fef6d Mon Sep 17 00:00:00 2001 From: Simon Pinfold Date: Wed, 26 Aug 2026 03:51:23 -0700 Subject: [PATCH] refactor(assets): delete dead pre-split code; relocate SeedAssetSpec (D18) Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- app/assets/database/models.py | 5 - app/assets/database/queries/__init__.py | 2 +- app/assets/database/queries/asset.py | 152 --- .../database/queries/asset_reference.py | 1067 ----------------- app/assets/database/queries/common.py | 16 - app/assets/database/queries/tags.py | 171 +-- app/assets/scanner.py | 36 +- app/assets/scanner_admission.py | 1 - app/assets/seeder.py | 1 - app/assets/services/__init__.py | 4 - app/assets/services/asset_management.py | 178 --- app/assets/services/bulk_ingest.py | 274 ----- app/assets/services/ingest.py | 404 +------ app/assets/services/schemas.py | 28 - .../services/test_recovery_gate.py | 3 +- .../services/test_scanner_seed_resilience.py | 4 +- 16 files changed, 23 insertions(+), 2323 deletions(-) delete mode 100644 app/assets/database/queries/asset.py delete mode 100644 app/assets/database/queries/asset_reference.py delete mode 100644 app/assets/services/bulk_ingest.py diff --git a/app/assets/database/models.py b/app/assets/database/models.py index 239b93d36..85ccd9bed 100644 --- a/app/assets/database/models.py +++ b/app/assets/database/models.py @@ -177,8 +177,3 @@ class AssetSystemState(Base): key: Mapped[str] = mapped_column(String(256), primary_key=True) value: Mapped[str] = mapped_column(Text, nullable=False) - - -AssetReference = None -AssetReferenceMeta = None -AssetReferenceTag = None diff --git a/app/assets/database/queries/__init__.py b/app/assets/database/queries/__init__.py index 4d201868b..990cbccd1 100644 --- a/app/assets/database/queries/__init__.py +++ b/app/assets/database/queries/__init__.py @@ -28,7 +28,7 @@ __all__ = [ def __getattr__(name: str): from importlib import import_module - for module_name in ("asset", "asset_reference", "tags"): + for module_name in ("tags",): module = import_module(f"app.assets.database.queries.{module_name}") candidate = getattr(module, name, None) if candidate is not None: diff --git a/app/assets/database/queries/asset.py b/app/assets/database/queries/asset.py deleted file mode 100644 index cc7168431..000000000 --- a/app/assets/database/queries/asset.py +++ /dev/null @@ -1,152 +0,0 @@ -import sqlalchemy as sa -from sqlalchemy import select -from sqlalchemy.dialects import sqlite -from sqlalchemy.orm import Session - -from app.assets.database.models import Asset, AssetReference -from app.assets.database.queries.common import MAX_BIND_PARAMS, calculate_rows_per_statement, iter_chunks - - -def asset_exists_by_hash( - session: Session, - asset_hash: str, -) -> bool: - """ - Check if an asset with a given hash exists in database. - """ - row = ( - session.execute( - select(sa.literal(True)) - .select_from(Asset) - .where(Asset.hash == asset_hash) - .limit(1) - ) - ).first() - return row is not None - - -def get_asset_by_hash( - session: Session, - asset_hash: str, -) -> Asset | None: - return ( - (session.execute(select(Asset).where(Asset.hash == asset_hash).limit(1))) - .scalars() - .first() - ) - - -def upsert_asset( - session: Session, - asset_hash: str, - size_bytes: int, - mime_type: str | None = None, -) -> tuple[Asset, bool, bool]: - """Upsert an Asset by hash. Returns (asset, created, updated).""" - vals = {"hash": asset_hash, "size_bytes": int(size_bytes)} - if mime_type: - vals["mime_type"] = mime_type - - ins = ( - sqlite.insert(Asset) - .values(**vals) - .on_conflict_do_nothing(index_elements=[Asset.hash]) - ) - res = session.execute(ins) - created = int(res.rowcount or 0) > 0 - - asset = ( - session.execute(select(Asset).where(Asset.hash == asset_hash).limit(1)) - .scalars() - .first() - ) - if not asset: - raise RuntimeError("Asset row not found after upsert.") - - updated = False - if not created: - changed = False - if asset.size_bytes != int(size_bytes) and int(size_bytes) > 0: - asset.size_bytes = int(size_bytes) - changed = True - if mime_type and not asset.mime_type: - asset.mime_type = mime_type - changed = True - if changed: - updated = True - - return asset, created, updated - - -def create_stub_asset( - session: Session, - size_bytes: int, - mime_type: str | None = None, -) -> Asset: - """Create a new asset with no hash (stub for later enrichment).""" - asset = Asset(size_bytes=size_bytes, mime_type=mime_type, hash=None) - session.add(asset) - session.flush() - return asset - - -def bulk_insert_assets( - session: Session, - rows: list[dict], -) -> None: - """Bulk insert Asset rows with ON CONFLICT DO NOTHING on hash.""" - if not rows: - return - ins = sqlite.insert(Asset).on_conflict_do_nothing(index_elements=[Asset.hash]) - for chunk in iter_chunks(rows, calculate_rows_per_statement(5)): - session.execute(ins, chunk) - - -def get_existing_asset_ids( - session: Session, - asset_ids: list[str], -) -> set[str]: - """Return the subset of asset_ids that exist in the database.""" - if not asset_ids: - return set() - found: set[str] = set() - for chunk in iter_chunks(asset_ids, MAX_BIND_PARAMS): - rows = session.execute( - select(Asset.id).where(Asset.id.in_(chunk)) - ).fetchall() - found.update(row[0] for row in rows) - return found - - -def update_asset_hash_and_mime( - session: Session, - asset_id: str, - asset_hash: str | None = None, - mime_type: str | None = None, -) -> bool: - """Update asset hash and/or mime_type. Returns True if asset was found.""" - asset = session.get(Asset, asset_id) - if not asset: - return False - if asset_hash is not None: - asset.hash = asset_hash - if mime_type is not None and not asset.mime_type: - asset.mime_type = mime_type - return True - - -def reassign_asset_references( - session: Session, - from_asset_id: str, - to_asset_id: str, - reference_id: str, -) -> None: - """Reassign a reference from one asset to another. - - Used when merging a stub asset into an existing asset with the same hash. - """ - ref = session.get(AssetReference, reference_id) - if ref and ref.asset_id == from_asset_id: - ref.asset_id = to_asset_id - - session.flush() diff --git a/app/assets/database/queries/asset_reference.py b/app/assets/database/queries/asset_reference.py deleted file mode 100644 index e1d708364..000000000 --- a/app/assets/database/queries/asset_reference.py +++ /dev/null @@ -1,1067 +0,0 @@ -"""Query functions for the unified AssetReference table. - -This module replaces the separate asset_info.py and cache_state.py query modules, -providing a unified interface for the merged asset_references table. -""" - -from __future__ import annotations - -from collections import defaultdict -from datetime import datetime -from decimal import Decimal -from typing import NamedTuple, Sequence - -import sqlalchemy as sa -from sqlalchemy import delete, select -from sqlalchemy.dialects import sqlite -from sqlalchemy.exc import IntegrityError -from sqlalchemy.orm import Session, noload - -from app.assets.database.models import ( - Asset, - AssetReference, - AssetReferenceMeta, - AssetReferenceTag, - Tag, -) -from app.assets.database.queries.common import ( - MAX_BIND_PARAMS, - apply_metadata_filter, - apply_tag_filters, - build_prefix_like_conditions, - build_visibility_clause, - calculate_rows_per_statement, - iter_chunks, -) -from app.assets.helpers import escape_sql_like_string, get_utc_now - - -def _check_is_scalar(v): - if v is None: - return True - if isinstance(v, bool): - return True - if isinstance(v, (int, float, Decimal, str)): - return True - return False - - -def _scalar_to_row(key: str, ordinal: int, value) -> dict: - """Convert a scalar value to a typed projection row.""" - if isinstance(value, bool): - return {"key": key, "ordinal": ordinal, "val_bool": bool(value)} - if isinstance(value, (int, float, Decimal)): - num = value if isinstance(value, Decimal) else Decimal(str(value)) - return {"key": key, "ordinal": ordinal, "val_num": num} - if isinstance(value, str): - return {"key": key, "ordinal": ordinal, "val_str": value} - return {"key": key, "ordinal": ordinal, "val_json": value} - - -def convert_metadata_to_rows(key: str, value) -> list[dict]: - """Turn a metadata key/value into typed projection rows.""" - if value is None: - return [] - - if _check_is_scalar(value): - return [_scalar_to_row(key, 0, value)] - - if isinstance(value, list): - if all(_check_is_scalar(x) for x in value): - return [_scalar_to_row(key, i, x) for i, x in enumerate(value) if x is not None] - return [{"key": key, "ordinal": i, "val_json": x} for i, x in enumerate(value) if x is not None] - - return [{"key": key, "ordinal": 0, "val_json": value}] - - - - -def get_reference_by_id( - session: Session, - reference_id: str, -) -> AssetReference | None: - return session.get(AssetReference, reference_id) - - -def get_reference_with_owner_check( - session: Session, - reference_id: str, - tenant_id: str, -) -> AssetReference: - """Fetch a reference and verify ownership. - - Raises: - ValueError: if reference not found - PermissionError: if tenant_id doesn't match - """ - ref = get_reference_by_id(session, reference_id=reference_id) - if not ref: - raise ValueError(f"AssetReference {reference_id} not found") - if ref.tenant_id and ref.tenant_id != tenant_id: - raise PermissionError("not owner") - return ref - - -def get_reference_by_file_path( - session: Session, - file_path: str, -) -> AssetReference | None: - """Get a reference by its file path.""" - return ( - session.execute( - select(AssetReference).where(AssetReference.file_path == file_path).limit(1) - ) - .scalars() - .first() - ) - - -def count_active_siblings( - session: Session, - asset_id: str, - exclude_reference_id: str, -) -> int: - """Count active references to an asset, excluding one reference.""" - return ( - session.query(AssetReference) - .filter( - AssetReference.asset_id == asset_id, - AssetReference.id != exclude_reference_id, - ) - .count() - ) - - -def reference_exists_for_asset_id( - session: Session, - asset_id: str, -) -> bool: - q = ( - select(sa.literal(True)) - .select_from(AssetReference) - .where(AssetReference.asset_id == asset_id) - .limit(1) - ) - return session.execute(q).first() is not None - - -def reference_exists( - session: Session, - reference_id: str, -) -> bool: - """Return True if a reference with the given ID exists.""" - q = ( - select(sa.literal(True)) - .select_from(AssetReference) - .where(AssetReference.id == reference_id) - .limit(1) - ) - return session.execute(q).first() is not None - - -def insert_reference( - session: Session, - asset_id: str, - name: str, - tenant_id: str = "", - file_path: str | None = None, - mtime_ns: int | None = None, - preview_id: str | None = None, -) -> AssetReference | None: - """Insert a new AssetReference. Returns None if unique constraint violated.""" - now = get_utc_now() - try: - with session.begin_nested(): - ref = AssetReference( - asset_id=asset_id, - name=name, - tenant_id=tenant_id, - file_path=file_path, - mtime_ns=mtime_ns, - preview_id=preview_id, - created_at=now, - updated_at=now, - last_access_time=now, - ) - session.add(ref) - session.flush() - return ref - except IntegrityError: - return None - - -def get_or_create_reference( - session: Session, - asset_id: str, - name: str, - tenant_id: str = "", - file_path: str | None = None, - mtime_ns: int | None = None, - preview_id: str | None = None, -) -> tuple[AssetReference, bool]: - """Get existing or create new AssetReference. - - For filesystem references (file_path is set), uniqueness is by file_path. - For API references (file_path is None), we look for matching - asset_id + tenant_id + name. - - Returns (reference, created). - """ - ref = insert_reference( - session, - asset_id=asset_id, - name=name, - tenant_id=tenant_id, - file_path=file_path, - mtime_ns=mtime_ns, - preview_id=preview_id, - ) - if ref: - return ref, True - - # Find existing - priority to file_path match, then name match - if file_path: - existing = get_reference_by_file_path(session, file_path) - else: - existing = ( - session.execute( - select(AssetReference) - .where( - AssetReference.asset_id == asset_id, - AssetReference.name == name, - AssetReference.tenant_id == tenant_id, - AssetReference.file_path.is_(None), - ) - .limit(1) - ) - .unique() - .scalar_one_or_none() - ) - if not existing: - raise RuntimeError("Failed to find AssetReference after insert conflict.") - return existing, False - - -def update_reference_timestamps( - session: Session, - reference: AssetReference, - preview_id: str | None = None, -) -> None: - """Update timestamps and optionally preview_id on existing AssetReference.""" - now = get_utc_now() - if preview_id and reference.preview_id != preview_id: - reference.preview_id = preview_id - reference.updated_at = now - - -def list_references_page( - session: Session, - tenant_id: str = "", - limit: int = 100, - offset: int = 0, - name_contains: str | None = None, - include_tags: Sequence[str] | None = None, - exclude_tags: Sequence[str] | None = None, - metadata_filter: dict | None = None, - sort: str | None = None, - order: str | None = None, - after_cursor_value: object | None = None, - after_cursor_id: str | None = None, - # Appended last so pre-existing positional callers keep binding correctly. - any_tags: Sequence[str] | None = None, -) -> tuple[list[AssetReference], dict[str, list[str]], int]: - """List references with pagination, filtering, and sorting. - - When ``after_cursor_value``/``after_cursor_id`` are supplied the query uses - keyset pagination — ``offset`` is ignored and a WHERE clause selects rows - strictly after the given ``(sort_col, id)`` position in the active sort - direction. The cursor value must already be typed for the column - (datetime for time sorts, int for size, str for name); the caller decodes - the opaque cursor string and resolves to the typed value. - - Returns (references, tag_map, total_count). - """ - base = ( - select(AssetReference) - .join(Asset, Asset.id == AssetReference.asset_id) - .where(build_visibility_clause(tenant_id)) - .where(AssetReference.is_missing == False) # noqa: E712 - .options(noload(AssetReference.tags)) - ) - - if name_contains: - escaped, esc = escape_sql_like_string(name_contains) - base = base.where(AssetReference.name.ilike(f"%{escaped}%", escape=esc)) - - base = apply_tag_filters(base, include_tags, exclude_tags, any_tags) - base = apply_metadata_filter(base, metadata_filter) - - sort = (sort or "created_at").lower() - order = (order or "desc").lower() - sort_map = { - "name": AssetReference.name, - "created_at": AssetReference.created_at, - "updated_at": AssetReference.updated_at, - "last_access_time": AssetReference.last_access_time, - "size": Asset.size_bytes, - } - sort_col = sort_map.get(sort, AssetReference.created_at) - descending = order == "desc" - - # Keyset WHERE: (sort_col, id) strictly less-than / greater-than the cursor. - # Equivalent to: sort_col v OR (sort_col = v AND id cursor_id). - if after_cursor_value is not None and after_cursor_id is not None: - if descending: - keyset = sa.or_( - sort_col < after_cursor_value, - sa.and_(sort_col == after_cursor_value, AssetReference.id < after_cursor_id), - ) - else: - keyset = sa.or_( - sort_col > after_cursor_value, - sa.and_(sort_col == after_cursor_value, AssetReference.id > after_cursor_id), - ) - base = base.where(keyset) - - # Secondary ORDER BY id (matching the primary direction) gives the keyset - # comparison a deterministic tiebreaker on duplicate sort_col values. - id_exp = AssetReference.id.desc() if descending else AssetReference.id.asc() - sort_exp = sort_col.desc() if descending else sort_col.asc() - - base = base.order_by(sort_exp, id_exp).limit(limit) - if after_cursor_id is None: - base = base.offset(offset) - - count_stmt = ( - select(sa.func.count()) - .select_from(AssetReference) - .join(Asset, Asset.id == AssetReference.asset_id) - .where(build_visibility_clause(tenant_id)) - .where(AssetReference.is_missing == False) # noqa: E712 - ) - if name_contains: - escaped, esc = escape_sql_like_string(name_contains) - count_stmt = count_stmt.where( - AssetReference.name.ilike(f"%{escaped}%", escape=esc) - ) - count_stmt = apply_tag_filters(count_stmt, include_tags, exclude_tags, any_tags) - count_stmt = apply_metadata_filter(count_stmt, metadata_filter) - - total = int(session.execute(count_stmt).scalar_one() or 0) - refs = session.execute(base).unique().scalars().all() - - id_list: list[str] = [r.id for r in refs] - tag_map: dict[str, list[str]] = defaultdict(list) - if id_list: - rows = session.execute( - select(AssetReferenceTag.asset_reference_id, Tag.name) - .join(Tag, Tag.name == AssetReferenceTag.tag_name) - .where(AssetReferenceTag.asset_reference_id.in_(id_list)) - .order_by(AssetReferenceTag.tag_name.asc()) - ) - for ref_id, tag_name in rows.all(): - tag_map[ref_id].append(tag_name) - - return list(refs), tag_map, total - - -def fetch_reference_asset_and_tags( - session: Session, - reference_id: str, - tenant_id: str = "", -) -> tuple[AssetReference, Asset, list[str]] | None: - stmt = ( - select(AssetReference, Asset, Tag.name) - .join(Asset, Asset.id == AssetReference.asset_id) - .join( - AssetReferenceTag, - AssetReferenceTag.asset_reference_id == AssetReference.id, - isouter=True, - ) - .join(Tag, Tag.name == AssetReferenceTag.tag_name, isouter=True) - .where( - AssetReference.id == reference_id, - build_visibility_clause(tenant_id), - ) - .options(noload(AssetReference.tags)) - .order_by(Tag.name.asc()) - ) - - rows = session.execute(stmt).all() - if not rows: - return None - - first_ref, first_asset, _ = rows[0] - tags: list[str] = [] - seen: set[str] = set() - for _ref, _asset, tag_name in rows: - if tag_name and tag_name not in seen: - seen.add(tag_name) - tags.append(tag_name) - return first_ref, first_asset, tags - - -def fetch_reference_and_asset( - session: Session, - reference_id: str, - tenant_id: str = "", -) -> tuple[AssetReference, Asset] | None: - stmt = ( - select(AssetReference, Asset) - .join(Asset, Asset.id == AssetReference.asset_id) - .where( - AssetReference.id == reference_id, - build_visibility_clause(tenant_id), - ) - .limit(1) - .options(noload(AssetReference.tags)) - ) - pair = session.execute(stmt).first() - if not pair: - return None - return pair[0], pair[1] - - -def update_reference_access_time( - session: Session, - reference_id: str, - ts: datetime | None = None, - only_if_newer: bool = True, -) -> None: - ts = ts or get_utc_now() - stmt = sa.update(AssetReference).where(AssetReference.id == reference_id) - if only_if_newer: - stmt = stmt.where( - sa.or_( - AssetReference.last_access_time.is_(None), - AssetReference.last_access_time < ts, - ) - ) - session.execute(stmt.values(last_access_time=ts)) - - -def update_reference_name( - session: Session, - reference_id: str, - name: str, -) -> None: - """Update the name of an AssetReference.""" - now = get_utc_now() - session.execute( - sa.update(AssetReference) - .where(AssetReference.id == reference_id) - .values(name=name, updated_at=now) - ) - - -def update_reference_updated_at( - session: Session, - reference_id: str, - ts: datetime | None = None, -) -> None: - """Update the updated_at timestamp of an AssetReference.""" - ts = ts or get_utc_now() - session.execute( - sa.update(AssetReference) - .where(AssetReference.id == reference_id) - .values(updated_at=ts) - ) - - -def rebuild_metadata_projection(session: Session, ref: AssetReference) -> None: - """Delete and rebuild AssetReferenceMeta rows from merged system+user metadata. - - The merged dict is ``{**system_metadata, **user_metadata}`` so user keys - override system keys of the same name. - """ - session.execute( - delete(AssetReferenceMeta).where( - AssetReferenceMeta.asset_reference_id == ref.id - ) - ) - session.flush() - - merged = {**(ref.system_metadata or {}), **(ref.user_metadata or {})} - if not merged: - return - - rows: list[AssetReferenceMeta] = [] - for k, v in merged.items(): - for r in convert_metadata_to_rows(k, v): - rows.append( - AssetReferenceMeta( - asset_reference_id=ref.id, - key=r["key"], - ordinal=int(r["ordinal"]), - val_str=r.get("val_str"), - val_num=r.get("val_num"), - val_bool=r.get("val_bool"), - val_json=r.get("val_json"), - ) - ) - if rows: - session.add_all(rows) - session.flush() - - -def set_reference_metadata( - session: Session, - reference_id: str, - user_metadata: dict | None = None, -) -> None: - ref = session.get(AssetReference, reference_id) - if not ref: - raise ValueError(f"AssetReference {reference_id} not found") - - ref.user_metadata = user_metadata or {} - ref.updated_at = get_utc_now() - session.flush() - - rebuild_metadata_projection(session, ref) - - -def set_reference_system_metadata( - session: Session, - reference_id: str, - system_metadata: dict | None = None, -) -> None: - """Set system_metadata on a reference and rebuild the merged projection.""" - ref = session.get(AssetReference, reference_id) - if not ref: - raise ValueError(f"AssetReference {reference_id} not found") - - ref.system_metadata = system_metadata or {} - ref.updated_at = get_utc_now() - session.flush() - - rebuild_metadata_projection(session, ref) - - -def delete_reference_by_id( - session: Session, - reference_id: str, - tenant_id: str, -) -> bool: - stmt = sa.delete(AssetReference).where( - AssetReference.id == reference_id, - build_visibility_clause(tenant_id), - ) - return int(session.execute(stmt).rowcount or 0) > 0 - - - -def set_reference_preview( - session: Session, - reference_id: str, - preview_reference_id: str | None = None, -) -> None: - """Set or clear preview_id and bump updated_at. Raises on unknown IDs.""" - ref = session.get(AssetReference, reference_id) - if not ref: - raise ValueError(f"AssetReference {reference_id} not found") - - if preview_reference_id is None: - ref.preview_id = None - else: - if not session.get(AssetReference, preview_reference_id): - raise ValueError(f"Preview AssetReference {preview_reference_id} not found") - ref.preview_id = preview_reference_id - - ref.updated_at = get_utc_now() - session.flush() - - -class CacheStateRow(NamedTuple): - """Row from reference query with cache state data.""" - - reference_id: str - file_path: str - mtime_ns: int | None - pending_verification: bool - asset_id: str - asset_hash: str | None - size_bytes: int | None - - -def list_references_by_asset_id( - session: Session, - asset_id: str, -) -> Sequence[AssetReference]: - return ( - session.execute( - select(AssetReference) - .where(AssetReference.asset_id == asset_id) - .where(AssetReference.is_missing == False) # noqa: E712 - .order_by(AssetReference.id.asc()) - ) - .scalars() - .all() - ) - - -def list_all_file_paths_by_asset_id( - session: Session, - asset_id: str, -) -> list[str]: - """Return every file_path for an asset, including soft-deleted/missing refs. - - Used for orphan cleanup where all on-disk files must be removed. - """ - return list( - session.execute( - select(AssetReference.file_path) - .where(AssetReference.asset_id == asset_id) - .where(AssetReference.file_path.isnot(None)) - ) - .scalars() - .all() - ) - - -def upsert_reference( - session: Session, - asset_id: str, - file_path: str, - name: str, - mtime_ns: int, - tenant_id: str = "", - loader_path: str | None = None, -) -> tuple[bool, bool]: - """Upsert a reference by file_path. Returns (created, updated). - - Also restores references that were previously marked as missing. - """ - now = get_utc_now() - vals = { - "asset_id": asset_id, - "file_path": file_path, - "loader_path": loader_path, - "name": name, - "tenant_id": tenant_id, - "mtime_ns": int(mtime_ns), - "is_missing": False, - "created_at": now, - "updated_at": now, - "last_access_time": now, - } - ins = ( - sqlite.insert(AssetReference) - .values(**vals) - .on_conflict_do_nothing(index_elements=[AssetReference.file_path]) - ) - res = session.execute(ins) - created = int(res.rowcount or 0) > 0 - - if created: - return True, False - - upd = ( - sa.update(AssetReference) - .where(AssetReference.file_path == file_path) - .where( - sa.or_( - AssetReference.asset_id != asset_id, - AssetReference.mtime_ns.is_(None), - AssetReference.mtime_ns != int(mtime_ns), - AssetReference.loader_path.is_distinct_from(loader_path), - AssetReference.is_missing == True, # noqa: E712 - ) - ) - .values( - asset_id=asset_id, mtime_ns=int(mtime_ns), loader_path=loader_path, - is_missing=False, updated_at=now, - ) - ) - res2 = session.execute(upd) - updated = int(res2.rowcount or 0) > 0 - return False, updated - - -def mark_references_missing_outside_prefixes( - session: Session, - valid_prefixes: list[str], -) -> int: - """Mark references as missing when file_path doesn't match any valid prefix. - - Returns number of references marked as missing. - """ - if not valid_prefixes: - return 0 - - conds = build_prefix_like_conditions(valid_prefixes) - matches_valid_prefix = sa.or_(*conds) - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.file_path.isnot(None)) - .where(~matches_valid_prefix) - .where(AssetReference.is_missing == False) # noqa: E712 - .values(is_missing=True) - ) - return result.rowcount - - -def restore_references_by_paths(session: Session, file_paths: list[str]) -> int: - """Restore references that were previously marked as missing. - - Returns number of references restored. - """ - if not file_paths: - return 0 - - total = 0 - for chunk in iter_chunks(file_paths, MAX_BIND_PARAMS): - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.file_path.in_(chunk)) - .where(AssetReference.is_missing == True) # noqa: E712 - .values(is_missing=False) - ) - total += result.rowcount - return total - - -def get_unreferenced_unhashed_asset_ids(session: Session) -> list[str]: - """Get IDs of unhashed assets (hash=None) with no active references. - - An asset is considered unreferenced if it has no references, - or all its references are marked as missing. - - Returns list of asset IDs that are unreferenced. - """ - active_ref_exists = ( - sa.select(sa.literal(1)) - .where(AssetReference.asset_id == Asset.id) - .where(AssetReference.is_missing == False) # noqa: E712 - .correlate(Asset) - .exists() - ) - unreferenced_subq = sa.select(Asset.id).where( - Asset.hash.is_(None), ~active_ref_exists - ) - return [row[0] for row in session.execute(unreferenced_subq).all()] - - -def delete_assets_by_ids(session: Session, asset_ids: list[str]) -> int: - """Delete assets and their references by ID. - - Returns number of assets deleted. - """ - if not asset_ids: - return 0 - total = 0 - for chunk in iter_chunks(asset_ids, MAX_BIND_PARAMS): - session.execute( - sa.delete(AssetReference).where(AssetReference.asset_id.in_(chunk)) - ) - result = session.execute(sa.delete(Asset).where(Asset.id.in_(chunk))) - total += result.rowcount - return total - - -def get_references_for_prefixes( - session: Session, - prefixes: list[str], - *, - include_missing: bool = False, -) -> list[CacheStateRow]: - """Get all references with file paths matching any of the given prefixes. - - Args: - session: Database session - prefixes: List of absolute directory prefixes to match - include_missing: If False (default), exclude references marked as missing - - Returns: - List of cache state rows with joined asset data - """ - if not prefixes: - return [] - - conds = build_prefix_like_conditions(prefixes) - - query = ( - sa.select( - AssetReference.id, - AssetReference.file_path, - AssetReference.mtime_ns, - AssetReference.pending_verification, - AssetReference.asset_id, - Asset.hash, - Asset.size_bytes, - ) - .join(Asset, Asset.id == AssetReference.asset_id) - .where(AssetReference.file_path.isnot(None)) - .where(sa.or_(*conds)) - ) - - if not include_missing: - query = query.where(AssetReference.is_missing == False) # noqa: E712 - - rows = session.execute( - query.order_by(AssetReference.asset_id.asc(), AssetReference.id.asc()) - ).all() - - return [ - CacheStateRow( - reference_id=row[0], - file_path=row[1], - mtime_ns=row[2], - pending_verification=row[3], - asset_id=row[4], - asset_hash=row[5], - size_bytes=int(row[6]) if row[6] is not None else None, - ) - for row in rows - ] - - -def bulk_update_pending_verification( - session: Session, reference_ids: list[str], value: bool -) -> int: - """Set pending_verification flag for multiple references. - - Returns: Number of rows updated - """ - if not reference_ids: - return 0 - total = 0 - for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.id.in_(chunk)) - .values(pending_verification=value) - ) - total += result.rowcount - return total - - -def bulk_update_is_missing( - session: Session, reference_ids: list[str], value: bool -) -> int: - """Set is_missing flag for multiple references. - - Returns: Number of rows updated - """ - if not reference_ids: - return 0 - total = 0 - for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.id.in_(chunk)) - .values(is_missing=value) - ) - total += result.rowcount - return total - - -def update_is_missing_by_asset_id( - session: Session, asset_id: str, value: bool -) -> int: - """Set is_missing flag for ALL references belonging to an asset. - - Returns: Number of rows updated - """ - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.asset_id == asset_id) - .values(is_missing=value) - ) - return result.rowcount - - -def delete_references_by_ids(session: Session, reference_ids: list[str]) -> int: - """Delete references by their IDs. - - Returns: Number of rows deleted - """ - if not reference_ids: - return 0 - total = 0 - for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): - result = session.execute( - sa.delete(AssetReference).where(AssetReference.id.in_(chunk)) - ) - total += result.rowcount - return total - - -def delete_orphaned_seed_asset(session: Session, asset_id: str) -> bool: - """Delete a seed asset (hash is None) and its references. - - Returns: True if asset was deleted, False if not found or has a hash - """ - asset = session.get(Asset, asset_id) - if not asset: - return False - if asset.hash is not None: - return False - session.execute( - sa.delete(AssetReference).where(AssetReference.asset_id == asset_id) - ) - session.delete(asset) - return True - - -class UnenrichedReferenceRow(NamedTuple): - """Row for references needing enrichment.""" - - reference_id: str - asset_id: str - file_path: str - hash_state: int - - -def get_unenriched_references( - session: Session, - prefixes: list[str], - max_level: int = 0, - limit: int = 1000, -) -> list[UnenrichedReferenceRow]: - """Get references that need enrichment (hash state <= max_level). - - Args: - session: Database session - prefixes: List of absolute directory prefixes to scan - max_level: Maximum enrichment level to include (0=stubs, 1=metadata done) - limit: Maximum number of rows to return - - Returns: - List of unenriched reference rows with file paths - """ - if not prefixes: - return [] - - conds = build_prefix_like_conditions(prefixes) - - query = ( - sa.select( - AssetReference.id, - AssetReference.asset_id, - AssetReference.file_path, - AssetReference.hash_state, - ) - .where(AssetReference.file_path.isnot(None)) - .where(sa.or_(*conds)) - .where(AssetReference.is_missing == False) # noqa: E712 - .where(AssetReference.hash_state <= max_level) - .order_by(AssetReference.id.asc()) - .limit(limit) - ) - - rows = session.execute(query).all() - return [ - UnenrichedReferenceRow( - reference_id=row[0], - asset_id=row[1], - file_path=row[2], - hash_state=row[3], - ) - for row in rows - ] - - -def bulk_update_hash_state( - session: Session, - reference_ids: list[str], - level: int, -) -> int: - """Update enrichment level for multiple references. - - Returns: Number of rows updated - """ - if not reference_ids: - return 0 - result = session.execute( - sa.update(AssetReference) - .where(AssetReference.id.in_(reference_ids)) - .values(hash_state=level) - ) - return result.rowcount - - -def bulk_insert_references_ignore_conflicts( - session: Session, - rows: list[dict], -) -> None: - """Bulk insert reference rows with ON CONFLICT DO NOTHING on file_path. - - Each dict should have: id, asset_id, file_path, name, tenant_id, mtime_ns, etc. - The is_missing field is automatically set to False for new inserts. - """ - if not rows: - return - enriched_rows = [{**row, "is_missing": False} for row in rows] - ins = sqlite.insert(AssetReference).on_conflict_do_nothing( - index_elements=[AssetReference.file_path] - ) - for chunk in iter_chunks(enriched_rows, calculate_rows_per_statement(14)): - session.execute(ins, chunk) - - -def get_references_by_paths_and_asset_ids( - session: Session, - path_to_asset: dict[str, str], -) -> set[str]: - """Query references to find paths where our asset_id won the insert. - - Args: - path_to_asset: Mapping of file_path -> asset_id we tried to insert - - Returns: - Set of file_paths where our asset_id is present - """ - if not path_to_asset: - return set() - - pairs = list(path_to_asset.items()) - winners: set[str] = set() - - # Each pair uses 2 bind params, so chunk at MAX_BIND_PARAMS // 2 - for chunk in iter_chunks(pairs, MAX_BIND_PARAMS // 2): - pairwise = sa.tuple_(AssetReference.file_path, AssetReference.asset_id).in_( - chunk - ) - result = session.execute( - select(AssetReference.file_path).where(pairwise) - ) - winners.update(result.scalars().all()) - - return winners - - -def get_reference_paths_by_ids( - session: Session, - reference_ids: list[str], -) -> dict[str, str]: - """Map reference id -> file_path for live, file-backed references.""" - if not reference_ids: - return {} - - paths: dict[str, str] = {} - for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): - rows = session.execute( - select(AssetReference.id, AssetReference.file_path).where( - AssetReference.id.in_(chunk), - AssetReference.file_path.is_not(None), - ) - ) - paths.update({rid: fp for rid, fp in rows}) - return paths - - -def get_reference_ids_by_ids( - session: Session, - reference_ids: list[str], -) -> set[str]: - """Query to find which reference IDs exist in the database.""" - if not reference_ids: - return set() - - found: set[str] = set() - for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): - result = session.execute( - select(AssetReference.id).where(AssetReference.id.in_(chunk)) - ) - found.update(result.scalars().all()) - return found diff --git a/app/assets/database/queries/common.py b/app/assets/database/queries/common.py index e937c48b5..70d496971 100644 --- a/app/assets/database/queries/common.py +++ b/app/assets/database/queries/common.py @@ -15,19 +15,3 @@ def iter_chunks(seq, n: int): def iter_row_chunks(rows, cols_per_row: int): yield from iter_chunks(rows, calculate_rows_per_statement(cols_per_row)) - - -def build_visibility_clause(*_args, **_kwargs): - raise NotImplementedError("Asset-reference queries were removed in the B schema") - - -def build_prefix_like_conditions(*_args, **_kwargs): - raise NotImplementedError("Asset-reference queries were removed in the B schema") - - -def apply_tag_filters(*_args, **_kwargs): - raise NotImplementedError("Asset-reference queries were removed in the B schema") - - -def apply_metadata_filter(*_args, **_kwargs): - raise NotImplementedError("Asset-reference queries were removed in the B schema") diff --git a/app/assets/database/queries/tags.py b/app/assets/database/queries/tags.py index 4b600c70d..a2ffbf110 100644 --- a/app/assets/database/queries/tags.py +++ b/app/assets/database/queries/tags.py @@ -5,26 +5,21 @@ from dataclasses import dataclass from typing import Iterable, Sequence import sqlalchemy as sa -from sqlalchemy import delete, func, select +from sqlalchemy import func, select from sqlalchemy.dialects import sqlite -from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.assets.database.models import ( Asset, AssetContent, - AssetReference, - AssetReferenceMeta, - AssetReferenceTag, AssetTag, Tag, ) -from app.assets.database.queries.common import iter_row_chunks from app.assets.database.queries.records import ( build_record_tag_filter_clauses, live_asset_content_clause, ) -from app.assets.helpers import escape_sql_like_string, get_utc_now, normalize_tags +from app.assets.helpers import escape_sql_like_string, normalize_tags @dataclass(frozen=True) @@ -41,13 +36,6 @@ class RemoveTagsResult: total_tags: list[str] -@dataclass(frozen=True) -class SetTagsResult: - added: list[str] - removed: list[str] - total: list[str] - - def validate_tags_exist(session: Session, tags: list[str]) -> None: """Raise ValueError if any of the given tag names do not exist.""" existing_tag_names = set( @@ -72,125 +60,6 @@ def ensure_tags_exist(session: Session, names: Iterable[str]) -> None: session.execute(ins) -def get_reference_tags(session: Session, reference_id: str) -> list[str]: - return [ - tag_name - for (tag_name,) in ( - session.execute( - select(AssetReferenceTag.tag_name) - .where(AssetReferenceTag.asset_reference_id == reference_id) - .order_by(AssetReferenceTag.tag_name.asc()) - ) - ).all() - ] - - -def set_reference_tags( - session: Session, - reference_id: str, - tags: Sequence[str], - origin: str = "manual", -) -> SetTagsResult: - desired = normalize_tags(tags) - - current = set(get_reference_tags(session, reference_id)) - - to_add = [t for t in desired if t not in current] - to_remove = [t for t in current if t not in desired] - - if to_add: - ensure_tags_exist(session, to_add) - session.add_all( - [ - AssetReferenceTag( - asset_reference_id=reference_id, - tag_name=t, - origin=origin, - added_at=get_utc_now(), - ) - for t in to_add - ] - ) - session.flush() - - if to_remove: - session.execute( - delete(AssetReferenceTag).where( - AssetReferenceTag.asset_reference_id == reference_id, - AssetReferenceTag.tag_name.in_(to_remove), - ) - ) - session.flush() - - return SetTagsResult(added=sorted(to_add), removed=sorted(to_remove), total=sorted(desired)) - - -def add_tags_to_reference( - session: Session, - reference_id: str, - tags: Sequence[str], - origin: str = "manual", - create_if_missing: bool = True, - reference_row: AssetReference | None = None, -) -> AddTagsResult: - if not reference_row: - ref = session.get(AssetReference, reference_id) - if not ref: - raise ValueError(f"AssetReference {reference_id} not found") - - norm = normalize_tags(tags) - if not norm: - total = get_reference_tags(session, reference_id=reference_id) - return AddTagsResult(added=[], already_present=[], total_tags=total) - - if create_if_missing: - ensure_tags_exist(session, norm) - - current = set(get_reference_tags(session, reference_id)) - - want = set(norm) - to_add = sorted(want - current) - - if to_add: - with session.begin_nested() as nested: - try: - session.add_all( - [ - AssetReferenceTag( - asset_reference_id=reference_id, - tag_name=t, - origin=origin, - added_at=get_utc_now(), - ) - for t in to_add - ] - ) - session.flush() - except IntegrityError: - nested.rollback() - - after = set(get_reference_tags(session, reference_id=reference_id)) - return AddTagsResult( - added=sorted(((after - current) & want)), - already_present=sorted(want & current), - total_tags=sorted(after), - ) - - -def remove_missing_tag_for_asset_id( - session: Session, - asset_id: str, -) -> None: - session.execute( - sa.delete(AssetReferenceTag).where( - AssetReferenceTag.asset_reference_id.in_( - sa.select(AssetReference.id).where(AssetReference.asset_id == asset_id) - ), - AssetReferenceTag.tag_name == "missing", - ) - ) - - def list_tags_with_usage( session: Session, prefix: str | None = None, @@ -308,39 +177,3 @@ def list_tag_counts_for_filtered_assets( rows = session.execute(q).all() return {tag_name: int(cnt) for tag_name, cnt in rows} - - -def bulk_insert_tags_and_meta( - session: Session, - tag_rows: list[dict], - meta_rows: list[dict], -) -> None: - """Batch insert into asset_reference_tags and asset_reference_meta. - - Uses ON CONFLICT DO NOTHING. - - Args: - session: Database session - tag_rows: Dicts with: asset_reference_id, tag_name, origin, added_at - meta_rows: Dicts with: asset_reference_id, key, ordinal, val_* - """ - if tag_rows: - ins_tags = sqlite.insert(AssetReferenceTag).on_conflict_do_nothing( - index_elements=[ - AssetReferenceTag.asset_reference_id, - AssetReferenceTag.tag_name, - ] - ) - for chunk in iter_row_chunks(tag_rows, cols_per_row=4): - session.execute(ins_tags, chunk) - - if meta_rows: - ins_meta = sqlite.insert(AssetReferenceMeta).on_conflict_do_nothing( - index_elements=[ - AssetReferenceMeta.asset_reference_id, - AssetReferenceMeta.key, - AssetReferenceMeta.ordinal, - ] - ) - for chunk in iter_row_chunks(meta_rows, cols_per_row=7): - session.execute(ins_meta, chunk) diff --git a/app/assets/scanner.py b/app/assets/scanner.py index ce96c29b8..0485542cd 100644 --- a/app/assets/scanner.py +++ b/app/assets/scanner.py @@ -2,7 +2,7 @@ import logging import os from dataclasses import dataclass from pathlib import Path -from typing import Callable, Literal +from typing import Callable, Literal, TypedDict import folder_paths import sqlalchemy as sa @@ -32,11 +32,9 @@ from app.assets.scanner_admission import ( _two_stat_admit, tick_watch_list as tick_watch_list, ) -from app.assets.services.bulk_ingest import SeedAssetSpec from app.assets.services.file_utils import get_mtime_ns, is_visible, list_files_recursively -from app.assets.services.hashing import compute_blake3_hash from app.assets.services.image_dimensions import extract_image_dimensions -from app.assets.services.metadata_extract import extract_file_metadata +from app.assets.services.metadata_extract import ExtractedMetadata, extract_file_metadata from app.assets.services.path_utils import ( compute_loader_path, get_comfy_models_folders, @@ -56,6 +54,20 @@ __all__ = [ RootType = Literal["models", "input", "output"] +class SeedAssetSpec(TypedDict): + """Spec for seeding an asset from filesystem.""" + + abs_path: str + size_bytes: int + mtime_ns: int + info_name: str + tags: list[str] + fname: str | None + metadata: ExtractedMetadata | None + mime_type: str | None + job_id: str | None + + @dataclass(frozen=True, slots=True) class UnenrichedContent: content_id: str @@ -244,7 +256,6 @@ def build_asset_specs( paths: list[str], existing_paths: set[str], enable_metadata_extraction: bool = True, - compute_hashes: bool = False, ) -> tuple[list[SeedAssetSpec], set[str], int]: """Build asset specs from paths, returning (specs, tag_pool, skipped_count). @@ -252,7 +263,6 @@ def build_asset_specs( paths: List of file paths to process existing_paths: Set of paths that already exist in the database enable_metadata_extraction: If True, extract tier 1 & 2 metadata - compute_hashes: If True, compute blake3 hashes (slow for large files) """ specs: list[SeedAssetSpec] = [] tag_pool: set[str] = set() @@ -291,15 +301,6 @@ def build_asset_specs( relative_filename=rel_fname, ) - # Compute hash if requested - asset_hash: str | None = None - if compute_hashes: - try: - digest, _ = compute_blake3_hash(abs_p) - asset_hash = digest - except Exception as e: - logging.warning("Failed to hash %s: %s", abs_p, e) - mime_type = metadata.content_type if metadata else None specs.append( { @@ -310,7 +311,6 @@ def build_asset_specs( "tags": tags, "fname": rel_fname, "metadata": metadata, - "hash": asset_hash, "mime_type": mime_type, "job_id": None, } @@ -465,9 +465,7 @@ def enrich_asset( except Exception as e: logging.warning("Failed to hash %s: %s", file_path, e) - # Optimistic guard: if the content's mtime_ns changed since we - # started (e.g. ingest_existing_file updated it), our results are - # stale — discard them to avoid overwriting fresh registration data. + # Optimistic guard: discard results if content changed during enrichment. content = session.get(AssetContent, content_id) record = session.get(Asset, record_id) if content is None or record is None or content.mtime_ns != initial_mtime_ns: diff --git a/app/assets/scanner_admission.py b/app/assets/scanner_admission.py index 23263e141..9666112ea 100644 --- a/app/assets/scanner_admission.py +++ b/app/assets/scanner_admission.py @@ -76,7 +76,6 @@ def tick_watch_list(session: Session) -> None: "tags": tags, "fname": compute_loader_path(entry.path), "metadata": None, - "hash": None, "mime_type": mimetypes.guess_type(entry.path, strict=False)[0], "job_id": None, } diff --git a/app/assets/seeder.py b/app/assets/seeder.py index 70663b721..98b50324d 100644 --- a/app/assets/seeder.py +++ b/app/assets/seeder.py @@ -708,7 +708,6 @@ class _AssetSeeder: paths, existing_paths, enable_metadata_extraction=False, - compute_hashes=False, ) logging.debug( "Fast scan: build_asset_specs took %.3fs (%d specs, %d skipped)", diff --git a/app/assets/services/__init__.py b/app/assets/services/__init__.py index 65d68ef39..da4e0fb83 100644 --- a/app/assets/services/__init__.py +++ b/app/assets/services/__init__.py @@ -11,9 +11,7 @@ from app.assets.services.asset_management import ( get_asset_detail, update_asset_metadata, delete_asset_reference, - set_asset_preview, asset_exists, - list_assets_page, get_preview_file_paths, resolve_asset_for_download, ) @@ -33,9 +31,7 @@ __all__ = [ "get_asset_detail", "update_asset_metadata", "delete_asset_reference", - "set_asset_preview", "asset_exists", - "list_assets_page", "get_preview_file_paths", "resolve_asset_for_download", "apply_tags", diff --git a/app/assets/services/asset_management.py b/app/assets/services/asset_management.py index 741266997..00442613f 100644 --- a/app/assets/services/asset_management.py +++ b/app/assets/services/asset_management.py @@ -1,29 +1,12 @@ import mimetypes import os -from datetime import timezone from typing import Sequence -from app.assets.services.cursor import ( - CursorPayload, - InvalidCursorError, - decode_cursor, - decode_cursor_int, - decode_cursor_time, - encode_cursor, - encode_cursor_from_time, -) - - from app.assets.database.models import AssetContent from app.assets.database.queries import ( delete_record, fetch_record_tags, get_record_by_id, - fetch_reference_asset_and_tags, - get_asset_by_hash as queries_get_asset_by_hash, - get_reference_with_owner_check, - list_references_page, - set_reference_preview, update_record_access_time, ) from app.assets.database.queries.records import get_preview_file_paths_by_ids @@ -31,13 +14,9 @@ from app.assets.helpers import normalize_tags from app.assets.services.schemas import ( AssetData, AssetDetailResult, - AssetSummaryData, DownloadResolutionResult, - ListAssetsResult, ReferenceData, UserMetadata, - extract_asset_data, - extract_reference_data, ) from app.database.db import create_session @@ -163,37 +142,6 @@ def delete_asset_reference( return True -def set_asset_preview( - reference_id: str, - preview_reference_id: str | None = None, - tenant_id: str = "", -) -> AssetDetailResult: - with create_session() as session: - get_reference_with_owner_check(session, reference_id, tenant_id) - - set_reference_preview( - session, - reference_id=reference_id, - preview_reference_id=preview_reference_id, - ) - - result = fetch_reference_asset_and_tags( - session, reference_id=reference_id, tenant_id=tenant_id - ) - if not result: - raise RuntimeError("State changed during preview update") - - ref, asset, tags = result - detail = AssetDetailResult( - ref=extract_reference_data(ref), - asset=extract_asset_data(asset), - tags=tags, - ) - session.commit() - - return detail - - def asset_exists(asset_hash: str) -> bool: from app.assets.helpers import validate_blake3_hash from app.assets.services.lookup import lookup_for_view @@ -206,132 +154,6 @@ def asset_exists(asset_hash: str) -> bool: return lookup_for_view(session, canonical) is not None -def get_asset_by_hash(asset_hash: str) -> AssetData | None: - with create_session() as session: - asset = queries_get_asset_by_hash(session, asset_hash=asset_hash) - return extract_asset_data(asset) - - -# Sort fields that support cursor pagination. `last_access_time` is not -# in this list — it falls back to offset/limit. -_CURSOR_SORT_FIELDS = ("created_at", "updated_at", "name", "size") - - -def list_assets_page( - tenant_id: str = "", - include_tags: Sequence[str] | None = None, - exclude_tags: Sequence[str] | None = None, - name_contains: str | None = None, - metadata_filter: dict | None = None, - limit: int = 20, - offset: int = 0, - sort: str = "created_at", - order: str = "desc", - after: str | None = None, - # Appended last so pre-existing positional callers keep binding correctly. - any_tags: Sequence[str] | None = None, -) -> ListAssetsResult: - """List assets with optional cursor pagination. - - When ``after`` is supplied it overrides ``offset``. The cursor's sort field - must match ``sort`` and be in the cursor-supported allowlist; mismatches - raise InvalidCursorError so the handler can map to 400 INVALID_CURSOR. - """ - cursor_value: object | None = None - cursor_id: str | None = None - # Mint next_cursor on every page where the sort is cursor-supported, not - # only when the request itself arrived with a cursor. Otherwise a first - # request (no `after`) returns next_cursor=None and the client can never - # enter cursor mode. - mint_cursor = sort in _CURSOR_SORT_FIELDS - - if after is not None: - if sort not in _CURSOR_SORT_FIELDS: - raise InvalidCursorError( - f"cursor pagination is not supported for sort={sort!r}" - ) - payload = decode_cursor(after, _CURSOR_SORT_FIELDS, expected_order=order) - if payload.sort_field != sort: - raise InvalidCursorError( - f"cursor sort field {payload.sort_field!r} does not match request sort {sort!r}" - ) - cursor_value, cursor_id = _resolve_cursor_value(payload), payload.id - - # Over-fetch by one row so we can distinguish "exactly `limit` rows total - # remaining" from "more rows past this page" without a second query. Drop - # the sentinel before returning. - fetch_limit = limit + 1 if mint_cursor else limit - - with create_session() as session: - refs, tag_map, total = list_references_page( - session, - tenant_id=tenant_id, - include_tags=include_tags, - exclude_tags=exclude_tags, - any_tags=any_tags, - name_contains=name_contains, - metadata_filter=metadata_filter, - limit=fetch_limit, - offset=offset, - sort=sort, - order=order, - after_cursor_value=cursor_value, - after_cursor_id=cursor_id, - ) - - next_cursor: str | None = None - if mint_cursor and len(refs) > limit: - # There's at least one more row past this page — mint a cursor from - # the last row of the page (i.e. index `limit - 1`, since we - # over-fetched), and drop the sentinel. - next_cursor = _encode_next_cursor(refs[limit - 1], sort, order) - refs = refs[:limit] - - items: list[AssetSummaryData] = [] - for ref in refs: - items.append( - AssetSummaryData( - ref=extract_reference_data(ref), - asset=extract_asset_data(ref.asset), - tags=tag_map.get(ref.id, []), - ) - ) - - return ListAssetsResult(items=items, total=total, next_cursor=next_cursor) - - -def _resolve_cursor_value(payload: CursorPayload) -> object: - """Map a decoded cursor payload to a column-typed Python value.""" - if payload.sort_field in ("created_at", "updated_at"): - # DB stores naive UTC; strip tzinfo so the comparison binds against a - # `TIMESTAMP WITHOUT TIME ZONE` column without an offset shift. - return decode_cursor_time(payload).replace(tzinfo=None) - if payload.sort_field == "size": - return decode_cursor_int(payload) - return payload.value # name, str-typed - - -def _encode_next_cursor(ref, sort: str, order: str) -> str | None: - """Mint a cursor pointing at *ref* for the given sort dimension. - - Returns None when the boundary row carries a NULL sort value (e.g. an asset - record whose size_bytes hasn't been backfilled). Continuing pagination - across a NULL boundary is undefined under keyset ordering — better to - truncate cleanly here than to mint a cursor that mis-positions. - """ - if sort == "name": - return encode_cursor("name", ref.name, ref.id, order=order) - if sort == "size": - if ref.asset is None or ref.asset.size_bytes is None: - return None - return encode_cursor("size", str(ref.asset.size_bytes), ref.id, order=order) - # created_at / updated_at — DB datetimes are naive UTC; attach tz before encoding. - value = ref.created_at if sort == "created_at" else ref.updated_at - if value is None: - return None - return encode_cursor_from_time(sort, value.replace(tzinfo=timezone.utc), ref.id, order=order) - - def resolve_hash_to_path( asset_hash: str, tenant_id: str = "", diff --git a/app/assets/services/bulk_ingest.py b/app/assets/services/bulk_ingest.py deleted file mode 100644 index 444495a47..000000000 --- a/app/assets/services/bulk_ingest.py +++ /dev/null @@ -1,274 +0,0 @@ -from __future__ import annotations - -import os -import uuid -from dataclasses import dataclass -from datetime import datetime -from typing import TYPE_CHECKING, Any, TypedDict - -from sqlalchemy.orm import Session - -from app.assets.database.queries import ( - bulk_insert_assets, - bulk_insert_references_ignore_conflicts, - bulk_insert_tags_and_meta, - get_existing_asset_ids, - get_reference_ids_by_ids, - get_references_by_paths_and_asset_ids, - restore_references_by_paths, -) -from app.assets.helpers import get_utc_now - -if TYPE_CHECKING: - from app.assets.services.metadata_extract import ExtractedMetadata - - -class SeedAssetSpec(TypedDict): - """Spec for seeding an asset from filesystem.""" - - abs_path: str - size_bytes: int - mtime_ns: int - info_name: str - tags: list[str] - fname: str - metadata: ExtractedMetadata | None - hash: str | None - mime_type: str | None - job_id: str | None - - -class AssetRow(TypedDict): - """Row data for inserting an Asset.""" - - id: str - hash: str | None - size_bytes: int - mime_type: str | None - created_at: datetime - - -class ReferenceRow(TypedDict): - """Row data for inserting an AssetReference.""" - - id: str - asset_id: str - file_path: str - loader_path: str | None - mtime_ns: int - tenant_id: str - name: str - preview_id: str | None - user_metadata: dict[str, Any] | None - job_id: str | None - created_at: datetime - updated_at: datetime - last_access_time: datetime - - -class TagRow(TypedDict): - """Row data for inserting a Tag.""" - - asset_reference_id: str - tag_name: str - origin: str - added_at: datetime - - -class MetadataRow(TypedDict): - """Row data for inserting asset metadata.""" - - asset_reference_id: str - key: str - ordinal: int - val_str: str | None - val_num: float | None - val_bool: bool | None - val_json: dict[str, Any] | None - - -@dataclass -class BulkInsertResult: - """Result of bulk asset insertion.""" - - inserted_refs: int - won_paths: int - lost_paths: int - - -def batch_insert_seed_assets( - session: Session, - specs: list[SeedAssetSpec], - tenant_id: str = "", -) -> BulkInsertResult: - """Seed assets from filesystem specs in batch. - - Each spec is a dict with keys: - - abs_path: str - - size_bytes: int - - mtime_ns: int - - info_name: str - - tags: list[str] - - fname: Optional[str] - - This function orchestrates: - 1. Insert seed Assets (hash=NULL) - 2. Claim references with ON CONFLICT DO NOTHING on file_path - 3. Query to find winners (paths where our asset_id was inserted) - 4. Retain losers for the scanner's B-schema replacement path to handle - 5. Insert tags and metadata for successfully inserted references - - Returns: - BulkInsertResult with inserted_refs, won_paths, lost_paths - """ - if not specs: - return BulkInsertResult(inserted_refs=0, won_paths=0, lost_paths=0) - - current_time = get_utc_now() - asset_rows: list[AssetRow] = [] - reference_rows: list[ReferenceRow] = [] - path_to_asset_id: dict[str, str] = {} - asset_id_to_ref_data: dict[str, dict] = {} - absolute_path_list: list[str] = [] - - for spec in specs: - absolute_path = os.path.abspath(spec["abs_path"]) - existing_asset_id = path_to_asset_id.get(absolute_path) - if existing_asset_id is not None: - existing_tags = asset_id_to_ref_data[existing_asset_id]["tags"] - asset_id_to_ref_data[existing_asset_id]["tags"] = list( - dict.fromkeys([*existing_tags, *spec["tags"]]) - ) - continue - - asset_id = str(uuid.uuid4()) - reference_id = str(uuid.uuid4()) - absolute_path_list.append(absolute_path) - path_to_asset_id[absolute_path] = asset_id - - mime_type = spec.get("mime_type") - asset_rows.append( - { - "id": asset_id, - "hash": spec.get("hash"), - "size_bytes": spec["size_bytes"], - "mime_type": mime_type, - "created_at": current_time, - } - ) - - # Build user_metadata from extracted metadata or fallback to filename - extracted_metadata = spec.get("metadata") - if extracted_metadata: - user_metadata: dict[str, Any] | None = extracted_metadata.to_user_metadata() - elif spec["fname"]: - user_metadata = {"filename": spec["fname"]} - else: - user_metadata = None - - reference_rows.append( - { - "id": reference_id, - "asset_id": asset_id, - "file_path": absolute_path, - # spec["fname"] is compute_loader_path(abs_path) from build_asset_specs. - "loader_path": spec["fname"], - "mtime_ns": spec["mtime_ns"], - "tenant_id": tenant_id, - "name": spec["info_name"], - "preview_id": None, - "user_metadata": user_metadata, - "job_id": spec.get("job_id"), - "created_at": current_time, - "updated_at": current_time, - "last_access_time": current_time, - } - ) - - asset_id_to_ref_data[asset_id] = { - "reference_id": reference_id, - "tags": spec["tags"], - "filename": spec["fname"], - "extracted_metadata": extracted_metadata, - } - - bulk_insert_assets(session, asset_rows) - - # Filter reference rows to only those whose assets were actually inserted - # (assets with duplicate hashes are silently dropped by ON CONFLICT DO NOTHING) - inserted_asset_ids = get_existing_asset_ids( - session, [r["asset_id"] for r in reference_rows] - ) - reference_rows = [r for r in reference_rows if r["asset_id"] in inserted_asset_ids] - - bulk_insert_references_ignore_conflicts(session, reference_rows) - restore_references_by_paths(session, absolute_path_list) - winning_paths = get_references_by_paths_and_asset_ids(session, path_to_asset_id) - - inserted_paths = { - path - for path in absolute_path_list - if path_to_asset_id[path] in inserted_asset_ids - } - losing_paths = inserted_paths - winning_paths - if not winning_paths: - return BulkInsertResult( - inserted_refs=0, - won_paths=0, - lost_paths=len(losing_paths), - ) - - # Get reference IDs for winners - winning_ref_ids = [ - asset_id_to_ref_data[path_to_asset_id[path]]["reference_id"] - for path in winning_paths - ] - inserted_ref_ids = get_reference_ids_by_ids(session, winning_ref_ids) - - tag_rows: list[TagRow] = [] - metadata_rows: list[MetadataRow] = [] - - if inserted_ref_ids: - for path in winning_paths: - asset_id = path_to_asset_id[path] - ref_data = asset_id_to_ref_data[asset_id] - ref_id = ref_data["reference_id"] - - if ref_id not in inserted_ref_ids: - continue - - for tag in ref_data["tags"]: - tag_rows.append( - { - "asset_reference_id": ref_id, - "tag_name": tag, - "origin": "automatic", - "added_at": current_time, - } - ) - - # Use extracted metadata for meta rows if available - extracted_metadata = ref_data.get("extracted_metadata") - if extracted_metadata: - metadata_rows.extend(extracted_metadata.to_meta_rows(ref_id)) - elif ref_data["filename"]: - # Fallback: just store filename - metadata_rows.append( - { - "asset_reference_id": ref_id, - "key": "filename", - "ordinal": 0, - "val_str": ref_data["filename"], - "val_num": None, - "val_bool": None, - "val_json": None, - } - ) - - bulk_insert_tags_and_meta(session, tag_rows=tag_rows, meta_rows=metadata_rows) - - return BulkInsertResult( - inserted_refs=len(inserted_ref_ids), - won_paths=len(winning_paths), - lost_paths=len(losing_paths), - ) diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py index 8b8278bdd..deac82c3b 100644 --- a/app/assets/services/ingest.py +++ b/app/assets/services/ingest.py @@ -7,47 +7,21 @@ from typing import Any, Sequence from sqlalchemy.orm import Session from app.assets.database.models import Asset -from app.assets.database.queries import ( - add_tags_to_reference, - count_active_siblings, - create_stub_asset, - ensure_tags_exist, - get_asset_by_hash, - get_reference_by_file_path, - get_reference_tags, - get_or_create_reference, - list_references_by_asset_id, - reference_exists, - remove_missing_tag_for_asset_id, - set_reference_metadata, - set_reference_system_metadata, - set_reference_tags, - update_asset_hash_and_mime, - upsert_asset, - upsert_reference as _legacy_upsert_reference, # wave-3-fixes: replaced by B-schema write paths in Wave 3 - validate_tags_exist, -) -from app.assets.helpers import get_utc_now, normalize_tags, to_stored_hash -from app.assets.services.bulk_ingest import batch_insert_seed_assets +from app.assets.helpers import normalize_tags, to_stored_hash from app.assets.services.file_utils import get_mtime_ns, get_size_and_mtime_ns from app.assets.services.image_dimensions import extract_image_dimensions from app.assets.services.metadata_extract import extract_file_metadata from app.assets.services.path_utils import ( compute_loader_path, get_name_and_tags_from_asset_path, - get_path_derived_tags_from_path, resolve_destination_from_tags, validate_path_within_base, ) from app.assets.services.schemas import ( AssetData, - IngestResult, ReferenceData, - RegisterAssetResult, UploadResult, UserMetadata, - extract_asset_data, - extract_reference_data, ) from app.database.db import create_session @@ -115,382 +89,6 @@ def _discard_unreferenced_content(session: Session, content_id: str) -> None: logging.exception("Failed to discard orphan content %s", content_id) -def _ingest_file_from_path( - abs_path: str, - asset_hash: str, - size_bytes: int, - mtime_ns: int, - mime_type: str | None = None, - info_name: str | None = None, - tenant_id: str = "", - preview_id: str | None = None, - user_metadata: UserMetadata = None, - tags: Sequence[str] = (), - tag_origin: str = "manual", - require_existing_tags: bool = False, -) -> IngestResult: - locator = os.path.abspath(abs_path) - user_metadata = user_metadata or {} - - asset_created = False - asset_updated = False - ref_created = False - ref_updated = False - reference_id: str | None = None - - with create_session() as session: - if preview_id: - if not reference_exists(session, preview_id): - preview_id = None - - asset, asset_created, asset_updated = upsert_asset( - session, - asset_hash=asset_hash, - size_bytes=size_bytes, - mime_type=mime_type, - ) - - ref_created, ref_updated = _legacy_upsert_reference( # wave-3-fixes: replaced in Wave 3 - session, - asset_id=asset.id, - file_path=locator, - name=info_name or os.path.basename(locator), - mtime_ns=mtime_ns, - tenant_id=tenant_id, - loader_path=compute_loader_path(locator), - ) - - # Get the reference we just created/updated - ref = get_reference_by_file_path(session, locator) - if ref: - reference_id = ref.id - - if preview_id and ref.preview_id != preview_id: - ref.preview_id = preview_id - - try: - backend_tags = get_path_derived_tags_from_path(locator) - except ValueError: - backend_tags = [] - caller_tags = normalize_tags(tags) - backend_tags = normalize_tags(backend_tags) - all_tags = normalize_tags([*caller_tags, *backend_tags]) - if all_tags: - if require_existing_tags: - validate_tags_exist(session, all_tags) - if backend_tags: - add_tags_to_reference( - session, - reference_id=reference_id, - tags=backend_tags, - origin="automatic", - create_if_missing=not require_existing_tags, - ) - if caller_tags: - add_tags_to_reference( - session, - reference_id=reference_id, - tags=caller_tags, - origin=tag_origin, - create_if_missing=not require_existing_tags, - ) - - _update_metadata_with_filename( - session, - reference_id=reference_id, - file_path=ref.file_path, - current_metadata=ref.user_metadata, - user_metadata=user_metadata, - ) - - _maybe_store_image_dimensions( - session, - reference_id=reference_id, - file_path=locator, - mime_type=mime_type, - current_system_metadata=ref.system_metadata, - ) - - try: - remove_missing_tag_for_asset_id(session, asset_id=asset.id) - except Exception: - logging.exception("Failed to clear 'missing' tag for asset %s", asset.id) - - session.commit() - - return IngestResult( - asset_created=asset_created, - asset_updated=asset_updated, - ref_created=ref_created, - ref_updated=ref_updated, - reference_id=reference_id, - ) - - -def ingest_existing_file( - abs_path: str, - user_metadata: UserMetadata = None, - extra_tags: Sequence[str] = (), - tenant_id: str = "", - job_id: str | None = None, -) -> bool: - """Register an existing on-disk file as an asset stub. - - If a reference already exists for this path, updates mtime_ns, job_id, - size_bytes, and resets enrichment so the enricher will re-hash it. - - For brand-new paths, inserts a stub record (hash=NULL) for immediate - UX visibility. - - Returns True if a row was inserted or updated, False otherwise. - """ - locator = os.path.abspath(abs_path) - size_bytes, mtime_ns = get_size_and_mtime_ns(abs_path) - mime_type = mimetypes.guess_type(abs_path, strict=False)[0] - name, path_tags = get_name_and_tags_from_asset_path(abs_path) - tags = list(dict.fromkeys(path_tags + list(extra_tags))) - - with create_session() as session: - existing_ref = get_reference_by_file_path(session, locator) - if existing_ref is not None: - now = get_utc_now() - existing_ref.mtime_ns = mtime_ns - existing_ref.job_id = job_id - existing_ref.is_missing = False - existing_ref.updated_at = now - existing_ref.hash_state = 0 - - asset = existing_ref.asset - if asset: - # If other refs share this asset, detach to a new stub - # instead of mutating the shared row. - siblings = count_active_siblings(session, asset.id, existing_ref.id) - if siblings > 0: - new_asset = create_stub_asset( - session, - size_bytes=size_bytes, - mime_type=mime_type or asset.mime_type, - ) - existing_ref.asset_id = new_asset.id - else: - asset.hash = None - asset.size_bytes = size_bytes - if mime_type: - asset.mime_type = mime_type - session.commit() - return True - - spec = { - "abs_path": abs_path, - "size_bytes": size_bytes, - "mtime_ns": mtime_ns, - "info_name": name, - "tags": tags, - "fname": compute_loader_path(abs_path), - "metadata": None, - "hash": None, - "mime_type": mime_type, - "job_id": job_id, - } - if tags: - ensure_tags_exist(session, tags) - result = batch_insert_seed_assets(session, [spec], tenant_id=tenant_id) - session.commit() - return result.won_paths > 0 - - -def _register_existing_asset( - asset_hash: str, - name: str, - user_metadata: UserMetadata = None, - tags: list[str] | None = None, - tag_origin: str = "manual", - tenant_id: str = "", - mime_type: str | None = None, - preview_id: str | None = None, -) -> RegisterAssetResult: - user_metadata = user_metadata or {} - - with create_session() as session: - asset = get_asset_by_hash(session, asset_hash=asset_hash) - if not asset: - raise ValueError(f"No asset with hash {asset_hash}") - - if mime_type and not asset.mime_type: - update_asset_hash_and_mime(session, asset_id=asset.id, mime_type=mime_type) - - if preview_id: - if not reference_exists(session, preview_id): - preview_id = None - - ref, ref_created = get_or_create_reference( - session, - asset_id=asset.id, - tenant_id=tenant_id, - name=name, - preview_id=preview_id, - ) - - if not ref_created: - if preview_id and ref.preview_id != preview_id: - ref.preview_id = preview_id - - tag_names = get_reference_tags(session, reference_id=ref.id) - result = RegisterAssetResult( - ref=extract_reference_data(ref), - asset=extract_asset_data(asset), - tags=tag_names, - created=False, - ) - session.commit() - return result - - new_meta = dict(user_metadata) - computed_filename = compute_loader_path(ref.file_path) if ref.file_path else None - if computed_filename: - new_meta["filename"] = computed_filename - - if new_meta: - set_reference_metadata( - session, - reference_id=ref.id, - user_metadata=new_meta, - ) - - _backfill_image_dimensions_from_siblings( - session, - asset_id=asset.id, - new_reference_id=ref.id, - current_system_metadata=ref.system_metadata, - ) - - if tags is not None: - set_reference_tags( - session, - reference_id=ref.id, - tags=tags, - origin=tag_origin, - ) - - tag_names = get_reference_tags(session, reference_id=ref.id) - session.refresh(ref) - result = RegisterAssetResult( - ref=extract_reference_data(ref), - asset=extract_asset_data(asset), - tags=tag_names, - created=True, - ) - session.commit() - - return result - - - -def _update_metadata_with_filename( - session: Session, - reference_id: str, - file_path: str | None, - current_metadata: dict | None, - user_metadata: dict[str, Any], -) -> None: - computed_filename = compute_loader_path(file_path) if file_path else None - - current_meta = current_metadata or {} - new_meta = dict(current_meta) - for k, v in user_metadata.items(): - new_meta[k] = v - if computed_filename: - new_meta["filename"] = computed_filename - - if new_meta != current_meta: - set_reference_metadata( - session, - reference_id=reference_id, - user_metadata=new_meta, - ) - - -_IMAGE_DIMENSION_KEYS = ("kind", "width", "height") - - -def _maybe_store_image_dimensions( - session: Session, - reference_id: str, - file_path: str, - mime_type: str | None, - current_system_metadata: dict | None, -) -> None: - """Populate ``kind``/``width``/``height`` on system_metadata for image refs. - - Non-image MIME types are a no-op. Pre-existing keys (e.g. enricher-written - safetensors metadata, download provenance) are preserved by merge. - """ - if not mime_type or not mime_type.startswith("image/"): - return - - dims = extract_image_dimensions(file_path, mime_type=mime_type) - if not dims: - return - - current = current_system_metadata or {} - merged = dict(current) - merged.update(dims) - if merged != current: - set_reference_system_metadata( - session, - reference_id=reference_id, - system_metadata=merged, - ) - - -def _backfill_image_dimensions_from_siblings( - session: Session, - asset_id: str, - new_reference_id: str, - current_system_metadata: dict | None, -) -> None: - """Copy image dimension keys from any sibling reference of the same asset. - - The from-hash path doesn't read the file bytes, so dimensions can't be - extracted there directly. When another reference of the same asset already - carries image dimensions, copy them onto the new reference so consumers - see consistent metadata regardless of how the asset was registered. - - Best-effort: missing siblings, non-image siblings, or absent dimension - keys leave the target reference unchanged. - """ - current = current_system_metadata or {} - if current.get("kind") == "image" and "width" in current and "height" in current: - return - - for sibling in list_references_by_asset_id(session, asset_id): - if sibling.id == new_reference_id: - continue - meta = sibling.system_metadata or {} - if meta.get("kind") != "image": - continue - width = meta.get("width") - height = meta.get("height") - if ( - type(width) is not int - or type(height) is not int - or width <= 0 - or height <= 0 - ): - continue - merged = dict(current) - merged["kind"] = "image" - merged["width"] = width - merged["height"] = height - if merged != current: - set_reference_system_metadata( - session, - reference_id=new_reference_id, - system_metadata=merged, - ) - return - - def _sanitize_filename(name: str | None, fallback: str) -> str: n = os.path.basename((name or "").strip() or fallback) return n if n else fallback diff --git a/app/assets/services/schemas.py b/app/assets/services/schemas.py index 0fda6871d..cbc56b7b3 100644 --- a/app/assets/services/schemas.py +++ b/app/assets/services/schemas.py @@ -2,8 +2,6 @@ from dataclasses import dataclass from datetime import datetime from typing import Any, NamedTuple -from app.assets.database.models import Asset, AssetReference - UserMetadata = dict[str, Any] | None @@ -87,29 +85,3 @@ class UploadResult: asset: AssetData tags: list[str] created_new: bool - - -def extract_reference_data(ref: AssetReference) -> ReferenceData: - return ReferenceData( - id=ref.id, - name=ref.name, - file_path=ref.file_path, - loader_path=ref.loader_path, - user_metadata=ref.user_metadata, - preview_id=ref.preview_id, - system_metadata=ref.system_metadata, - job_id=ref.job_id, - created_at=ref.created_at, - updated_at=ref.updated_at, - last_access_time=ref.last_access_time, - ) - - -def extract_asset_data(asset: Asset | None) -> AssetData | None: - if asset is None: - return None - return AssetData( - hash=asset.hash, - size_bytes=asset.size_bytes, - mime_type=asset.mime_type, - ) diff --git a/tests-unit/assets_test/services/test_recovery_gate.py b/tests-unit/assets_test/services/test_recovery_gate.py index ce9c71aea..fb0a7b8df 100644 --- a/tests-unit/assets_test/services/test_recovery_gate.py +++ b/tests-unit/assets_test/services/test_recovery_gate.py @@ -7,11 +7,11 @@ from sqlalchemy import select from app.assets.database.models import Asset, AssetContent, AssetTag, Tag from app.assets.helpers import to_stored_hash from app.assets.scanner import ( + SeedAssetSpec, clear_pending_verifications, pending_recovery_count, seed_asset_specs, ) -from app.assets.services.bulk_ingest import SeedAssetSpec from app.assets.services.snapshot_hash import snapshot_hash @@ -46,7 +46,6 @@ def _spec(path: Path) -> SeedAssetSpec: "tags": ["input"], "fname": path.name, "metadata": None, - "hash": None, "mime_type": None, "job_id": None, } diff --git a/tests-unit/assets_test/services/test_scanner_seed_resilience.py b/tests-unit/assets_test/services/test_scanner_seed_resilience.py index f3ac40ba5..9f10f9d49 100644 --- a/tests-unit/assets_test/services/test_scanner_seed_resilience.py +++ b/tests-unit/assets_test/services/test_scanner_seed_resilience.py @@ -9,8 +9,7 @@ from sqlalchemy.orm import Session from app.assets.database.models import Asset from app.assets.database.queries import create_record as create_record_query -from app.assets.scanner import seed_asset_specs -from app.assets.services.bulk_ingest import SeedAssetSpec +from app.assets.scanner import SeedAssetSpec, seed_asset_specs from app.assets.services.snapshot_hash import snapshot_hash @@ -24,7 +23,6 @@ def _spec(path: Path) -> SeedAssetSpec: "tags": ["input"], "fname": path.name, "metadata": None, - "hash": None, "mime_type": None, "job_id": None, }