refactor: move in-function imports to top-level and remove keyword-only argument pattern

- Move imports from inside functions to module top-level in:
  - app/assets/database/queries/asset.py
  - app/assets/database/queries/asset_info.py
  - app/assets/database/queries/cache_state.py
  - app/assets/manager.py
  - app/assets/services/asset_management.py
  - app/assets/services/ingest.py

- Remove keyword-only argument markers (*,) from app/assets/ to match codebase conventions

Amp-Thread-ID: https://ampcode.com/threads/T-019c24eb-bfa2-727f-8212-8bc976048604
Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
Luke Mino-Altherr
2026-02-03 11:18:28 -08:00
parent 7dd0cac0c9
commit 77be514cd9
10 changed files with 12 additions and 56 deletions

View File

@@ -11,6 +11,7 @@ import contextlib
import os
from typing import Sequence
from app.assets.database.models import Asset
from app.database.db import create_session
from app.assets.helpers import (
compute_relative_filename,
@@ -31,7 +32,6 @@ from app.assets.database.queries import (
def get_asset_detail(
*,
asset_info_id: str,
owner_id: str = "",
) -> dict | None:
@@ -57,7 +57,6 @@ def get_asset_detail(
def update_asset_metadata(
*,
asset_info_id: str,
name: str | None = None,
tags: Sequence[str] | None = None,
@@ -136,7 +135,6 @@ def update_asset_metadata(
def delete_asset_reference(
*,
asset_info_id: str,
owner_id: str,
delete_content_if_orphan: bool = True,
@@ -168,7 +166,6 @@ def delete_asset_reference(
states = list_cache_states_by_asset_id(session, asset_id=asset_id)
file_paths = [s.file_path for s in (states or []) if getattr(s, "file_path", None)]
from app.assets.database.models import Asset
asset_row = session.get(Asset, asset_id)
if asset_row is not None:
session.delete(asset_row)
@@ -185,7 +182,6 @@ def delete_asset_reference(
def set_asset_preview(
*,
asset_info_id: str,
preview_asset_id: str | None = None,
owner_id: str = "",

View File

@@ -9,6 +9,9 @@ import logging
import os
from typing import Sequence
from sqlalchemy import select
from app.assets.database.models import Asset, Tag
from app.database.db import create_session
from app.assets.helpers import (
compute_relative_filename,
@@ -33,7 +36,6 @@ from app.assets.database.queries import (
def ingest_file_from_path(
*,
abs_path: str,
asset_hash: str,
size_bytes: int,
@@ -67,7 +69,6 @@ def ingest_file_from_path(
with create_session() as session:
# Validate preview_id if provided
if preview_id:
from app.assets.database.models import Asset
if not session.get(Asset, preview_id):
preview_id = None
@@ -141,7 +142,6 @@ def ingest_file_from_path(
def register_existing_asset(
*,
asset_hash: str,
name: str,
user_metadata: dict | None = None,
@@ -212,8 +212,6 @@ def register_existing_asset(
def _validate_tags_exist(session, tags: list[str]) -> None:
"""Raise ValueError if any tags don't exist."""
from sqlalchemy import select
from app.assets.database.models import Tag
existing_tag_names = set(
name for (name,) in session.execute(select(Tag.name).where(Tag.name.in_(tags))).all()
)
@@ -230,7 +228,6 @@ def _compute_filename_for_asset(session, asset_id: str) -> str | None:
def _update_metadata_with_filename(
session,
*,
asset_info_id: str,
asset_id: str,
info,

View File

@@ -16,7 +16,6 @@ from app.assets.database.queries import (
def apply_tags(
*,
asset_info_id: str,
tags: list[str],
origin: str = "manual",
@@ -47,7 +46,6 @@ def apply_tags(
def remove_tags(
*,
asset_info_id: str,
tags: list[str],
owner_id: str = "",