kicksdk

 1"""Kick (kick.com) without the I/O.
 2
 3kicksdk knows Kick - which requests to make, what the answers look like, how
 4the realtime chat feed speaks, how webhooks are signed - and never sends a
 5request or opens a socket itself. Send the :class:`Request` objects with any
 6HTTP client and hand the status and body back to the matching ``parse_*``.
 7For a ready chat connection over asyncio, see :mod:`kicksdk.aio`.
 8
 9Values come back Kick-shaped: dicts with snake_case keys, as Kick writes them.
10
11Modules: :mod:`kicksdk.events`, :mod:`kicksdk.content`,
12:mod:`kicksdk.realtime`, :mod:`kicksdk.public_api`, :mod:`kicksdk.website`,
13:mod:`kicksdk.webhooks`.
14"""
15
16from __future__ import annotations
17
18from dataclasses import dataclass
19from typing import Any, Mapping, Optional
20
21from . import _native
22
23__all__ = ["KickError", "Request", "events", "content", "realtime", "public_api", "website", "webhooks"]
24
25
26class KickError(Exception):
27    """Kick answered with an error, or a call could not be made.
28
29    ``kind`` is ``status`` (Kick's HTTP status is in ``status``), ``shape``,
30    ``json``, ``invalid``, ``webhook`` or ``call``.
31    """
32
33    def __init__(self, kind: str, status: Optional[int], message: str) -> None:
34        super().__init__(message)
35        self.kind = kind
36        self.status = status
37        self.message = message
38
39    def __repr__(self) -> str:
40        return f"KickError({self.kind!r}, {self.status!r}, {self.message!r})"
41
42
43@dataclass(frozen=True)
44class Request:
45    """A request to send with any HTTP client."""
46
47    method: str
48    url: str
49    headers: list[tuple[str, str]]
50    body: Optional[str]
51
52    @classmethod
53    def _from(cls, v: Mapping[str, Any]) -> "Request":
54        return cls(v["method"], v["url"], [(k, val) for k, val in v["headers"]], v.get("body"))
55
56
57def _call(_fn: str, /, endpoints: Optional[Mapping[str, str]] = None, **args: Any) -> Any:
58    # Positional-only, so a function's own `name` argument (events.parse) is
59    # never taken for the function's name.
60    if endpoints:
61        args["endpoints"] = dict(endpoints)
62    try:
63        return _native.call(_fn, args)
64    except _native.NativeError as e:
65        kind, status, message = e.args
66        raise KickError(kind, status, message) from None
67
68
69def _request(_fn: str, /, **kw: Any) -> Request:
70    return Request._from(_call(_fn, **kw))
71
72
73def _requests(_fn: str, /, **kw: Any) -> list[Request]:
74    return [Request._from(r) for r in _call(_fn, **kw)]
75
76
77from . import content, events, public_api, realtime, webhooks, website  # noqa: E402