"""High-level Waifu.im Python API.
Three thin wrappers over the JSON read endpoints on
``api.waifu.im``: :func:`tags`, :func:`images`, :func:`artists`.
NSFW posture mirrors the upstream: the ``/images`` endpoint defaults
to SFW only when ``isNsfw`` is omitted; pass ``is_nsfw=True`` from
Python (``--is-nsfw true`` from the CLI) to opt in to NSFW results.
The flag is a transparent passthrough to the upstream's ``isNsfw``
query parameter, not a paternalistic confirmation gate.
"""
from __future__ import annotations
import json as _json
import os as _os
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from animedex.api import waifu as _raw_waifu
from animedex.backends.waifu.models import WaifuArtist, WaifuImage, WaifuStats, WaifuTag, WaifuUser
from animedex.config import Config
from animedex.models.common import ApiError, SourceTag
# ---------- internals ----------
def _src(envelope) -> SourceTag:
return SourceTag(
backend="waifu",
fetched_at=datetime.now(timezone.utc),
cached=envelope.cache.hit,
rate_limited=envelope.timing.rate_limit_wait_ms > 0,
)
def _fetch(path: str, *, params: Optional[Dict[str, Any]] = None, config: Optional[Config] = None, **kw):
"""Issue a Waifu.im GET, parse the body, validate the envelope.
:return: ``(parsed_payload_dict, source_tag)``.
:raises ApiError: ``not-found`` for 404, ``upstream-error`` for
5xx, ``upstream-decode`` for non-text or
non-JSON bodies.
"""
raw = _raw_waifu.call(path=path, params=params, config=config, **kw)
if raw.firewall_rejected is not None: # pragma: no cover - defensive
raise ApiError(
raw.firewall_rejected.get("message", "request blocked"),
backend="waifu",
reason=raw.firewall_rejected.get("reason", "firewall"),
)
if raw.body_text is None: # pragma: no cover - waifu returns JSON
raise ApiError("waifu returned a non-text body", backend="waifu", reason="upstream-decode")
if raw.status == 401 or raw.status == 403:
raise ApiError(
f"waifu {raw.status} on {path} (rejected credentials)",
backend="waifu",
reason="auth-required",
)
if raw.status == 404:
raise ApiError(f"waifu 404 on {path}", backend="waifu", reason="not-found")
if raw.status >= 500:
raise ApiError(f"waifu {raw.status} on {path}", backend="waifu", reason="upstream-error")
try:
payload = _json.loads(raw.body_text)
except ValueError as exc:
raise ApiError(f"waifu returned non-JSON body: {exc}", backend="waifu", reason="upstream-decode") from exc
return payload, _src(raw)
def _items(payload: dict) -> List[dict]:
"""Pull the ``items`` list out of a paginated envelope."""
if not isinstance(payload, dict) or "items" not in payload:
raise ApiError("waifu response missing 'items' key", backend="waifu", reason="upstream-shape")
items = payload.get("items") or []
if not isinstance(items, list):
raise ApiError("waifu 'items' is not a list", backend="waifu", reason="upstream-shape")
return items
# ---------- /tags ----------
# ---------- /artists ----------
[docs]
def artists(
*,
page_number: Optional[int] = None,
page_size: Optional[int] = None,
config: Optional[Config] = None,
**kw,
) -> List[WaifuArtist]:
"""List every artist via ``/artists``.
:param page_number: 1-indexed page number; ``None`` returns
page 1 (the upstream default).
:type page_number: int or None
:param page_size: ``pageSize`` override.
:type page_size: int or None
:return: List of typed artists.
:rtype: list[WaifuArtist]
"""
params: Dict[str, Any] = {}
if page_number is not None:
params["pageNumber"] = page_number
if page_size is not None:
params["pageSize"] = page_size
payload, src = _fetch("/artists", params=params, config=config, **kw)
return [WaifuArtist.model_validate({**row, "source_tag": src}) for row in _items(payload)]
# ---------- /images ----------
[docs]
def images(
*,
included_tags: Optional[List[str]] = None,
excluded_tags: Optional[List[str]] = None,
is_nsfw: Optional[bool] = None,
is_animated: Optional[bool] = None,
page_number: Optional[int] = None,
page_size: Optional[int] = None,
config: Optional[Config] = None,
**kw,
) -> List[WaifuImage]:
"""List image records via ``/images`` with optional filters.
NSFW posture: when ``is_nsfw`` is ``None`` (the default), the
parameter is omitted from the request and the upstream's
SFW-only default applies. When ``is_nsfw=True``, NSFW images
are returned. When ``is_nsfw=False``, SFW images are returned
explicitly. The flag is a transparent passthrough; the project's
posture is to inform the user about the upstream's defaults,
not to gate.
:param included_tags: Tag slugs that **must** be present on
returned images (multiple → all-of).
:type included_tags: list of str or None
:param excluded_tags: Tag slugs that **must not** be present.
:type excluded_tags: list of str or None
:param is_nsfw: NSFW filter. ``None`` (default) honours the
upstream's SFW-only default; ``True`` → NSFW
only; ``False`` → SFW only (explicit).
:type is_nsfw: bool or None
:param is_animated: ``True`` → only animated assets; ``False``
→ only static; ``None`` → no filter.
:type is_animated: bool or None
:param page_number: 1-indexed page number.
:type page_number: int or None
:param page_size: Rows per page.
:type page_size: int or None
:return: List of typed images.
:rtype: list[WaifuImage]
"""
params: Dict[str, Any] = {}
if included_tags:
params["included_tags"] = list(included_tags)
if excluded_tags:
params["excluded_tags"] = list(excluded_tags)
if is_nsfw is not None:
params["isNsfw"] = "true" if is_nsfw else "false"
if is_animated is not None:
params["isAnimated"] = "true" if is_animated else "false"
if page_number is not None:
params["pageNumber"] = page_number
if page_size is not None:
params["pageSize"] = page_size
payload, src = _fetch("/images", params=params, config=config, **kw)
return [WaifuImage.model_validate({**row, "source_tag": src}) for row in _items(payload)]
[docs]
def tag(id: int, *, config: Optional[Config] = None, **kw) -> WaifuTag:
"""Fetch one tag by numeric ID via ``/tags/{id}``.
:param id: Numeric Waifu.im tag ID.
:type id: int
:return: Typed tag.
:rtype: WaifuTag
"""
payload, src = _fetch(f"/tags/{id}", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /tags/{id} did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuTag.model_validate({**payload, "source_tag": src})
[docs]
def tag_by_slug(slug: str, *, config: Optional[Config] = None, **kw) -> WaifuTag:
"""Fetch one tag by URL-safe slug via ``/tags/by-slug/{slug}``.
:param slug: Lowercased tag slug (e.g. ``"waifu"``).
:type slug: str
:return: Typed tag.
:rtype: WaifuTag
"""
payload, src = _fetch(f"/tags/by-slug/{slug}", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /tags/by-slug/{slug} did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuTag.model_validate({**payload, "source_tag": src})
[docs]
def artist(id: int, *, config: Optional[Config] = None, **kw) -> WaifuArtist:
"""Fetch one artist by numeric ID via ``/artists/{id}``.
:param id: Numeric Waifu.im artist ID.
:type id: int
:return: Typed artist.
:rtype: WaifuArtist
"""
payload, src = _fetch(f"/artists/{id}", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /artists/{id} did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuArtist.model_validate({**payload, "source_tag": src})
[docs]
def artist_by_name(name: str, *, config: Optional[Config] = None, **kw) -> WaifuArtist:
"""Fetch one artist by display name via ``/artists/by-name/{name}``.
The response is the same artist envelope as ``/artists/{id}`` but
additionally includes the artist's ``images`` list.
:param name: Artist display name (case-sensitive).
:type name: str
:return: Typed artist (with extra ``images`` field via
``extra='allow'``).
:rtype: WaifuArtist
"""
payload, src = _fetch(f"/artists/by-name/{name}", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /artists/by-name/{name} did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuArtist.model_validate({**payload, "source_tag": src})
[docs]
def image(id: int, *, config: Optional[Config] = None, **kw) -> WaifuImage:
"""Fetch one image by numeric ID via ``/images/{id}``.
:param id: Numeric Waifu.im image ID.
:type id: int
:return: Typed image.
:rtype: WaifuImage
"""
payload, src = _fetch(f"/images/{id}", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /images/{id} did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuImage.model_validate({**payload, "source_tag": src})
[docs]
def stats_public(*, config: Optional[Config] = None, **kw) -> WaifuStats:
"""Fetch the public statistics envelope via ``/stats/public``.
Returns a small object summarising the catalogue size + lifetime
request volume; useful as a cheap upstream-liveness probe.
:return: Typed statistics envelope.
:rtype: WaifuStats
"""
payload, src = _fetch("/stats/public", config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /stats/public did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuStats.model_validate({**payload, "source_tag": src})
# ---------- authenticated reads ----------
def _resolve_api_key(token: Optional[str] = None, *, config: Optional[Config] = None) -> str:
"""Locate the Waifu.im API key.
Resolution order: explicit ``token=`` → ``ANIMEDEX_WAIFU_TOKEN``
env var → :class:`~animedex.auth.store.TokenStore` entry under
``"waifu"``.
:raises ApiError: ``auth-required`` when no key resolves.
"""
if token:
return token
env = _os.environ.get("ANIMEDEX_WAIFU_TOKEN")
if env:
return env
if config is not None:
stored = config.effective_token_store().get("waifu")
if stored:
return stored
raise ApiError(
"waifu auth required: pass token=, set ANIMEDEX_WAIFU_TOKEN, or "
"store the API key under 'waifu' in the token store",
backend="waifu",
reason="auth-required",
)
def _authed_fetch(
path: str,
*,
params: Optional[Dict[str, Any]] = None,
token: Optional[str] = None,
config: Optional[Config] = None,
**kw,
):
"""``_fetch()`` variant that injects ``X-Api-Key`` (Waifu.im's
own auth scheme; not Bearer)."""
key = _resolve_api_key(token, config=config)
headers = dict(kw.pop("headers", None) or {})
headers["X-Api-Key"] = key
return _fetch(path, params=params, config=config, headers=headers, **kw)
[docs]
def me(*, token: Optional[str] = None, config: Optional[Config] = None, **kw) -> WaifuUser:
"""Authenticated current user via ``/users/me``.
:return: Typed user record (id, Discord identity, role,
request counters).
:rtype: WaifuUser
"""
payload, src = _authed_fetch("/users/me", token=token, config=config, **kw)
if not isinstance(payload, dict):
raise ApiError(
"waifu /users/me did not return a single object",
backend="waifu",
reason="upstream-shape",
)
return WaifuUser.model_validate({**payload, "source_tag": src})
[docs]
def selftest() -> bool:
"""Smoke-test the public Waifu.im Python API (signatures only,
no network).
:return: ``True`` on success.
:rtype: bool
"""
import inspect
public_callables = [
tags,
tag,
tag_by_slug,
artists,
artist,
artist_by_name,
images,
image,
stats_public,
me,
]
for fn in public_callables:
sig = inspect.signature(fn)
assert "config" in sig.parameters, f"{fn.__name__} missing config kwarg"
return True