"""
``animedex api`` Click group + shared helpers.
The group itself lives here. Each backend subcommand is registered
from its own sibling module (``animedex/entry/api/<backend>.py``)
which imports and decorates onto :data:`api_group`. Keeping each
subcommand in its own file keeps any one file readable and lets
contributors edit one backend's CLI without touching the others.
Shared utilities used by every subcommand:
* :func:`_default_cache` - lazy singleton
:class:`~animedex.cache.sqlite.SqliteCache` at the platform-default
path.
* :func:`_output_mode_from_flags` - mutual-exclusion check on
``-i / -I / --debug``.
* :func:`_render_output` - dispatch to the four renderers.
* :func:`_exit_code_for` - status-class-aware exit code.
* :func:`_emit` - print + ctx.exit one-shot.
* :func:`_parse_extra_headers` - turn ``-H "Name: Value"`` into a dict.
* :func:`_common_request_options`, :func:`_common_output_options` -
decorator factories.
"""
import click
from animedex.api._envelope import RawResponse
from animedex.api._params import split_path_query
from animedex.models.common import ApiError
from animedex.render.raw import render_body, render_debug, render_head, render_include
_DEFAULT_CACHE = None
def _close_default_cache() -> None:
"""Close the lazy default cache singleton if it was constructed.
Registered via :func:`atexit` on first use of
:func:`_default_cache` so the SQLite connection releases cleanly
on interpreter shutdown (). Idempotent: a second call
after teardown is a no-op.
"""
global _DEFAULT_CACHE
if _DEFAULT_CACHE is not None:
try:
_DEFAULT_CACHE.close()
except Exception:
# Defensive: a connection that already failed to close
# should not surface a traceback during interpreter
# teardown. The file handle gets released by the OS
# regardless.
pass
_DEFAULT_CACHE = None
def _default_cache():
"""Lazy singleton :class:`SqliteCache` at the default platform path.
Created on first use so paths that pass ``--no-cache`` everywhere
or that run in unit tests never instantiate it.
The first construction registers :func:`_close_default_cache` as
an :func:`atexit` hook (). Without it a long-running
embedding (Jupyter, MCP server, future REPL) would leak the file
handle.
:return: A reusable ``SqliteCache`` instance.
:rtype: SqliteCache
"""
global _DEFAULT_CACHE
if _DEFAULT_CACHE is None:
import atexit
from animedex.cache.sqlite import SqliteCache
_DEFAULT_CACHE = SqliteCache()
atexit.register(_close_default_cache)
return _DEFAULT_CACHE
def _output_mode_from_flags(include_flag: bool, head_flag: bool, debug_flag: bool) -> str:
"""Pick the output mode from the three mutually-exclusive flags.
:raises click.UsageError: When more than one of the three flags
is set.
"""
selected = [name for name, on in (("-i", include_flag), ("-I", head_flag), ("--debug", debug_flag)) if on]
if len(selected) > 1:
raise click.UsageError(f"output flags are mutually exclusive: {' '.join(selected)}")
if include_flag:
return "include"
if head_flag:
return "head"
if debug_flag:
return "debug"
return "body"
def _render_output(envelope: RawResponse, mode: str, full_body: bool) -> str:
"""Dispatch to the four renderers in :mod:`animedex.render.raw`."""
if mode == "include":
return render_include(envelope)
if mode == "head":
return render_head(envelope)
if mode == "debug":
return render_debug(envelope, full_body=full_body)
return render_body(envelope)
def _exit_code_for(envelope: RawResponse) -> int:
"""Map an envelope to a CLI exit code.
* ``firewall_rejected`` -> 2 (legacy envelope state)
* 2xx -> 0
* 3xx (only with ``--no-follow``) -> 3
* 4xx -> 4
* 5xx -> 5
* other -> 1
"""
if envelope.firewall_rejected is not None:
return 2
s = envelope.status
if 200 <= s < 300:
return 0
if 300 <= s < 400:
return 3
if 400 <= s < 500:
return 4
if 500 <= s < 600:
return 5
return 1
def _common_output_options(func):
"""Decorator factory: add ``-i / -I / --debug / --no-follow / --debug-full-body``."""
func = click.option(
"--debug-full-body",
"debug_full_body",
is_flag=True,
default=False,
help="With --debug, do not truncate the body at 64 KiB.",
)(func)
func = click.option(
"--no-follow", "no_follow", is_flag=True, default=False, help="Do not auto-follow 3xx redirects."
)(func)
func = click.option(
"--debug",
"debug_flag",
is_flag=True,
default=False,
help="Emit the full RawResponse envelope as JSON (data + debug).",
)(func)
func = click.option(
"-I", "--head", "head_flag", is_flag=True, default=False, help="Print status line + response headers; no body."
)(func)
func = click.option(
"-i",
"--include",
"include_flag",
is_flag=True,
default=False,
help="Print status line + response headers + body (curl-style).",
)(func)
return func
def _common_request_options(func):
"""Decorator factory: add universal raw request options."""
func = click.option(
"--max-items",
type=int,
default=None,
help="With --paginate, stop after at most N accumulated items.",
)(func)
func = click.option(
"--max-pages",
type=int,
default=10,
show_default=True,
help="With --paginate, stop after at most N pages.",
)(func)
func = click.option("--paginate", is_flag=True, default=False, help="Auto-paginate supported GET endpoints.")(func)
func = click.option(
"--raw-field",
"-F",
"api_fields",
multiple=True,
expose_kind="raw",
cls=ApiFieldOption,
help="Add a string field (repeatable): K=V.",
)(func)
func = click.option(
"--field",
"-f",
"api_fields",
multiple=True,
expose_kind="typed",
cls=ApiFieldOption,
help="Add a typed field (repeatable): K=V.",
)(func)
func = click.option("--method", "-X", default="GET", help="HTTP method for raw REST calls.")(func)
func = click.option(
"--no-cache", is_flag=True, default=False, help="Skip cache lookup and write for cache-eligible requests."
)(func)
func = click.option("--cache", "cache_ttl", type=int, default=None, help="Override cache TTL in seconds.")(func)
func = click.option(
"--rate",
type=click.Choice(["normal", "slow"]),
default="normal",
help="Voluntary slowdown: 'slow' halves the rate-limit refill rate.",
)(func)
func = click.option(
"--header", "-H", "extra_headers", multiple=True, help="Extra header (repeatable): 'Name: Value'."
)(func)
return func
[docs]
class ApiFieldOption(click.Option):
"""Click option that preserves mixed ``-f`` / ``-F`` order.
Click normally collects values per option declaration, which loses
order when two flags write to the same parameter. The raw
passthrough needs gh-style last-write-wins across both flags, so
each parsed value carries its kind before landing in ``api_fields``.
"""
[docs]
def __init__(self, *args, expose_kind: str, **kwargs):
self.expose_kind = expose_kind
super().__init__(*args, **kwargs)
[docs]
def add_to_parser(self, parser, ctx) -> None:
def _wrap(opts, explicit_value, state):
value = explicit_value
if value is None:
option = parser._long_opt.get(opts[0]) or parser._short_opt.get(opts[0])
value = parser._get_value_from_state(opts[0], option, state)
state.opts.setdefault(self.name, []).append((self.expose_kind, value))
state.order.append(self)
for opt in self.opts:
norm = _normalize_option(opt, ctx)
option = _ApiFieldParserOption(obj=self, opts=[norm], process_value=_wrap)
if norm.startswith("--"):
parser._long_opt[norm] = option
else:
parser._short_opt[norm] = option
parser._opt_prefixes.update(_option_prefixes(norm))
[docs]
def handle_parse_result(self, ctx, opts, args):
if self.expose_kind != "typed" and self.name in ctx.params:
return None, args
value, args = super().handle_parse_result(ctx, opts, args)
if self.expose_kind == "typed" and self.name in ctx.params:
ctx.params[self.name] = value
return value, args
[docs]
def type_cast_value(self, ctx, value):
if value is None:
return ()
return tuple(value)
class _ApiFieldParserOption:
def __init__(self, *, obj, opts, process_value):
self.obj = obj
self.opts = opts
self.dest = obj.name
self.action = "append"
self.nargs = 1
self.const = None
self.takes_value = True
self._short_opts = [opt for opt in opts if opt.startswith("-") and not opt.startswith("--")]
self._long_opts = [opt for opt in opts if opt.startswith("--")]
self.prefixes = _option_prefixes(opts[0])
self._process_value = process_value
def process(self, value, state):
self._process_value(self.opts, value, state)
def _normalize_option(opt, ctx):
"""Apply Click's token normalizer to an option string."""
token_normalize_func = getattr(ctx, "token_normalize_func", None)
if token_normalize_func is None:
return opt
prefix = ""
rest = opt
while rest.startswith("-"):
prefix += "-"
rest = rest[1:]
return f"{prefix}{token_normalize_func(rest)}"
def _option_prefixes(opt: str) -> set:
"""Return option prefixes for Click's parser bookkeeping."""
if opt.startswith("--"):
return {"--"}
if opt.startswith("-"):
return {"-"}
return set()
def _parse_extra_headers(extra_headers):
"""Turn ``("Name: Value", ...)`` into a dict; raise on malformed."""
out = {}
for h in extra_headers or []:
if ":" not in h:
raise click.UsageError(f"--header must be 'Name: Value', got {h!r}")
name, value = h.split(":", 1)
out[name.strip()] = value.strip()
return out
def _coerce_field_value(value: str):
"""Coerce one ``-f`` value using gh-compatible raw API rules."""
try:
return int(value)
except ValueError:
pass
try:
return float(value)
except ValueError:
pass
if value == "true":
return True
if value == "false":
return False
return value
def _parse_api_fields(api_fields):
"""Parse mixed ``-f`` and ``-F`` values into a last-write mapping."""
out = {}
for kind, item in api_fields or []:
if "=" not in item:
raise click.UsageError(f"--field/--raw-field must be K=V, got {item!r}")
key, value = item.split("=", 1)
if not key:
raise click.UsageError("--field/--raw-field key must not be empty")
out[key] = value if kind == "raw" else _coerce_field_value(value)
return out
def _merge_path_and_fields(path: str, fields: dict):
"""Merge ``-f``/``-F`` values over query parameters in ``path``."""
if not fields:
return path, None
return split_path_query(path, fields)
def _merge_json_objects(left, right, *, left_name: str, right_name: str):
"""Merge two optional JSON objects for CLI request-body assembly."""
if left is None:
base = {}
elif isinstance(left, dict):
base = dict(left)
else:
raise click.UsageError(f"{left_name} must decode to a JSON object")
if right is None:
return base
if not isinstance(right, dict):
raise click.UsageError(f"{right_name} must decode to a JSON object")
base.update(right)
return base
def _call_or_paginate(
backend_module,
*,
backend: str,
paginate: bool,
max_pages: int,
max_items,
method_explicit: bool = True,
**kwargs,
):
"""Call one raw request or the central pagination helper."""
if not paginate:
return backend_module.call(**kwargs)
method = kwargs.get("method", "GET").upper()
from animedex.api._paginate import get_strategy
try:
get_strategy(backend)
except ApiError:
if method != "GET" and method_explicit:
_warn_paginate_ignored_non_get(method, explicit=True)
return backend_module.call(**kwargs)
_warn_paginate_ignored_no_strategy(backend)
return backend_module.call(**kwargs)
if method != "GET":
_warn_paginate_ignored_non_get(method, explicit=method_explicit)
return backend_module.call(**kwargs)
paginated_kwargs = {k: v for k, v in kwargs.items() if k not in ("json_body", "raw_body")}
try:
from animedex.api._paginate import call_paginated
return call_paginated(
backend=backend,
max_pages=max_pages,
max_items=max_items,
**paginated_kwargs,
)
except ApiError as exc:
raise click.ClickException(str(exc)) from exc
def _warn_paginate_ignored_non_get(method: str, *, explicit: bool) -> None:
"""Warn that ``--paginate`` cannot apply to a non-GET request."""
reason = f"explicit -X {method}" if explicit else f"non-GET {method}"
click.echo(f"--paginate ignored: {reason}; sending a single forwarded request.", err=True)
def _warn_paginate_ignored_no_strategy(backend: str) -> None:
"""Warn that ``--paginate`` has no backend strategy to run."""
click.echo(
f"--paginate ignored: backend {backend!r} has no pagination strategy; sending a single forwarded request.",
err=True,
)
def _emit(ctx: click.Context, envelope: RawResponse, mode: str, full_body: bool) -> None:
"""Print the rendered output and exit with the status-class code."""
click.echo(_render_output(envelope, mode, full_body))
ctx.exit(_exit_code_for(envelope))
def _resolve_cache(no_cache: bool):
"""Pick the cache to pass into :func:`animedex.api._dispatch.call`.
:param no_cache: When ``True``, the call should bypass the cache.
:type no_cache: bool
:return: ``None`` when ``no_cache`` is set; otherwise the lazy
singleton from :func:`_default_cache`.
"""
if no_cache:
return None
return _default_cache()
@click.group(name="api")
def api_group() -> None:
"""Raw HTTP / GraphQL passthrough to one of the 12 upstream backends.
Each subcommand wraps one backend's raw API surface. The
dispatcher injects the project ``User-Agent``, applies a
per-backend rate-limit token bucket, and consults the local
SQLite cache before issuing the request.
\b
Output modes (mutually exclusive):
(default) response body only (gh-api equivalent)
-i, --include status line + response headers + body
-I, --head status line + response headers (no body)
--debug full RawResponse envelope as indented JSON
(request snapshot, redirect chain, timing,
cache provenance; credentials redacted)
\b
Other behaviour:
-X, --method METHOD set the HTTP method on raw requests
-f, --field K=V add a typed field (repeatable)
-F, --raw-field K=V add a string field (repeatable)
--paginate auto-paginate supported GET endpoints
--max-pages N page ceiling for --paginate (default 10)
--max-items N item ceiling for --paginate
--no-follow disable 3xx auto-following
--debug-full-body opt out of the 64 KiB body cap in --debug
--no-cache skip cache lookup/write for eligible requests
--cache TTL_SECONDS override default cache TTL
--rate {normal,slow} voluntary slowdown (slow halves refill)
-H, --header K:V add request header (repeatable)
A caller-supplied ``User-Agent`` via ``--header`` overrides the
project default verbatim.
\b
Examples:
animedex api jikan /anime/52991
animedex api anilist '{ Media(id:154587){ title{romaji} } }'
animedex api kitsu '/anime?filter[text]=Frieren&page[limit]=2' -i
animedex api shikimori /api/graphql --graphql '{ animes(ids:"52991"){ id name }}'
animedex api jikan /anime --paginate --field q=Naruto --field limit=2 --max-pages 3
animedex api jikan /anime/52991 --method DELETE
animedex api jikan /anime/52991 --debug | jq '{cache, timing}'
\b
Per-backend docs (each subcommand's --help has more detail):
anilist https://docs.anilist.co/
jikan https://docs.api.jikan.moe/
kitsu https://kitsu.docs.apiary.io/
mangadex https://api.mangadex.org/docs/
trace https://soruly.github.io/trace.moe-api/
danbooru https://danbooru.donmai.us/wiki_pages/help:api
shikimori https://shikimori.io/api/doc
ann https://www.animenewsnetwork.com/encyclopedia/api.php
ghibli https://ghibliapi.vercel.app/
nekos https://docs.nekos.best/
quote https://animechan.io/docs
waifu https://docs.waifu.im/
\f
Backend: animedex (local; routes to one of 12 upstream backends).
Rate limit: not applicable at this level (each backend's bucket
applies inside the call).
--- LLM Agent Guidance ---
The api group is the project's escape hatch for endpoints not
covered by the higher-level commands. Each subcommand wraps one
backend's raw HTTP/GraphQL surface; the dispatcher injects the
project User-Agent, applies rate limiting, and consults the local
cache. Method/path choices are forwarded verbatim; unsupported
--paginate combinations fall back to a single raw request rather
than blocking the call. The caller owns the upstream result. The output flags
(-i / -I / --debug) are shared and mutually exclusive. Use
--debug when you need to inspect the full envelope (redirect
chain, timing, cache provenance, fingerprint-redacted request
headers) - this is the "data + debug" mode.
Caller-supplied User-Agent via --header overrides the project
default verbatim.
--- End ---
"""
# Importing the subcommand modules registers them onto the api_group
# at import time. Order does not matter functionally; alphabetical for
# diff-friendliness.
from animedex.entry.api import anilist # noqa: E402, F401
from animedex.entry.api import ann # noqa: E402, F401
from animedex.entry.api import danbooru # noqa: E402, F401
from animedex.entry.api import ghibli # noqa: E402, F401
from animedex.entry.api import jikan # noqa: E402, F401
from animedex.entry.api import kitsu # noqa: E402, F401
from animedex.entry.api import mangadex # noqa: E402, F401
from animedex.entry.api import nekos # noqa: E402, F401
from animedex.entry.api import quote # noqa: E402, F401
from animedex.entry.api import shikimori # noqa: E402, F401
from animedex.entry.api import trace # noqa: E402, F401
from animedex.entry.api import waifu # noqa: E402, F401
__all__ = ["api_group"]