Make console parameter optional in search_civitai

- Console is now optional for headless/server usage
- Progress spinner only shown when console is provided
- Error messages print only when console is provided

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Adam Ladachowski
2026-02-15 19:59:55 +01:00
parent fb27f8a9b0
commit 87ab5c44bb
+21 -11
View File
@@ -233,7 +233,7 @@ def search_civitai(
sort: SortOrder, sort: SortOrder,
limit: int, limit: int,
api_key: str | None, api_key: str | None,
console: Console, console: Console | None = None,
*, *,
period: Period | None = None, period: Period | None = None,
nsfw: NsfwLevel | bool | None = None, nsfw: NsfwLevel | bool | None = None,
@@ -249,6 +249,9 @@ def search_civitai(
Implements workarounds for API limitations: Implements workarounds for API limitations:
- Query + filters: fetches more results and filters client-side - Query + filters: fetches more results and filters client-side
- NSFW: defaults to including all content (like website behavior) - NSFW: defaults to including all content (like website behavior)
Args:
console: Rich console for progress display. If None, runs without progress spinner.
""" """
params, has_filters = _build_search_params( params, has_filters = _build_search_params(
query, query,
@@ -267,26 +270,33 @@ def search_civitai(
) )
url = f"{CIVITAI_API_BASE}/models" url = f"{CIVITAI_API_BASE}/models"
with Progress( def _do_search() -> dict[str, Any] | None:
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=console,
transient=True,
) as progress:
progress.add_task("[cyan]Searching CivitAI...", total=None)
try: try:
response = httpx.get(url, params=params, headers=_get_headers(api_key), timeout=30.0) response = httpx.get(url, params=params, headers=_get_headers(api_key), timeout=30.0)
response.raise_for_status() response.raise_for_status()
result: dict[str, Any] = response.json() result: dict[str, Any] = response.json()
return _filter_results(result, query, has_filters, limit) return _filter_results(result, query, has_filters, limit)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
console.print(f"[red]API error: {e.response.status_code}[/red]") if console:
console.print(f"[red]API error: {e.response.status_code}[/red]")
return None return None
except httpx.RequestError as e: except httpx.RequestError as e:
console.print(f"[red]Request error: {e}[/red]") if console:
console.print(f"[red]Request error: {e}[/red]")
return None return None
if console:
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=console,
transient=True,
) as progress:
progress.add_task("[cyan]Searching CivitAI...", total=None)
return _do_search()
else:
return _do_search()
def _setup_resume(dest_path: Path, resume: bool, console: Console) -> tuple[dict[str, str], str, int]: def _setup_resume(dest_path: Path, resume: bool, console: Console) -> tuple[dict[str, str], str, int]:
"""Set up resume headers and mode for download.""" """Set up resume headers and mode for download."""