Module genshin.client.components.transaction

Transaction client.

Classes

class TransactionClient (cookies: Union[ForwardRef('http.cookies.BaseCookie[Any]'), Mapping[Any, Any], str, Sequence[Union[ForwardRef('http.cookies.BaseCookie[Any]'), Mapping[Any, Any], str]], ForwardRef(None)] = None, *, authkey: Optional[str] = None, lang: str = 'en-us', region: Region = Region.OVERSEAS, proxy: Optional[str] = None, game: Optional[Game] = None, uid: Optional[int] = None, hoyolab_id: Optional[int] = None, device_id: Optional[str] = None, device_fp: Optional[str] = None, headers: Union[Mapping[str, str], Mapping[multidict._multidict.istr, str], multidict._multidict.CIMultiDict, multidict._multidict.CIMultiDictProxy, Iterable[Tuple[Union[str, multidict._multidict.istr], str]], ForwardRef(None)] = None, cache: Optional[BaseCache] = None, debug: bool = False)

Transaction component.

Expand source code 
class TransactionClient(base.BaseClient):
    """Transaction component."""

    async def request_transaction(
        self,
        endpoint: str,
        *,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        authkey: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the transaction log endpoint."""
        params = dict(params or {})
        authkey = authkey or self.authkey

        if authkey is None:
            raise RuntimeError("No authkey provided")

        base_url = routes.YSULOG_URL.get_url(self.region)
        url = base_url / endpoint

        params["authkey_ver"] = 1
        params["sign_type"] = 2
        params["authkey"] = urllib.parse.unquote(authkey)
        params["lang"] = utility.create_short_lang_code(lang or self.lang)

        return await self.request(url, method=method, params=params, **kwargs)

    async def _get_transaction_page(
        self,
        end_id: int,
        kind: str,
        *,
        lang: typing.Optional[str] = None,
        authkey: typing.Optional[str] = None,
    ) -> typing.Sequence[models.BaseTransaction]:
        """Get a single page of transactions."""
        kind = models.TransactionKind(kind)
        endpoint = "Get" + kind.value.capitalize() + "Log"

        data = await self.request_transaction(
            endpoint,
            lang=lang,
            authkey=authkey,
            params=dict(end_id=end_id, size=20),
        )

        transactions: list[models.BaseTransaction] = []
        for trans in data["list"]:
            model = models.ItemTransaction if "name" in trans else models.Transaction
            model = typing.cast("type[models.BaseTransaction]", model)
            transactions.append(model(**trans, kind=kind))

        return transactions

    def transaction_log(
        self,
        kind: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
        *,
        limit: typing.Optional[int] = None,
        lang: typing.Optional[str] = None,
        authkey: typing.Optional[str] = None,
        end_id: int = 0,
    ) -> paginators.Paginator[models.BaseTransaction]:
        """Get the transaction log of a user."""
        kinds = kind or ["primogem", "crystal", "resin", "artifact", "weapon"]

        if isinstance(kinds, str):
            kinds = [kinds]

        iterators: list[paginators.Paginator[models.BaseTransaction]] = []
        for kind in kinds:
            iterators.append(
                paginators.CursorPaginator(
                    functools.partial(
                        self._get_transaction_page,
                        kind=kind,
                        lang=lang,
                        authkey=authkey,
                    ),
                    limit=limit,
                    end_id=end_id,
                )
            )

        if len(iterators) == 1:
            return iterators[0]

        return paginators.MergedPaginator(iterators, key=lambda trans: trans.time.timestamp())

Ancestors

Subclasses

Class variables

var logger : logging.Logger

Instance variables

var authkeys : dict[Game, str]
var cache : BaseCache
var cookie_manager : BaseCookieManager
var custom_headers : multidict._multidict.CIMultiDict[str]
var uids : dict[Game, int]

Methods

async def request_transaction(self, endpoint: str, *, method: str = 'GET', lang: Optional[str] = None, authkey: Optional[str] = None, params: Optional[Mapping[str, Any]] = None, **kwargs: Any) ‑> Mapping[str, Any]

Make a request towards the transaction log endpoint.

def transaction_log(self, kind: Union[str, Sequence[str], ForwardRef(None)] = None, *, limit: Optional[int] = None, lang: Optional[str] = None, authkey: Optional[str] = None, end_id: int = 0) ‑> Paginator[BaseTransaction]

Get the transaction log of a user.

Inherited members