aiobungie.rest

Basic implementation of a RESTful clients for Bungie's REST API.

View Source

   1# MIT License
   2#
   3# Copyright (c) 2020 = Present nxtlo
   4#
   5# Permission is hereby granted, free of charge, to typing.Any person obtaining a copy
   6# of this software and associated documentation files (the "Software"), to deal
   7# in the Software without restriction, including without limitation the rights
   8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9# copies of the Software, and to permit persons to whom the Software is
  10# furnished to do so, subject to the following conditions:
  11#
  12# The above copyright notice and this permission notice shall be included in all
  13# copies or substantial portions of the Software.
  14#
  15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF typing.Any KIND, EXPRESS OR
  16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR typing.Any CLAIM, DAMAGES OR OTHER
  19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21# SOFTWARE.
  22
  23"""Basic implementation of a RESTful clients for Bungie's REST API."""
  24
  25from __future__ import annotations
  26
  27__all__ = ("RESTClient", "RESTPool", "TRACE")
  28
  29import asyncio
  30import contextlib
  31import datetime
  32import http
  33import logging
  34import os
  35import pathlib
  36import sys
  37import typing
  38import uuid
  39import zipfile
  40
  41import aiohttp
  42
  43from aiobungie import api, builders, error, metadata, typedefs, url
  44from aiobungie.crates import clans, fireteams
  45from aiobungie.internal import _backoff as backoff
  46from aiobungie.internal import enums, helpers, time
  47
  48if typing.TYPE_CHECKING:
  49    import collections.abc as collections
  50    import concurrent.futures
  51    import types
  52
  53    _HTTP_METHOD = typing.Literal["GET", "DELETE", "POST", "PUT", "PATCH"]
  54    _ALLOWED_LANGS = typing.Literal[
  55        "en",
  56        "fr",
  57        "es",
  58        "es-mx",
  59        "de",
  60        "it",
  61        "ja",
  62        "pt-br",
  63        "ru",
  64        "pl",
  65        "ko",
  66        "zh-cht",
  67        "zh-chs",
  68    ]
  69
  70_MANIFEST_LANGUAGES: typing.Final[frozenset[_ALLOWED_LANGS]] = frozenset(
  71    (
  72        "en",
  73        "fr",
  74        "es",
  75        "es-mx",
  76        "de",
  77        "it",
  78        "ja",
  79        "pt-br",
  80        "ru",
  81        "pl",
  82        "ko",
  83        "zh-cht",
  84        "zh-chs",
  85    )
  86)
  87
  88# Client headers.
  89_APP_JSON: str = "application/json"
  90_AUTH_HEADER: str = sys.intern("Authorization")
  91_USER_AGENT_HEADERS: str = sys.intern("User-Agent")
  92_USER_AGENT: str = (
  93    f"AiobungieClient (Author: {metadata.__author__}), "
  94    f"(Version: {metadata.__version__}), (URL: {metadata.__url__})"
  95)
  96
  97# Possible internal error codes.
  98_RETRY_5XX: set[int] = {500, 502, 503, 504}
  99
 100# HTTP methods.
 101_GET: typing.Final[str] = "GET"
 102_POST: typing.Final[str] = "POST"
 103
 104# These are Currently unused.
 105# _DELETE: typing.Final[str] = "DELETE"
 106# _PUT: typing.Final[str] = "PUT"
 107# _PATCH: typing.Final[str] = "PATCH"
 108
 109
 110_LOGGER = logging.getLogger("aiobungie.rest")
 111
 112TRACE: typing.Final[int] = logging.DEBUG - 5
 113"""The trace logging level for the `RESTClient` responses.
 114
 115You can enable this with the following code
 116
 117>>> import logging
 118>>> logging.getLogger("aiobungie.rest").setLevel(aiobungie.TRACE)
 119# or
 120>>> logging.basicConfig(level=aiobungie.TRACE)
 121# Or
 122>>> client = aiobungie.RESTClient(debug="TRACE")
 123# Or if you're using `aiobungie.Client`
 124>>> client = aiobungie.Client()
 125>>> client.rest.with_debug(level=aiobungie.TRACE, file="rest_logs.txt")
 126"""
 127
 128logging.addLevelName(TRACE, "TRACE")
 129
 130
 131def _collect_components(
 132    components: collections.Sequence[enums.ComponentType],
 133    /,
 134) -> str:
 135    collector: collections.MutableSequence[str] = []
 136
 137    for component in components:
 138        if isinstance(component.value, tuple):
 139            collector.extend(str(c) for c in component.value)  # pyright: ignore
 140        else:
 141            collector.append(str(component.value))
 142    return ",".join(collector)
 143
 144
 145def _uuid() -> str:
 146    return uuid.uuid4().hex
 147
 148
 149def _ensure_manifest_language(language: str) -> None:
 150    if language not in _MANIFEST_LANGUAGES:
 151        langs = "\n".join(_MANIFEST_LANGUAGES)
 152        raise ValueError(
 153            f"{language} is not a valid manifest language, valid languages are: {langs}"
 154        )
 155
 156
 157def _get_path(
 158    file_name: str, path: str | pathlib.Path, sql: bool = False
 159) -> pathlib.Path:
 160    if sql:
 161        return pathlib.Path(path).joinpath(file_name + ".sqlite3")
 162    return pathlib.Path(path).joinpath(file_name + ".json")
 163
 164
 165def _write_json_bytes(
 166    data: bytes,
 167    file_name: str = "manifest",
 168    path: pathlib.Path | str = "./",
 169) -> None:
 170    with _get_path(file_name, path).open("wb") as p:
 171        p.write(helpers.dumps(helpers.loads(data)))
 172
 173
 174def _write_sqlite_bytes(
 175    data: bytes,
 176    path: pathlib.Path | str = "./",
 177    file_name: str = "manifest",
 178) -> None:
 179    with open(f"{_uuid()}.zip", "wb") as tmp:
 180        tmp.write(data)
 181        try:
 182            with zipfile.ZipFile(tmp.name) as zipped:
 183                file = zipped.namelist()
 184
 185                if file:
 186                    zipped.extractall(".")
 187
 188                    os.rename(file[0], _get_path(file_name, path, sql=True))
 189
 190        finally:
 191            pathlib.Path(tmp.name).unlink(missing_ok=True)
 192
 193
 194class _JSONPayload(aiohttp.BytesPayload):
 195    def __init__(
 196        self, value: typing.Any, dumps: typedefs.Dumps = helpers.dumps
 197    ) -> None:
 198        super().__init__(dumps(value), content_type=_APP_JSON, encoding="utf-8")
 199
 200
 201class RESTPool:
 202    """a Pool of `RESTClient` instances that shares the same TCP client connection.
 203
 204    This allows you to acquire instances of `RESTClient`s from single settings and credentials.
 205
 206    Example
 207    -------
 208    ```py
 209    import aiobungie
 210    import asyncio
 211
 212    pool = aiobungie.RESTPool("token")
 213
 214    async def get() -> None:
 215        await pool.start()
 216
 217        async with pool.acquire() as client:
 218            await client.fetch_character(...)
 219
 220        await pool.stop()
 221
 222    asyncio.run(get())
 223    ```
 224
 225    Parameters
 226    ----------
 227    token : `str`
 228        A valid application token from Bungie's developer portal.
 229
 230    Other Parameters
 231    ----------------
 232    client_secret : `str | None`
 233        An optional application client secret,
 234        This is only needed if you're fetching OAuth2 tokens with this client.
 235    client_id : `int | None`
 236        An optional application client id,
 237        This is only needed if you're fetching OAuth2 tokens with this client.
 238    settings: `aiobungie.builders.Settings | None`
 239        The client settings to use, if `None` the default will be used.
 240    max_retries : `int`
 241        The max retries number to retry if the request hit a `5xx` status code.
 242    debug : `bool | str`
 243        Whether to enable logging responses or not.
 244
 245    Logging Levels
 246    --------------
 247    * `False`: This will disable logging.
 248    * `True`: This will set the level to `DEBUG` and enable logging minimal information.
 249    Like the response status, route, taken time and so on.
 250    * `"TRACE" | aiobungie.TRACE`: This will log the response headers along with the minimal information.
 251    """
 252
 253    __slots__ = (
 254        "_token",
 255        "_max_retries",
 256        "_client_secret",
 257        "_client_id",
 258        "_metadata",
 259        "_enable_debug",
 260        "_client_session",
 261        "_loads",
 262        "_dumps",
 263        "_settings",
 264    )
 265
 266    # Looks like mypy doesn't like this.
 267    if typing.TYPE_CHECKING:
 268        _enable_debug: typing.Literal["TRACE"] | bool | int
 269
 270    def __init__(
 271        self,
 272        token: str,
 273        /,
 274        *,
 275        client_secret: str | None = None,
 276        client_id: int | None = None,
 277        settings: builders.Settings | None = None,
 278        dumps: typedefs.Dumps = helpers.dumps,
 279        loads: typedefs.Loads = helpers.loads,
 280        max_retries: int = 4,
 281        debug: typing.Literal["TRACE"] | bool | int = False,
 282    ) -> None:
 283        self._client_secret = client_secret
 284        self._client_id = client_id
 285        self._token = token
 286        self._max_retries = max_retries
 287        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
 288        self._enable_debug = debug
 289        self._client_session: aiohttp.ClientSession | None = None
 290        self._loads = loads
 291        self._dumps = dumps
 292        self._settings = settings or builders.Settings()
 293
 294    @property
 295    def client_id(self) -> int | None:
 296        """Return the client id of this REST client if provided, Otherwise None."""
 297        return self._client_id
 298
 299    @property
 300    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
 301        """A general-purpose mutable mapping you can use to store data.
 302
 303        This mapping can be accessed from any process that has a reference to this pool.
 304        """
 305        return self._metadata
 306
 307    @property
 308    def settings(self) -> builders.Settings:
 309        """Internal client settings used within the HTTP client session."""
 310        return self._settings
 311
 312    @typing.overload
 313    def build_oauth2_url(self, client_id: int) -> builders.OAuthURL: ...
 314
 315    @typing.overload
 316    def build_oauth2_url(self) -> builders.OAuthURL | None: ...
 317
 318    @typing.final
 319    def build_oauth2_url(
 320        self, client_id: int | None = None
 321    ) -> builders.OAuthURL | None:
 322        """Construct a new `OAuthURL` url object.
 323
 324        You can get the complete string representation of the url by calling `.compile()` on it.
 325
 326        Parameters
 327        ----------
 328        client_id : `int | None`
 329            An optional client id to provide, If left `None` it will roll back to the id passed
 330            to the `RESTClient`, If both is `None` this method will return `None`.
 331
 332        Returns
 333        -------
 334        `aiobungie.builders.OAuthURL | None`
 335            * If `client_id` was provided as a parameter, It guarantees to return a complete `OAuthURL` object
 336            * If `client_id` is set to `aiobungie.RESTClient` will be.
 337            * If both are `None` this method will return `None.
 338        """
 339        client_id = client_id or self._client_id
 340        if client_id is None:
 341            return None
 342
 343        return builders.OAuthURL(client_id=client_id)
 344
 345    async def start(self) -> None:
 346        """Start the TCP connection of this client pool.
 347
 348        This will raise `RuntimeError` if the connection has already been started.
 349
 350        Example
 351        -------
 352        ```py
 353        pool = aiobungie.RESTPool(...)
 354
 355        async def run() -> None:
 356            await pool.start()
 357            async with pool.acquire() as client:
 358                # use client
 359
 360        async def stop(self) -> None:
 361            await pool.close()
 362        ```
 363        """
 364        if self._client_session is not None:
 365            raise RuntimeError("<RESTPool> has already been started.") from None
 366
 367        self._client_session = aiohttp.ClientSession(
 368            connector=aiohttp.TCPConnector(
 369                use_dns_cache=self._settings.use_dns_cache,
 370                ttl_dns_cache=self._settings.ttl_dns_cache,
 371                ssl_context=self._settings.ssl_context,
 372                ssl=self._settings.ssl,
 373            ),
 374            connector_owner=True,
 375            raise_for_status=False,
 376            timeout=self._settings.http_timeout,
 377            trust_env=self._settings.trust_env,
 378            headers=self._settings.headers,
 379        )
 380
 381    async def stop(self) -> None:
 382        """Stop the TCP connection of this client pool.
 383
 384        This will raise `RuntimeError` if the connection has already been closed.
 385
 386        Example
 387        -------
 388        ```py
 389        pool = aiobungie.RESTPool(...)
 390
 391        async def run() -> None:
 392            await pool.start()
 393            async with pool.acquire() as client:
 394                # use client
 395
 396        async def stop(self) -> None:
 397            await pool.close()
 398        ```
 399        """
 400        if self._client_session is None:
 401            raise RuntimeError("<RESTPool> is already stopped.")
 402
 403        await self._client_session.close()
 404        self._client_session = None
 405
 406    @typing.final
 407    def acquire(self) -> RESTClient:
 408        """Acquires a new `RESTClient` instance from this pool.
 409
 410        Returns
 411        -------
 412        `RESTClient`
 413            An instance of a `RESTClient`.
 414        """
 415        return RESTClient(
 416            self._token,
 417            client_secret=self._client_secret,
 418            client_id=self._client_id,
 419            loads=self._loads,
 420            dumps=self._dumps,
 421            max_retries=self._max_retries,
 422            debug=self._enable_debug,
 423            client_session=self._client_session,
 424            owned_client=False,
 425            settings=self._settings,
 426        )
 427
 428
 429class RESTClient(api.RESTClient):
 430    """A single process REST client implementation.
 431
 432    This client is designed to only make HTTP requests and return raw JSON objects.
 433
 434    Example
 435    -------
 436    ```py
 437    import aiobungie
 438
 439    client = aiobungie.RESTClient("TOKEN")
 440    async with client:
 441        response = await client.fetch_clan_members(4389205)
 442        for member in response['results']:
 443            print(member['destinyUserInfo'])
 444    ```
 445
 446    Parameters
 447    ----------
 448    token : `str`
 449        A valid application token from Bungie's developer portal.
 450
 451    Other Parameters
 452    ----------------
 453    client_secret : `str | None`
 454        An optional application client secret,
 455        This is only needed if you're fetching OAuth2 tokens with this client.
 456    client_id : `int | None`
 457        An optional application client id,
 458        This is only needed if you're fetching OAuth2 tokens with this client.
 459    settings: `aiobungie.builders.Settings | None`
 460        The client settings to use, if `None` the default will be used.
 461    owned_client: `bool`
 462        * If set to `True`, this client will use the provided `client_session` parameter instead,
 463        * If set to `True` and `client_session` is `None`, `ValueError` will be raised.
 464        * If set to `False`, aiobungie will initialize a new client session for you.
 465
 466    client_session: `aiohttp.ClientSession | None`
 467        If provided, this client session will be used to make all the HTTP requests.
 468        The `owned_client` must be set to `True` for this to work.
 469    max_retries : `int`
 470        The max retries number to retry if the request hit a `5xx` status code.
 471    debug : `bool | str`
 472        Whether to enable logging responses or not.
 473
 474    Logging Levels
 475    --------------
 476    * `False`: This will disable logging.
 477    * `True`: This will set the level to `DEBUG` and enable logging minimal information.
 478    * `"TRACE" | aiobungie.TRACE`: This will log the response headers along with the minimal information.
 479    """
 480
 481    __slots__ = (
 482        "_token",
 483        "_session",
 484        "_lock",
 485        "_max_retries",
 486        "_client_secret",
 487        "_client_id",
 488        "_metadata",
 489        "_dumps",
 490        "_loads",
 491        "_owned_client",
 492        "_settings",
 493    )
 494
 495    def __init__(
 496        self,
 497        token: str,
 498        /,
 499        *,
 500        client_secret: str | None = None,
 501        client_id: int | None = None,
 502        settings: builders.Settings | None = None,
 503        owned_client: bool = True,
 504        client_session: aiohttp.ClientSession | None = None,
 505        dumps: typedefs.Dumps = helpers.dumps,
 506        loads: typedefs.Loads = helpers.loads,
 507        max_retries: int = 4,
 508        debug: typing.Literal["TRACE"] | bool | int = False,
 509    ) -> None:
 510        if owned_client is False and client_session is None:
 511            raise ValueError(
 512                "Expected an owned client session, but got `None`, Cannot have `owned_client` set to `False` and `client_session` to `None`"
 513            )
 514
 515        self._settings = settings or builders.Settings()
 516        self._session = client_session
 517        self._owned_client = owned_client
 518        self._lock: asyncio.Lock | None = None
 519        self._client_secret = client_secret
 520        self._client_id = client_id
 521        self._token: str = token
 522        self._max_retries = max_retries
 523        self._dumps = dumps
 524        self._loads = loads
 525        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
 526        self.with_debug(debug)
 527
 528    @property
 529    def client_id(self) -> int | None:
 530        return self._client_id
 531
 532    @property
 533    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
 534        return self._metadata
 535
 536    @property
 537    def is_alive(self) -> bool:
 538        return self._session is not None
 539
 540    @property
 541    def settings(self) -> builders.Settings:
 542        return self._settings
 543
 544    async def close(self) -> None:
 545        if self._session is None:
 546            raise RuntimeError("REST client is not running.")
 547
 548        if self._owned_client:
 549            await self._session.close()
 550            self._session = None
 551
 552    def open(self) -> None:
 553        """Open a new client session. This is called internally with contextmanager usage."""
 554        if self.is_alive and self._owned_client:
 555            raise RuntimeError("Cannot open REST client when it's already open.")
 556
 557        if self._owned_client:
 558            self._session = aiohttp.ClientSession(
 559                connector=aiohttp.TCPConnector(
 560                    use_dns_cache=self._settings.use_dns_cache,
 561                    ttl_dns_cache=self._settings.ttl_dns_cache,
 562                    ssl_context=self._settings.ssl_context,
 563                    ssl=self._settings.ssl,
 564                ),
 565                connector_owner=True,
 566                raise_for_status=False,
 567                timeout=self._settings.http_timeout,
 568                trust_env=self._settings.trust_env,
 569                headers=self._settings.headers,
 570            )
 571
 572    @typing.final
 573    async def static_request(
 574        self,
 575        method: _HTTP_METHOD,
 576        path: str,
 577        *,
 578        auth: str | None = None,
 579        json: collections.Mapping[str, typing.Any] | None = None,
 580        params: collections.Mapping[str, typing.Any] | None = None,
 581    ) -> typedefs.JSONIsh:
 582        return await self._request(method, path, auth=auth, json=json, params=params)
 583
 584    @typing.overload
 585    def build_oauth2_url(self, client_id: int) -> builders.OAuthURL: ...
 586
 587    @typing.overload
 588    def build_oauth2_url(self) -> builders.OAuthURL | None: ...
 589
 590    @typing.final
 591    def build_oauth2_url(
 592        self, client_id: int | None = None
 593    ) -> builders.OAuthURL | None:
 594        client_id = client_id or self._client_id
 595        if client_id is None:
 596            return None
 597
 598        return builders.OAuthURL(client_id=client_id)
 599
 600    @typing.final
 601    async def _request(
 602        self,
 603        method: _HTTP_METHOD,
 604        route: str,
 605        *,
 606        base: bool = False,
 607        oauth2: bool = False,
 608        auth: str | None = None,
 609        unwrap_bytes: bool = False,
 610        json: collections.Mapping[str, typing.Any] | None = None,
 611        data: collections.Mapping[str, typing.Any] | None = None,
 612        params: collections.Mapping[str, typing.Any] | None = None,
 613    ) -> typedefs.JSONIsh:
 614        # This is not None when opening the client.
 615        assert self._session is not None, (
 616            "This client hasn't been opened yet. Use `async with client` or `async with client.rest` "
 617            "before performing any request."
 618        )
 619
 620        retries: int = 0
 621        headers: collections.MutableMapping[str, typing.Any] = {}
 622
 623        headers[_USER_AGENT_HEADERS] = _USER_AGENT
 624        headers["X-API-KEY"] = self._token
 625
 626        if auth is not None:
 627            headers[_AUTH_HEADER] = f"Bearer {auth}"
 628
 629        # Handling endpoints
 630        endpoint = url.BASE
 631
 632        if not base:
 633            endpoint = endpoint + url.REST_EP
 634
 635        if oauth2:
 636            assert self._client_id, "Client ID is required to make authorized requests."
 637            assert self._client_secret, (
 638                "Client secret is required to make authorized requests."
 639            )
 640            headers["client_secret"] = self._client_secret
 641
 642            headers["Content-Type"] = "application/x-www-form-urlencoded"
 643            endpoint = endpoint + url.TOKEN_EP
 644
 645        if self._lock is None:
 646            self._lock = asyncio.Lock()
 647
 648        if json:
 649            headers["Content-Type"] = _APP_JSON
 650
 651        stack = contextlib.AsyncExitStack()
 652        while True:
 653            try:
 654                await stack.enter_async_context(self._lock)
 655
 656                # We make the request here.
 657                taken_time = time.monotonic()
 658                response = await self._session.request(
 659                    method=method,
 660                    url=f"{endpoint}/{route}",
 661                    headers=headers,
 662                    data=_JSONPayload(json) if json else data,
 663                    params=params,
 664                )
 665                response_time = (time.monotonic() - taken_time) * 1_000
 666
 667                _LOGGER.debug(
 668                    "METHOD: %s ROUTE: %s STATUS: %i ELAPSED: %.4fms",
 669                    method,
 670                    f"{endpoint}/{route}",
 671                    response.status,
 672                    response_time,
 673                )
 674
 675                await self._handle_ratelimit(response, method, route)
 676
 677            except aiohttp.ClientConnectionError as exc:
 678                if retries >= self._max_retries:
 679                    raise error.HTTPError(
 680                        str(exc),
 681                        http.HTTPStatus.SERVICE_UNAVAILABLE,
 682                    )
 683                backoff_ = backoff.ExponentialBackOff(maximum=8)
 684
 685                timer = next(backoff_)
 686                _LOGGER.warning(
 687                    "Client received a connection error <%s> Retrying in %.2fs. Remaining retries: %s",
 688                    type(exc).__qualname__,
 689                    timer,
 690                    self._max_retries - retries,
 691                )
 692                retries += 1
 693                await asyncio.sleep(timer)
 694                continue
 695
 696            finally:
 697                await stack.aclose()
 698
 699            if response.status == http.HTTPStatus.NO_CONTENT:
 700                return None
 701
 702            # Handle the successful response.
 703            if 300 > response.status >= 200:
 704                if unwrap_bytes:
 705                    # We need to read the bytes for the manifest response.
 706                    return await response.read()
 707
 708                # Bungie get funky and return HTML instead of JSON when making an authorized
 709                # request with a dummy access token. We could technically read the page content
 710                # but that's Bungie's fault for not returning a JSON response.
 711                if response.content_type != _APP_JSON:
 712                    raise error.HTTPError(
 713                        message=f"Expected JSON response, Got {response.content_type}, "
 714                        f"{response.real_url.human_repr()}",
 715                        http_status=http.HTTPStatus(response.status),
 716                    )
 717
 718                json_data = self._loads(await response.read())
 719
 720                if _LOGGER.isEnabledFor(TRACE):
 721                    _LOGGER.log(
 722                        TRACE,
 723                        "%s",
 724                        error.stringify_headers(dict(response.headers)),
 725                    )
 726
 727                    details: collections.MutableMapping[str, typing.Any] = {}
 728                    if json:
 729                        details["json"] = error.filtered_headers(json)
 730
 731                    if data:
 732                        details["data"] = error.filtered_headers(data)
 733
 734                    if params:
 735                        details["params"] = error.filtered_headers(params)
 736
 737                    if details:
 738                        _LOGGER.log(TRACE, "%s", error.stringify_headers(details))
 739
 740                # Return the response.
 741                # auth responses are not inside a Response object.
 742                if oauth2:
 743                    return json_data
 744
 745                # The reason we have a type ignore is because the actual response type
 746                # is within this `Response` key.
 747                return json_data["Response"]  # type: ignore
 748
 749            if (
 750                response.status in _RETRY_5XX and retries < self._max_retries  # noqa: W503
 751            ):
 752                backoff_ = backoff.ExponentialBackOff(maximum=6)
 753                sleep_time = next(backoff_)
 754                _LOGGER.warning(
 755                    "Got %i - %s. Sleeping for %.2f seconds. Remaining retries: %i",
 756                    response.status,
 757                    response.reason,
 758                    sleep_time,
 759                    self._max_retries - retries,
 760                )
 761
 762                retries += 1
 763                await asyncio.sleep(sleep_time)
 764                continue
 765
 766            raise await error.panic(response)
 767
 768    async def __aenter__(self) -> RESTClient:
 769        self.open()
 770        return self
 771
 772    async def __aexit__(
 773        self,
 774        exception_type: type[BaseException] | None,
 775        exception: BaseException | None,
 776        exception_traceback: types.TracebackType | None,
 777    ) -> None:
 778        await self.close()
 779
 780    # We don't want this to be super complicated.
 781    async def _handle_ratelimit(
 782        self,
 783        response: aiohttp.ClientResponse,
 784        method: str,
 785        route: str,
 786    ) -> None:
 787        if response.status != http.HTTPStatus.TOO_MANY_REQUESTS:
 788            return
 789
 790        if response.content_type != _APP_JSON:
 791            raise error.HTTPError(
 792                f"Being ratelimited on non JSON request, {response.content_type}.",
 793                http.HTTPStatus.TOO_MANY_REQUESTS,
 794            )
 795
 796        # The reason we have a type ignore here is that we guaranteed the content type is JSON above.
 797        json: typedefs.JSONObject = self._loads(await response.read())  # type: ignore
 798        retry_after = float(json.get("ThrottleSeconds", 15.0)) + 0.1
 799        max_calls: int = 0
 800
 801        while True:
 802            if max_calls == 10:
 803                # Max retries by default. We raise an error here.
 804                raise error.RateLimitedError(
 805                    body=json,
 806                    url=str(response.real_url),
 807                    retry_after=retry_after,
 808                )
 809
 810            # We sleep for a little bit to avoid funky behavior.
 811            _LOGGER.warning(
 812                "We're being ratelimited, Method %s Route %s. Sleeping for %.2fs.",
 813                method,
 814                route,
 815                retry_after,
 816            )
 817            await asyncio.sleep(retry_after)
 818            max_calls += 1
 819            continue
 820
 821    async def fetch_oauth2_tokens(self, code: str, /) -> builders.OAuth2Response:
 822        data = {
 823            "grant_type": "authorization_code",
 824            "code": code,
 825            "client_id": self._client_id,
 826            "client_secret": self._client_secret,
 827        }
 828
 829        response = await self._request(_POST, "", data=data, oauth2=True)
 830        assert isinstance(response, dict)
 831        return builders.OAuth2Response.build_response(response)
 832
 833    async def refresh_access_token(
 834        self, refresh_token: str, /
 835    ) -> builders.OAuth2Response:
 836        data = {
 837            "grant_type": "refresh_token",
 838            "refresh_token": refresh_token,
 839            "client_id": self._client_id,
 840            "client_secret": self._client_secret,
 841        }
 842
 843        response = await self._request(_POST, "", data=data, oauth2=True)
 844        assert isinstance(response, dict)
 845        return builders.OAuth2Response.build_response(response)
 846
 847    async def fetch_bungie_user(self, id: int) -> typedefs.JSONObject:
 848        resp = await self._request(_GET, f"User/GetBungieNetUserById/{id}/")
 849        assert isinstance(resp, dict)
 850        return resp
 851
 852    async def fetch_user_themes(self) -> typedefs.JSONArray:
 853        resp = await self._request(_GET, "User/GetAvailableThemes/")
 854        assert isinstance(resp, list)
 855        return resp
 856
 857    async def fetch_membership_from_id(
 858        self,
 859        id: int,
 860        type: enums.MembershipType | int = enums.MembershipType.NONE,
 861        /,
 862    ) -> typedefs.JSONObject:
 863        resp = await self._request(_GET, f"User/GetMembershipsById/{id}/{int(type)}")
 864        assert isinstance(resp, dict)
 865        return resp
 866
 867    async def fetch_membership(
 868        self,
 869        name: str,
 870        code: int,
 871        type: enums.MembershipType | int = enums.MembershipType.ALL,
 872        /,
 873    ) -> typedefs.JSONArray:
 874        resp = await self._request(
 875            _POST,
 876            f"Destiny2/SearchDestinyPlayerByBungieName/{int(type)}",
 877            json={"displayName": name, "displayNameCode": code},
 878        )
 879        assert isinstance(resp, list)
 880        return resp
 881
 882    async def fetch_sanitized_membership(
 883        self, membership_id: int, /
 884    ) -> typedefs.JSONObject:
 885        response = await self._request(
 886            _GET, f"User/GetSanitizedPlatformDisplayNames/{membership_id}/"
 887        )
 888        assert isinstance(response, dict)
 889        return response
 890
 891    async def search_users(self, name: str, /) -> typedefs.JSONObject:
 892        resp = await self._request(
 893            _POST,
 894            "User/Search/GlobalName/0",
 895            json={"displayNamePrefix": name},
 896        )
 897        assert isinstance(resp, dict)
 898        return resp
 899
 900    async def fetch_clan_from_id(
 901        self, id: int, /, access_token: str | None = None
 902    ) -> typedefs.JSONObject:
 903        resp = await self._request(_GET, f"GroupV2/{id}", auth=access_token)
 904        assert isinstance(resp, dict)
 905        return resp
 906
 907    async def fetch_clan(
 908        self,
 909        name: str,
 910        /,
 911        access_token: str | None = None,
 912        *,
 913        type: enums.GroupType | int = enums.GroupType.CLAN,
 914    ) -> typedefs.JSONObject:
 915        resp = await self._request(
 916            _GET, f"GroupV2/Name/{name}/{int(type)}", auth=access_token
 917        )
 918        assert isinstance(resp, dict)
 919        return resp
 920
 921    async def search_group(
 922        self,
 923        name: str,
 924        group_type: enums.GroupType | int = enums.GroupType.CLAN,
 925        *,
 926        creation_date: clans.GroupDate | int = 0,
 927        sort_by: int | None = None,
 928        group_member_count_filter: typing.Literal[0, 1, 2, 3] | None = None,
 929        locale_filter: str | None = None,
 930        tag_text: str | None = None,
 931        items_per_page: int | None = None,
 932        current_page: int | None = None,
 933        request_token: str | None = None,
 934    ) -> typedefs.JSONObject:
 935        payload: collections.MutableMapping[str, typing.Any] = {"name": name}
 936
 937        # as the official documentation says, you're not allowed to use those fields
 938        # on a clan search. it is safe to send the request with them being `null` but not filled with a value.
 939        if (
 940            group_type == enums.GroupType.CLAN
 941            and group_member_count_filter is not None
 942            and locale_filter
 943            and tag_text
 944        ):
 945            raise ValueError(
 946                "If you're searching for clans, (group_member_count_filter, locale_filter, tag_text) must be None."
 947            )
 948
 949        payload["groupType"] = int(group_type)
 950        payload["creationDate"] = int(creation_date)
 951        payload["sortBy"] = sort_by
 952        payload["groupMemberCount"] = group_member_count_filter
 953        payload["locale"] = locale_filter
 954        payload["tagText"] = tag_text
 955        payload["itemsPerPage"] = items_per_page
 956        payload["currentPage"] = current_page
 957        payload["requestToken"] = request_token
 958        payload["requestContinuationToken"] = request_token
 959
 960        resp = await self._request(_POST, "GroupV2/Search/", json=payload)
 961        assert isinstance(resp, dict)
 962        return resp
 963
 964    async def fetch_clan_admins(self, clan_id: int, /) -> typedefs.JSONObject:
 965        resp = await self._request(_GET, f"GroupV2/{clan_id}/AdminsAndFounder/")
 966        assert isinstance(resp, dict)
 967        return resp
 968
 969    async def fetch_clan_conversations(self, clan_id: int, /) -> typedefs.JSONArray:
 970        resp = await self._request(_GET, f"GroupV2/{clan_id}/OptionalConversations/")
 971        assert isinstance(resp, list)
 972        return resp
 973
 974    async def fetch_application(self, appid: int, /) -> typedefs.JSONObject:
 975        resp = await self._request(_GET, f"App/Application/{appid}")
 976        assert isinstance(resp, dict)
 977        return resp
 978
 979    async def fetch_character(
 980        self,
 981        member_id: int,
 982        membership_type: enums.MembershipType | int,
 983        character_id: int,
 984        components: collections.Sequence[enums.ComponentType],
 985        auth: str | None = None,
 986    ) -> typedefs.JSONObject:
 987        collector = _collect_components(components)
 988        response = await self._request(
 989            _GET,
 990            f"Destiny2/{int(membership_type)}/Profile/{member_id}/"
 991            f"Character/{character_id}/?components={collector}",
 992            auth=auth,
 993        )
 994        assert isinstance(response, dict)
 995        return response
 996
 997    async def fetch_activities(
 998        self,
 999        member_id: int,
1000        character_id: int,
1001        mode: enums.GameMode | int,
1002        membership_type: enums.MembershipType | int = enums.MembershipType.ALL,
1003        *,
1004        page: int = 0,
1005        limit: int = 1,
1006    ) -> typedefs.JSONObject:
1007        resp = await self._request(
1008            _GET,
1009            f"Destiny2/{int(membership_type)}/Account/"
1010            f"{member_id}/Character/{character_id}/Stats/Activities"
1011            f"/?mode={int(mode)}&count={limit}&page={page}",
1012        )
1013        assert isinstance(resp, dict)
1014        return resp
1015
1016    async def fetch_vendor_sales(self) -> typedefs.JSONObject:
1017        resp = await self._request(
1018            _GET,
1019            f"Destiny2/Vendors/?components={int(enums.ComponentType.VENDOR_SALES)}",
1020        )
1021        assert isinstance(resp, dict)
1022        return resp
1023
1024    async def fetch_profile(
1025        self,
1026        membership_id: int,
1027        type: enums.MembershipType | int,
1028        components: collections.Sequence[enums.ComponentType],
1029        auth: str | None = None,
1030    ) -> typedefs.JSONObject:
1031        collector = _collect_components(components)
1032        response = await self._request(
1033            _GET,
1034            f"Destiny2/{int(type)}/Profile/{membership_id}/?components={collector}",
1035            auth=auth,
1036        )
1037        assert isinstance(response, dict)
1038        return response
1039
1040    async def fetch_entity(self, type: str, hash: int) -> typedefs.JSONObject:
1041        response = await self._request(_GET, route=f"Destiny2/Manifest/{type}/{hash}")
1042        assert isinstance(response, dict)
1043        return response
1044
1045    async def fetch_inventory_item(self, hash: int, /) -> typedefs.JSONObject:
1046        resp = await self.fetch_entity("DestinyInventoryItemDefinition", hash)
1047        assert isinstance(resp, dict)
1048        return resp
1049
1050    async def fetch_objective_entity(self, hash: int, /) -> typedefs.JSONObject:
1051        resp = await self.fetch_entity("DestinyObjectiveDefinition", hash)
1052        assert isinstance(resp, dict)
1053        return resp
1054
1055    async def fetch_groups_for_member(
1056        self,
1057        member_id: int,
1058        member_type: enums.MembershipType | int,
1059        /,
1060        *,
1061        filter: int = 0,
1062        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1063    ) -> typedefs.JSONObject:
1064        resp = await self._request(
1065            _GET,
1066            f"GroupV2/User/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1067        )
1068        assert isinstance(resp, dict)
1069        return resp
1070
1071    async def fetch_potential_groups_for_member(
1072        self,
1073        member_id: int,
1074        member_type: enums.MembershipType | int,
1075        /,
1076        *,
1077        filter: int = 0,
1078        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1079    ) -> typedefs.JSONObject:
1080        resp = await self._request(
1081            _GET,
1082            f"GroupV2/User/Potential/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1083        )
1084        assert isinstance(resp, dict)
1085        return resp
1086
1087    async def fetch_clan_members(
1088        self,
1089        clan_id: int,
1090        /,
1091        *,
1092        name: str | None = None,
1093        type: enums.MembershipType | int = enums.MembershipType.NONE,
1094    ) -> typedefs.JSONObject:
1095        resp = await self._request(
1096            _GET,
1097            f"/GroupV2/{clan_id}/Members/?memberType={int(type)}&nameSearch={name if name else ''}&currentpage=1",
1098        )
1099        assert isinstance(resp, dict)
1100        return resp
1101
1102    async def fetch_hardlinked_credentials(
1103        self,
1104        credential: int,
1105        type: enums.CredentialType | int = enums.CredentialType.STEAMID,
1106        /,
1107    ) -> typedefs.JSONObject:
1108        resp = await self._request(
1109            _GET,
1110            f"User/GetMembershipFromHardLinkedCredential/{int(type)}/{credential}/",
1111        )
1112        assert isinstance(resp, dict)
1113        return resp
1114
1115    async def fetch_user_credentials(
1116        self, access_token: str, membership_id: int, /
1117    ) -> typedefs.JSONArray:
1118        resp = await self._request(
1119            _GET,
1120            f"User/GetCredentialTypesForTargetAccount/{membership_id}",
1121            auth=access_token,
1122        )
1123        assert isinstance(resp, list)
1124        return resp
1125
1126    async def insert_socket_plug(
1127        self,
1128        action_token: str,
1129        /,
1130        instance_id: int,
1131        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1132        character_id: int,
1133        membership_type: enums.MembershipType | int,
1134    ) -> typedefs.JSONObject:
1135        if isinstance(plug, builders.PlugSocketBuilder):
1136            plug = plug.collect()
1137
1138        body = {
1139            "actionToken": action_token,
1140            "itemInstanceId": instance_id,
1141            "plug": plug,
1142            "characterId": character_id,
1143            "membershipType": int(membership_type),
1144        }
1145        resp = await self._request(
1146            _POST, "Destiny2/Actions/Items/InsertSocketPlug", json=body
1147        )
1148        assert isinstance(resp, dict)
1149        return resp
1150
1151    async def insert_socket_plug_free(
1152        self,
1153        access_token: str,
1154        /,
1155        instance_id: int,
1156        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1157        character_id: int,
1158        membership_type: enums.MembershipType | int,
1159    ) -> typedefs.JSONObject:
1160        if isinstance(plug, builders.PlugSocketBuilder):
1161            plug = plug.collect()
1162
1163        body = {
1164            "itemInstanceId": instance_id,
1165            "plug": plug,
1166            "characterId": character_id,
1167            "membershipType": int(membership_type),
1168        }
1169        resp = await self._request(
1170            _POST,
1171            "Destiny2/Actions/Items/InsertSocketPlugFree",
1172            json=body,
1173            auth=access_token,
1174        )
1175        assert isinstance(resp, dict)
1176        return resp
1177
1178    @helpers.unstable
1179    async def set_item_lock_state(
1180        self,
1181        access_token: str,
1182        state: bool,
1183        /,
1184        item_id: int,
1185        character_id: int,
1186        membership_type: enums.MembershipType | int,
1187    ) -> int:
1188        body = {
1189            "state": state,
1190            "itemId": item_id,
1191            "characterId": character_id,
1192            "membershipType": int(membership_type),
1193        }
1194        response = await self._request(
1195            _POST,
1196            "Destiny2/Actions/Items/SetLockState",
1197            json=body,
1198            auth=access_token,
1199        )
1200        assert isinstance(response, int)
1201        return response
1202
1203    async def set_quest_track_state(
1204        self,
1205        access_token: str,
1206        state: bool,
1207        /,
1208        item_id: int,
1209        character_id: int,
1210        membership_type: enums.MembershipType | int,
1211    ) -> int:
1212        body = {
1213            "state": state,
1214            "itemId": item_id,
1215            "characterId": character_id,
1216            "membership_type": int(membership_type),
1217        }
1218        response = await self._request(
1219            _POST,
1220            "Destiny2/Actions/Items/SetTrackedState",
1221            json=body,
1222            auth=access_token,
1223        )
1224        assert isinstance(response, int)
1225        return response
1226
1227    async def fetch_manifest_path(self) -> typedefs.JSONObject:
1228        path = await self._request(_GET, "Destiny2/Manifest")
1229        assert isinstance(path, dict)
1230        return path
1231
1232    async def read_manifest_bytes(self, language: _ALLOWED_LANGS = "en", /) -> bytes:
1233        _ensure_manifest_language(language)
1234
1235        content = await self.fetch_manifest_path()
1236        resp = await self._request(
1237            _GET,
1238            content["mobileWorldContentPaths"][language],
1239            unwrap_bytes=True,
1240            base=True,
1241        )
1242        assert isinstance(resp, bytes)
1243        return resp
1244
1245    async def download_sqlite_manifest(
1246        self,
1247        language: _ALLOWED_LANGS = "en",
1248        name: str = "manifest",
1249        path: pathlib.Path | str = ".",
1250        *,
1251        force: bool = False,
1252        executor: concurrent.futures.Executor | None = None,
1253    ) -> pathlib.Path:
1254        complete_path = _get_path(name, path, sql=True)
1255
1256        if complete_path.exists():
1257            if force:
1258                _LOGGER.info(
1259                    f"Found manifest in {complete_path!s}. Forcing to Re-Download."
1260                )
1261                complete_path.unlink(missing_ok=True)
1262
1263                return await self.download_sqlite_manifest(
1264                    language, name, path, force=force
1265                )
1266
1267            else:
1268                raise FileExistsError(
1269                    "Manifest file already exists, "
1270                    "To force download, set the `force` parameter to `True`."
1271                )
1272
1273        _LOGGER.info(f"Downloading manifest. Location: {complete_path!s}")
1274        data_bytes = await self.read_manifest_bytes(language)
1275        await asyncio.get_running_loop().run_in_executor(
1276            executor, _write_sqlite_bytes, data_bytes, path, name
1277        )
1278        _LOGGER.info("Finished downloading manifest.")
1279        return _get_path(name, path, sql=True)
1280
1281    async def download_json_manifest(
1282        self,
1283        file_name: str = "manifest",
1284        path: str | pathlib.Path = ".",
1285        *,
1286        language: _ALLOWED_LANGS = "en",
1287        executor: concurrent.futures.Executor | None = None,
1288    ) -> pathlib.Path:
1289        _ensure_manifest_language(language)
1290        full_path = _get_path(file_name, path)
1291        _LOGGER.info(f"Downloading manifest JSON to {full_path!r}...")
1292
1293        content = await self.fetch_manifest_path()
1294        json_bytes = await self._request(
1295            _GET,
1296            content["jsonWorldContentPaths"][language],
1297            unwrap_bytes=True,
1298            base=True,
1299        )
1300
1301        assert isinstance(json_bytes, bytes)
1302        await asyncio.get_running_loop().run_in_executor(
1303            executor, _write_json_bytes, json_bytes, file_name, path
1304        )
1305        _LOGGER.info("Finished downloading manifest JSON.")
1306        return full_path
1307
1308    async def fetch_manifest_version(self) -> str:
1309        # This is guaranteed str.
1310        return (await self.fetch_manifest_path())["version"]
1311
1312    async def fetch_linked_profiles(
1313        self,
1314        member_id: int,
1315        member_type: enums.MembershipType | int,
1316        /,
1317        *,
1318        all: bool = False,
1319    ) -> typedefs.JSONObject:
1320        resp = await self._request(
1321            _GET,
1322            f"Destiny2/{int(member_type)}/Profile/{member_id}/LinkedProfiles/?getAllMemberships={all}",
1323        )
1324        assert isinstance(resp, dict)
1325        return resp
1326
1327    async def fetch_clan_banners(self) -> typedefs.JSONObject:
1328        resp = await self._request(_GET, "Destiny2/Clan/ClanBannerDictionary/")
1329        assert isinstance(resp, dict)
1330        return resp
1331
1332    async def fetch_public_milestones(self) -> typedefs.JSONObject:
1333        resp = await self._request(_GET, "Destiny2/Milestones/")
1334        assert isinstance(resp, dict)
1335        return resp
1336
1337    async def fetch_public_milestone_content(
1338        self, milestone_hash: int, /
1339    ) -> typedefs.JSONObject:
1340        resp = await self._request(
1341            _GET, f"Destiny2/Milestones/{milestone_hash}/Content/"
1342        )
1343        assert isinstance(resp, dict)
1344        return resp
1345
1346    async def fetch_current_user_memberships(
1347        self, access_token: str, /
1348    ) -> typedefs.JSONObject:
1349        resp = await self._request(
1350            _GET,
1351            "User/GetMembershipsForCurrentUser/",
1352            auth=access_token,
1353        )
1354        assert isinstance(resp, dict)
1355        return resp
1356
1357    async def equip_item(
1358        self,
1359        access_token: str,
1360        /,
1361        item_id: int,
1362        character_id: int,
1363        membership_type: enums.MembershipType | int,
1364    ) -> None:
1365        payload = {
1366            "itemId": item_id,
1367            "characterId": character_id,
1368            "membershipType": int(membership_type),
1369        }
1370
1371        await self._request(
1372            _POST,
1373            "Destiny2/Actions/Items/EquipItem/",
1374            json=payload,
1375            auth=access_token,
1376        )
1377
1378    async def equip_items(
1379        self,
1380        access_token: str,
1381        /,
1382        item_ids: collections.Sequence[int],
1383        character_id: int,
1384        membership_type: enums.MembershipType | int,
1385    ) -> None:
1386        payload = {
1387            "itemIds": item_ids,
1388            "characterId": character_id,
1389            "membershipType": int(membership_type),
1390        }
1391        await self._request(
1392            _POST,
1393            "Destiny2/Actions/Items/EquipItems/",
1394            json=payload,
1395            auth=access_token,
1396        )
1397
1398    async def ban_clan_member(
1399        self,
1400        access_token: str,
1401        /,
1402        group_id: int,
1403        membership_id: int,
1404        membership_type: enums.MembershipType | int,
1405        *,
1406        length: int = 0,
1407        comment: str | None = None,
1408    ) -> None:
1409        payload = {"comment": str(comment), "length": length}
1410        await self._request(
1411            _POST,
1412            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Ban/",
1413            json=payload,
1414            auth=access_token,
1415        )
1416
1417    async def unban_clan_member(
1418        self,
1419        access_token: str,
1420        /,
1421        group_id: int,
1422        membership_id: int,
1423        membership_type: enums.MembershipType | int,
1424    ) -> None:
1425        await self._request(
1426            _POST,
1427            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Unban/",
1428            auth=access_token,
1429        )
1430
1431    async def kick_clan_member(
1432        self,
1433        access_token: str,
1434        /,
1435        group_id: int,
1436        membership_id: int,
1437        membership_type: enums.MembershipType | int,
1438    ) -> typedefs.JSONObject:
1439        resp = await self._request(
1440            _POST,
1441            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Kick/",
1442            auth=access_token,
1443        )
1444        assert isinstance(resp, dict)
1445        return resp
1446
1447    async def edit_clan(
1448        self,
1449        access_token: str,
1450        /,
1451        group_id: int,
1452        *,
1453        name: str | None = None,
1454        about: str | None = None,
1455        motto: str | None = None,
1456        theme: str | None = None,
1457        tags: collections.Sequence[str] | None = None,
1458        is_public: bool | None = None,
1459        locale: str | None = None,
1460        avatar_image_index: int | None = None,
1461        membership_option: enums.MembershipOption | int | None = None,
1462        allow_chat: bool | None = None,
1463        chat_security: typing.Literal[0, 1] | None = None,
1464        call_sign: str | None = None,
1465        homepage: typing.Literal[0, 1, 2] | None = None,
1466        enable_invite_messaging_for_admins: bool | None = None,
1467        default_publicity: typing.Literal[0, 1, 2] | None = None,
1468        is_public_topic_admin: bool | None = None,
1469    ) -> None:
1470        payload = {
1471            "name": name,
1472            "about": about,
1473            "motto": motto,
1474            "theme": theme,
1475            "tags": tags,
1476            "isPublic": is_public,
1477            "avatarImageIndex": avatar_image_index,
1478            "isPublicTopicAdminOnly": is_public_topic_admin,
1479            "allowChat": allow_chat,
1480            "chatSecurity": chat_security,
1481            "callsign": call_sign,
1482            "homepage": homepage,
1483            "enableInvitationMessagingForAdmins": enable_invite_messaging_for_admins,
1484            "defaultPublicity": default_publicity,
1485            "locale": locale,
1486        }
1487        if membership_option is not None:
1488            payload["membershipOption"] = int(membership_option)
1489
1490        await self._request(
1491            _POST,
1492            f"GroupV2/{group_id}/Edit",
1493            json=payload,
1494            auth=access_token,
1495        )
1496
1497    async def edit_clan_options(
1498        self,
1499        access_token: str,
1500        /,
1501        group_id: int,
1502        *,
1503        invite_permissions_override: bool | None = None,
1504        update_culture_permissionOverride: bool | None = None,
1505        host_guided_game_permission_override: typing.Literal[0, 1, 2] | None = None,
1506        update_banner_permission_override: bool | None = None,
1507        join_level: enums.ClanMemberType | int | None = None,
1508    ) -> None:
1509        payload = {
1510            "InvitePermissionOverride": invite_permissions_override,
1511            "UpdateCulturePermissionOverride": update_culture_permissionOverride,
1512            "HostGuidedGamePermissionOverride": host_guided_game_permission_override,
1513            "UpdateBannerPermissionOverride": update_banner_permission_override,
1514            "JoinLevel": int(join_level) if join_level else None,
1515        }
1516
1517        await self._request(
1518            _POST,
1519            f"GroupV2/{group_id}/EditFounderOptions",
1520            json=payload,
1521            auth=access_token,
1522        )
1523
1524    async def report_player(
1525        self,
1526        access_token: str,
1527        /,
1528        activity_id: int,
1529        character_id: int,
1530        reason_hashes: collections.Sequence[int],
1531        reason_category_hashes: collections.Sequence[int],
1532    ) -> None:
1533        await self._request(
1534            _POST,
1535            f"Destiny2/Stats/PostGameCarnageReport/{activity_id}/Report/",
1536            json={
1537                "reasonCategoryHashes": reason_category_hashes,
1538                "reasonHashes": reason_hashes,
1539                "offendingCharacterId": character_id,
1540            },
1541            auth=access_token,
1542        )
1543
1544    async def fetch_friends(self, access_token: str, /) -> typedefs.JSONObject:
1545        resp = await self._request(
1546            _GET,
1547            "Social/Friends/",
1548            auth=access_token,
1549        )
1550        assert isinstance(resp, dict)
1551        return resp
1552
1553    async def fetch_friend_requests(self, access_token: str, /) -> typedefs.JSONObject:
1554        resp = await self._request(
1555            _GET,
1556            "Social/Friends/Requests",
1557            auth=access_token,
1558        )
1559        assert isinstance(resp, dict)
1560        return resp
1561
1562    async def accept_friend_request(self, access_token: str, /, member_id: int) -> None:
1563        await self._request(
1564            _POST,
1565            f"Social/Friends/Requests/Accept/{member_id}",
1566            auth=access_token,
1567        )
1568
1569    async def send_friend_request(self, access_token: str, /, member_id: int) -> None:
1570        await self._request(
1571            _POST,
1572            f"Social/Friends/Add/{member_id}",
1573            auth=access_token,
1574        )
1575
1576    async def decline_friend_request(
1577        self, access_token: str, /, member_id: int
1578    ) -> None:
1579        await self._request(
1580            _POST,
1581            f"Social/Friends/Requests/Decline/{member_id}",
1582            auth=access_token,
1583        )
1584
1585    async def remove_friend(self, access_token: str, /, member_id: int) -> None:
1586        await self._request(
1587            _POST,
1588            f"Social/Friends/Remove/{member_id}",
1589            auth=access_token,
1590        )
1591
1592    async def remove_friend_request(self, access_token: str, /, member_id: int) -> None:
1593        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1594        await self._request(
1595            _POST,
1596            f"Social/Friends/Requests/Remove/{member_id}",
1597            auth=access_token,
1598        )
1599
1600    async def approve_all_pending_group_users(
1601        self,
1602        access_token: str,
1603        /,
1604        group_id: int,
1605        message: str | None = None,
1606    ) -> None:
1607        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1608        await self._request(
1609            _POST,
1610            f"GroupV2/{group_id}/Members/ApproveAll",
1611            auth=access_token,
1612            json={"message": str(message)},
1613        )
1614
1615    async def deny_all_pending_group_users(
1616        self,
1617        access_token: str,
1618        /,
1619        group_id: int,
1620        *,
1621        message: str | None = None,
1622    ) -> None:
1623        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1624        await self._request(
1625            _POST,
1626            f"GroupV2/{group_id}/Members/DenyAll",
1627            auth=access_token,
1628            json={"message": str(message)},
1629        )
1630
1631    async def add_optional_conversation(
1632        self,
1633        access_token: str,
1634        /,
1635        group_id: int,
1636        *,
1637        name: str | None = None,
1638        security: typing.Literal[0, 1] = 0,
1639    ) -> None:
1640        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1641        payload = {"chatName": str(name), "chatSecurity": security}
1642        await self._request(
1643            _POST,
1644            f"GroupV2/{group_id}/OptionalConversations/Add",
1645            json=payload,
1646            auth=access_token,
1647        )
1648
1649    async def edit_optional_conversation(
1650        self,
1651        access_token: str,
1652        /,
1653        group_id: int,
1654        conversation_id: int,
1655        *,
1656        name: str | None = None,
1657        security: typing.Literal[0, 1] = 0,
1658        enable_chat: bool = False,
1659    ) -> None:
1660        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1661        payload = {
1662            "chatEnabled": enable_chat,
1663            "chatName": str(name),
1664            "chatSecurity": security,
1665        }
1666        await self._request(
1667            _POST,
1668            f"GroupV2/{group_id}/OptionalConversations/Edit/{conversation_id}",
1669            json=payload,
1670            auth=access_token,
1671        )
1672
1673    async def transfer_item(
1674        self,
1675        access_token: str,
1676        /,
1677        item_id: int,
1678        item_hash: int,
1679        character_id: int,
1680        member_type: enums.MembershipType | int,
1681        *,
1682        stack_size: int = 1,
1683        vault: bool = False,
1684    ) -> None:
1685        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1686        payload = {
1687            "characterId": character_id,
1688            "membershipType": int(member_type),
1689            "itemId": item_id,
1690            "itemReferenceHash": item_hash,
1691            "stackSize": stack_size,
1692            "transferToVault": vault,
1693        }
1694        await self._request(
1695            _POST,
1696            "Destiny2/Actions/Items/TransferItem",
1697            json=payload,
1698            auth=access_token,
1699        )
1700
1701    async def pull_item(
1702        self,
1703        access_token: str,
1704        /,
1705        item_id: int,
1706        item_hash: int,
1707        character_id: int,
1708        member_type: enums.MembershipType | int,
1709        *,
1710        stack_size: int = 1,
1711        vault: bool = False,
1712    ) -> None:
1713        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1714        payload = {
1715            "characterId": character_id,
1716            "membershipType": int(member_type),
1717            "itemId": item_id,
1718            "itemReferenceHash": item_hash,
1719            "stackSize": stack_size,
1720        }
1721        await self._request(
1722            _POST,
1723            "Destiny2/Actions/Items/PullFromPostmaster",
1724            json=payload,
1725            auth=access_token,
1726        )
1727        if vault:
1728            await self.transfer_item(
1729                access_token,
1730                item_id=item_id,
1731                item_hash=item_hash,
1732                character_id=character_id,
1733                member_type=member_type,
1734                stack_size=stack_size,
1735                vault=True,
1736            )
1737
1738    @helpers.unstable
1739    async def fetch_fireteams(
1740        self,
1741        activity_type: fireteams.FireteamActivity | int,
1742        *,
1743        platform: fireteams.FireteamPlatform | int = fireteams.FireteamPlatform.ANY,
1744        language: fireteams.FireteamLanguage | str = fireteams.FireteamLanguage.ALL,
1745        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1746        page: int = 0,
1747        slots_filter: int = 0,
1748    ) -> typedefs.JSONObject:
1749        resp = await self._request(
1750            _GET,
1751            f"Fireteam/Search/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{page}/?langFilter={str(language)}",  # noqa: E501 Line too long
1752        )
1753        assert isinstance(resp, dict)
1754        return resp
1755
1756    async def fetch_available_clan_fireteams(
1757        self,
1758        access_token: str,
1759        group_id: int,
1760        activity_type: fireteams.FireteamActivity | int,
1761        *,
1762        platform: fireteams.FireteamPlatform | int,
1763        language: fireteams.FireteamLanguage | str,
1764        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1765        page: int = 0,
1766        public_only: bool = False,
1767        slots_filter: int = 0,
1768    ) -> typedefs.JSONObject:
1769        resp = await self._request(
1770            _GET,
1771            f"Fireteam/Clan/{group_id}/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{public_only}/{page}",  # noqa: E501
1772            json={"langFilter": str(language)},
1773            auth=access_token,
1774        )
1775        assert isinstance(resp, dict)
1776        return resp
1777
1778    async def fetch_clan_fireteam(
1779        self, access_token: str, fireteam_id: int, group_id: int
1780    ) -> typedefs.JSONObject:
1781        resp = await self._request(
1782            _GET,
1783            f"Fireteam/Clan/{group_id}/Summary/{fireteam_id}",
1784            auth=access_token,
1785        )
1786        assert isinstance(resp, dict)
1787        return resp
1788
1789    async def fetch_my_clan_fireteams(
1790        self,
1791        access_token: str,
1792        group_id: int,
1793        *,
1794        include_closed: bool = True,
1795        platform: fireteams.FireteamPlatform | int,
1796        language: fireteams.FireteamLanguage | str,
1797        filtered: bool = True,
1798        page: int = 0,
1799    ) -> typedefs.JSONObject:
1800        payload = {"groupFilter": filtered, "langFilter": str(language)}
1801
1802        resp = await self._request(
1803            _GET,
1804            f"Fireteam/Clan/{group_id}/My/{int(platform)}/{include_closed}/{page}",
1805            json=payload,
1806            auth=access_token,
1807        )
1808        assert isinstance(resp, dict)
1809        return resp
1810
1811    async def fetch_private_clan_fireteams(
1812        self, access_token: str, group_id: int, /
1813    ) -> int:
1814        resp = await self._request(
1815            _GET,
1816            f"Fireteam/Clan/{group_id}/ActiveCount",
1817            auth=access_token,
1818        )
1819        assert isinstance(resp, int)
1820        return resp
1821
1822    async def fetch_post_activity(self, instance_id: int, /) -> typedefs.JSONObject:
1823        resp = await self._request(
1824            _GET, f"Destiny2/Stats/PostGameCarnageReport/{instance_id}"
1825        )
1826        assert isinstance(resp, dict)
1827        return resp
1828
1829    @helpers.unstable
1830    async def search_entities(
1831        self, name: str, entity_type: str, *, page: int = 0
1832    ) -> typedefs.JSONObject:
1833        resp = await self._request(
1834            _GET,
1835            f"Destiny2/Armory/Search/{entity_type}/{name}/",
1836            json={"page": page},
1837        )
1838        assert isinstance(resp, dict)
1839        return resp
1840
1841    async def fetch_unique_weapon_history(
1842        self,
1843        membership_id: int,
1844        character_id: int,
1845        membership_type: enums.MembershipType | int,
1846    ) -> typedefs.JSONObject:
1847        resp = await self._request(
1848            _GET,
1849            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/UniqueWeapons/",
1850        )
1851        assert isinstance(resp, dict)
1852        return resp
1853
1854    async def fetch_item(
1855        self,
1856        member_id: int,
1857        item_id: int,
1858        membership_type: enums.MembershipType | int,
1859        components: collections.Sequence[enums.ComponentType],
1860    ) -> typedefs.JSONObject:
1861        collector = _collect_components(components)
1862
1863        resp = await self._request(
1864            _GET,
1865            f"Destiny2/{int(membership_type)}/Profile/{member_id}/Item/{item_id}/?components={collector}",
1866        )
1867        assert isinstance(resp, dict)
1868        return resp
1869
1870    async def fetch_clan_weekly_rewards(self, clan_id: int, /) -> typedefs.JSONObject:
1871        resp = await self._request(_GET, f"Destiny2/Clan/{clan_id}/WeeklyRewardState/")
1872        assert isinstance(resp, dict)
1873        return resp
1874
1875    async def fetch_available_locales(self) -> typedefs.JSONObject:
1876        resp = await self._request(_GET, "Destiny2/Manifest/DestinyLocaleDefinition/")
1877        assert isinstance(resp, dict)
1878        return resp
1879
1880    async def fetch_common_settings(self) -> typedefs.JSONObject:
1881        resp = await self._request(_GET, "Settings")
1882        assert isinstance(resp, dict)
1883        return resp
1884
1885    async def fetch_user_systems_overrides(self) -> typedefs.JSONObject:
1886        resp = await self._request(_GET, "UserSystemOverrides")
1887        assert isinstance(resp, dict)
1888        return resp
1889
1890    async def fetch_global_alerts(
1891        self, *, include_streaming: bool = False
1892    ) -> typedefs.JSONArray:
1893        resp = await self._request(
1894            _GET, f"GlobalAlerts/?includestreaming={include_streaming}"
1895        )
1896        assert isinstance(resp, list)
1897        return resp
1898
1899    async def awainitialize_request(
1900        self,
1901        access_token: str,
1902        type: typing.Literal[0, 1],
1903        membership_type: enums.MembershipType | int,
1904        /,
1905        *,
1906        affected_item_id: int | None = None,
1907        character_id: int | None = None,
1908    ) -> typedefs.JSONObject:
1909        body = {"type": type, "membershipType": int(membership_type)}
1910
1911        if affected_item_id is not None:
1912            body["affectedItemId"] = affected_item_id
1913
1914        if character_id is not None:
1915            body["characterId"] = character_id
1916
1917        resp = await self._request(
1918            _POST, "Destiny2/Awa/Initialize", json=body, auth=access_token
1919        )
1920        assert isinstance(resp, dict)
1921        return resp
1922
1923    async def awaget_action_token(
1924        self, access_token: str, correlation_id: str, /
1925    ) -> typedefs.JSONObject:
1926        resp = await self._request(
1927            _POST,
1928            f"Destiny2/Awa/GetActionToken/{correlation_id}",
1929            auth=access_token,
1930        )
1931        assert isinstance(resp, dict)
1932        return resp
1933
1934    async def awa_provide_authorization_result(
1935        self,
1936        access_token: str,
1937        selection: int,
1938        correlation_id: str,
1939        nonce: collections.MutableSequence[str | bytes],
1940    ) -> int:
1941        body = {"selection": selection, "correlationId": correlation_id, "nonce": nonce}
1942
1943        resp = await self._request(
1944            _POST,
1945            "Destiny2/Awa/AwaProvideAuthorizationResult",
1946            json=body,
1947            auth=access_token,
1948        )
1949        assert isinstance(resp, int)
1950        return resp
1951
1952    async def fetch_vendors(
1953        self,
1954        access_token: str,
1955        character_id: int,
1956        membership_id: int,
1957        membership_type: enums.MembershipType | int,
1958        /,
1959        components: collections.Sequence[enums.ComponentType],
1960        filter: int | None = None,
1961    ) -> typedefs.JSONObject:
1962        components_ = _collect_components(components)
1963        route = (
1964            f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1965            f"/Character/{character_id}/Vendors/?components={components_}"
1966        )
1967
1968        if filter is not None:
1969            route = route + f"&filter={filter}"
1970
1971        resp = await self._request(
1972            _GET,
1973            route,
1974            auth=access_token,
1975        )
1976        assert isinstance(resp, dict)
1977        return resp
1978
1979    async def fetch_vendor(
1980        self,
1981        access_token: str,
1982        character_id: int,
1983        membership_id: int,
1984        membership_type: enums.MembershipType | int,
1985        vendor_hash: int,
1986        /,
1987        components: collections.Sequence[enums.ComponentType],
1988    ) -> typedefs.JSONObject:
1989        components_ = _collect_components(components)
1990        resp = await self._request(
1991            _GET,
1992            (
1993                f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1994                f"/Character/{character_id}/Vendors/{vendor_hash}/?components={components_}"
1995            ),
1996            auth=access_token,
1997        )
1998        assert isinstance(resp, dict)
1999        return resp
2000
2001    async def fetch_application_api_usage(
2002        self,
2003        access_token: str,
2004        application_id: int,
2005        /,
2006        *,
2007        start: datetime.datetime | None = None,
2008        end: datetime.datetime | None = None,
2009    ) -> typedefs.JSONObject:
2010        end_date, start_date = time.parse_date_range(end, start)
2011        resp = await self._request(
2012            _GET,
2013            f"App/ApiUsage/{application_id}/?end={end_date}&start={start_date}",
2014            auth=access_token,
2015        )
2016        assert isinstance(resp, dict)
2017        return resp
2018
2019    async def fetch_bungie_applications(self) -> typedefs.JSONArray:
2020        resp = await self._request(_GET, "App/FirstParty")
2021        assert isinstance(resp, list)
2022        return resp
2023
2024    async def fetch_content_type(self, type: str, /) -> typedefs.JSONObject:
2025        resp = await self._request(_GET, f"Content/GetContentType/{type}/")
2026        assert isinstance(resp, dict)
2027        return resp
2028
2029    async def fetch_content_by_id(
2030        self, id: int, locale: str, /, *, head: bool = False
2031    ) -> typedefs.JSONObject:
2032        resp = await self._request(
2033            _GET,
2034            f"Content/GetContentById/{id}/{locale}/",
2035            json={"head": head},
2036        )
2037        assert isinstance(resp, dict)
2038        return resp
2039
2040    async def fetch_content_by_tag_and_type(
2041        self, locale: str, tag: str, type: str, *, head: bool = False
2042    ) -> typedefs.JSONObject:
2043        resp = await self._request(
2044            _GET,
2045            f"Content/GetContentByTagAndType/{tag}/{type}/{locale}/",
2046            json={"head": head},
2047        )
2048        assert isinstance(resp, dict)
2049        return resp
2050
2051    async def search_content_with_text(
2052        self,
2053        locale: str,
2054        /,
2055        content_type: str,
2056        search_text: str,
2057        tag: str,
2058        *,
2059        page: int | None = None,
2060        source: str | None = None,
2061    ) -> typedefs.JSONObject:
2062        body: typedefs.JSONObject = {
2063            "locale": locale,
2064            "currentpage": page or 1,
2065            "ctype": content_type,
2066            "searchtxt": search_text,
2067            "searchtext": search_text,
2068            "tag": tag,
2069            "source": source,
2070        }
2071
2072        resp = await self._request(_GET, "Content/Search", params=body)
2073        assert isinstance(resp, dict)
2074        return resp
2075
2076    async def search_content_by_tag_and_type(
2077        self,
2078        locale: str,
2079        tag: str,
2080        type: str,
2081        *,
2082        page: int | None = None,
2083    ) -> typedefs.JSONObject:
2084        body: typedefs.JSONObject = {"currentpage": page or 1}
2085
2086        resp = await self._request(
2087            _GET,
2088            f"Content/SearchContentByTagAndType/{tag}/{type}/{locale}/",
2089            params=body,
2090        )
2091        assert isinstance(resp, dict)
2092        return resp
2093
2094    async def search_help_articles(
2095        self, text: str, size: str, /
2096    ) -> typedefs.JSONObject:
2097        resp = await self._request(_GET, f"Content/SearchHelpArticles/{text}/{size}/")
2098        assert isinstance(resp, dict)
2099        return resp
2100
2101    async def fetch_topics_page(
2102        self,
2103        category_filter: int,
2104        group: int,
2105        date_filter: int,
2106        sort: str | bytes,
2107        *,
2108        page: int | None = None,
2109        locales: collections.Iterable[str] | None = None,
2110        tag_filter: str | None = None,
2111    ) -> typedefs.JSONObject:
2112        params = {
2113            "locales": ",".join(locales) if locales is not None else "en",
2114        }
2115        if tag_filter:
2116            params["tagstring"] = tag_filter
2117
2118        resp = await self._request(
2119            _GET,
2120            f"Forum/GetTopicsPaged/{page or 0}/0/{group}/{sort!s}/{date_filter}/{category_filter}/",
2121            params=params,
2122        )
2123        assert isinstance(resp, dict)
2124        return resp
2125
2126    async def fetch_core_topics_page(
2127        self,
2128        category_filter: int,
2129        date_filter: int,
2130        sort: str | bytes,
2131        *,
2132        page: int | None = None,
2133        locales: collections.Iterable[str] | None = None,
2134    ) -> typedefs.JSONObject:
2135        resp = await self._request(
2136            _GET,
2137            f"Forum/GetCoreTopicsPaged/{page or 0}"
2138            f"/{sort!s}/{date_filter}/{category_filter}/?locales={','.join(locales) if locales else 'en'}",
2139        )
2140        assert isinstance(resp, dict)
2141        return resp
2142
2143    async def fetch_posts_threaded_page(
2144        self,
2145        parent_post: bool,
2146        page: int,
2147        page_size: int,
2148        parent_post_id: int,
2149        reply_size: int,
2150        root_thread_mode: bool,
2151        sort_mode: int,
2152        show_banned: str | None = None,
2153    ) -> typedefs.JSONObject:
2154        resp = await self._request(
2155            _GET,
2156            f"Forum/GetPostsThreadedPaged/{parent_post}/{page}/"
2157            f"{page_size}/{reply_size}/{parent_post_id}/{root_thread_mode}/{sort_mode}/",
2158            json={"showbanned": show_banned},
2159        )
2160        assert isinstance(resp, dict)
2161        return resp
2162
2163    async def fetch_posts_threaded_page_from_child(
2164        self,
2165        child_id: bool,
2166        page: int,
2167        page_size: int,
2168        reply_size: int,
2169        root_thread_mode: bool,
2170        sort_mode: int,
2171        show_banned: str | None = None,
2172    ) -> typedefs.JSONObject:
2173        resp = await self._request(
2174            _GET,
2175            f"Forum/GetPostsThreadedPagedFromChild/{child_id}/"
2176            f"{page}/{page_size}/{reply_size}/{root_thread_mode}/{sort_mode}/",
2177            json={"showbanned": show_banned},
2178        )
2179        assert isinstance(resp, dict)
2180        return resp
2181
2182    async def fetch_post_and_parent(
2183        self, child_id: int, /, *, show_banned: str | None = None
2184    ) -> typedefs.JSONObject:
2185        resp = await self._request(
2186            _GET,
2187            f"Forum/GetPostAndParent/{child_id}/",
2188            json={"showbanned": show_banned},
2189        )
2190        assert isinstance(resp, dict)
2191        return resp
2192
2193    async def fetch_posts_and_parent_awaiting(
2194        self, child_id: int, /, *, show_banned: str | None = None
2195    ) -> typedefs.JSONObject:
2196        resp = await self._request(
2197            _GET,
2198            f"Forum/GetPostAndParentAwaitingApproval/{child_id}/",
2199            json={"showbanned": show_banned},
2200        )
2201        assert isinstance(resp, dict)
2202        return resp
2203
2204    async def fetch_topic_for_content(self, content_id: int, /) -> int:
2205        resp = await self._request(_GET, f"Forum/GetTopicForContent/{content_id}/")
2206        assert isinstance(resp, int)
2207        return resp
2208
2209    async def fetch_forum_tag_suggestions(
2210        self, partial_tag: str, /
2211    ) -> typedefs.JSONObject:
2212        resp = await self._request(
2213            _GET,
2214            "Forum/GetForumTagSuggestions/",
2215            json={"partialtag": partial_tag},
2216        )
2217        assert isinstance(resp, dict)
2218        return resp
2219
2220    async def fetch_poll(self, topic_id: int, /) -> typedefs.JSONObject:
2221        resp = await self._request(_GET, f"Forum/Poll/{topic_id}/")
2222        assert isinstance(resp, dict)
2223        return resp
2224
2225    async def fetch_recruitment_thread_summaries(self) -> typedefs.JSONArray:
2226        resp = await self._request(_POST, "Forum/Recruit/Summaries/")
2227        assert isinstance(resp, list)
2228        return resp
2229
2230    async def fetch_recommended_groups(
2231        self,
2232        access_token: str,
2233        /,
2234        *,
2235        date_range: int = 0,
2236        group_type: enums.GroupType | int = enums.GroupType.CLAN,
2237    ) -> typedefs.JSONArray:
2238        resp = await self._request(
2239            _POST,
2240            f"GroupV2/Recommended/{int(group_type)}/{date_range}/",
2241            auth=access_token,
2242        )
2243        assert isinstance(resp, list)
2244        return resp
2245
2246    async def fetch_available_avatars(self) -> collections.Mapping[str, int]:
2247        resp = await self._request(_GET, "GroupV2/GetAvailableAvatars/")
2248        assert isinstance(resp, dict)
2249        return resp
2250
2251    async def fetch_user_clan_invite_setting(
2252        self,
2253        access_token: str,
2254        /,
2255        membership_type: enums.MembershipType | int,
2256    ) -> bool:
2257        resp = await self._request(
2258            _GET,
2259            f"GroupV2/GetUserClanInviteSetting/{int(membership_type)}/",
2260            auth=access_token,
2261        )
2262        assert isinstance(resp, bool)
2263        return resp
2264
2265    async def fetch_banned_group_members(
2266        self, access_token: str, group_id: int, /, *, page: int = 1
2267    ) -> typedefs.JSONObject:
2268        resp = await self._request(
2269            _GET,
2270            f"GroupV2/{group_id}/Banned/?currentpage={page}",
2271            auth=access_token,
2272        )
2273        assert isinstance(resp, dict)
2274        return resp
2275
2276    async def fetch_pending_group_memberships(
2277        self, access_token: str, group_id: int, /, *, current_page: int = 1
2278    ) -> typedefs.JSONObject:
2279        resp = await self._request(
2280            _GET,
2281            f"GroupV2/{group_id}/Members/Pending/?currentpage={current_page}",
2282            auth=access_token,
2283        )
2284        assert isinstance(resp, dict)
2285        return resp
2286
2287    async def fetch_invited_group_memberships(
2288        self, access_token: str, group_id: int, /, *, current_page: int = 1
2289    ) -> typedefs.JSONObject:
2290        resp = await self._request(
2291            _GET,
2292            f"GroupV2/{group_id}/Members/InvitedIndividuals/?currentpage={current_page}",
2293            auth=access_token,
2294        )
2295        assert isinstance(resp, dict)
2296        return resp
2297
2298    async def invite_member_to_group(
2299        self,
2300        access_token: str,
2301        /,
2302        group_id: int,
2303        membership_id: int,
2304        membership_type: enums.MembershipType | int,
2305        *,
2306        message: str | None = None,
2307    ) -> typedefs.JSONObject:
2308        resp = await self._request(
2309            _POST,
2310            f"GroupV2/{group_id}/Members/IndividualInvite/{int(membership_type)}/{membership_id}/",
2311            auth=access_token,
2312            json={"message": str(message)},
2313        )
2314        assert isinstance(resp, dict)
2315        return resp
2316
2317    async def cancel_group_member_invite(
2318        self,
2319        access_token: str,
2320        /,
2321        group_id: int,
2322        membership_id: int,
2323        membership_type: enums.MembershipType | int,
2324    ) -> typedefs.JSONObject:
2325        resp = await self._request(
2326            _POST,
2327            f"GroupV2/{group_id}/Members/IndividualInviteCancel/{int(membership_type)}/{membership_id}/",
2328            auth=access_token,
2329        )
2330        assert isinstance(resp, dict)
2331        return resp
2332
2333    async def fetch_historical_definition(self) -> typedefs.JSONObject:
2334        resp = await self._request(_GET, "Destiny2/Stats/Definition/")
2335        assert isinstance(resp, dict)
2336        return resp
2337
2338    async def fetch_historical_stats(
2339        self,
2340        character_id: int,
2341        membership_id: int,
2342        membership_type: enums.MembershipType | int,
2343        day_start: datetime.datetime,
2344        day_end: datetime.datetime,
2345        groups: collections.Sequence[enums.StatsGroupType | int],
2346        modes: collections.Sequence[enums.GameMode | int],
2347        *,
2348        period_type: enums.PeriodType = enums.PeriodType.ALL_TIME,
2349    ) -> typedefs.JSONObject:
2350        end, start = time.parse_date_range(day_end, day_start)
2351        resp = await self._request(
2352            _GET,
2353            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/",
2354            json={
2355                "dayend": end,
2356                "daystart": start,
2357                "groups": [str(int(group)) for group in groups],
2358                "modes": [str(int(mode)) for mode in modes],
2359                "periodType": int(period_type),
2360            },
2361        )
2362        assert isinstance(resp, dict)
2363        return resp
2364
2365    async def fetch_historical_stats_for_account(
2366        self,
2367        membership_id: int,
2368        membership_type: enums.MembershipType | int,
2369        groups: collections.Sequence[enums.StatsGroupType | int],
2370    ) -> typedefs.JSONObject:
2371        resp = await self._request(
2372            _GET,
2373            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Stats/",
2374            json={"groups": [str(int(group)) for group in groups]},
2375        )
2376        assert isinstance(resp, dict)
2377        return resp
2378
2379    async def fetch_aggregated_activity_stats(
2380        self,
2381        character_id: int,
2382        membership_id: int,
2383        membership_type: enums.MembershipType | int,
2384        /,
2385    ) -> typedefs.JSONObject:
2386        resp = await self._request(
2387            _GET,
2388            f"Destiny2/{int(membership_type)}/Account/{membership_id}/"
2389            f"Character/{character_id}/Stats/AggregateActivityStats/",
2390        )
2391        assert isinstance(resp, dict)
2392        return resp
2393
2394    async def equip_loadout(
2395        self,
2396        access_token: str,
2397        /,
2398        loadout_index: int,
2399        character_id: int,
2400        membership_type: enums.MembershipType | int,
2401    ) -> None:
2402        response = await self._request(
2403            _POST,
2404            "Destiny2/Actions/Loadouts/EquipLoadout/",
2405            json={
2406                "loadoutIndex": loadout_index,
2407                "characterId": character_id,
2408                "membership_type": int(membership_type),
2409            },
2410            auth=access_token,
2411        )
2412        assert isinstance(response, int)
2413
2414    async def snapshot_loadout(
2415        self,
2416        access_token: str,
2417        /,
2418        loadout_index: int,
2419        character_id: int,
2420        membership_type: enums.MembershipType | int,
2421        *,
2422        color_hash: int | None = None,
2423        icon_hash: int | None = None,
2424        name_hash: int | None = None,
2425    ) -> None:
2426        response = await self._request(
2427            _POST,
2428            "Destiny2/Actions/Loadouts/SnapshotLoadout/",
2429            auth=access_token,
2430            json={
2431                "colorHash": color_hash,
2432                "iconHash": icon_hash,
2433                "nameHash": name_hash,
2434                "loadoutIndex": loadout_index,
2435                "characterId": character_id,
2436                "membershipType": int(membership_type),
2437            },
2438        )
2439        assert isinstance(response, int)
2440
2441    async def update_loadout(
2442        self,
2443        access_token: str,
2444        /,
2445        loadout_index: int,
2446        character_id: int,
2447        membership_type: enums.MembershipType | int,
2448        *,
2449        color_hash: int | None = None,
2450        icon_hash: int | None = None,
2451        name_hash: int | None = None,
2452    ) -> None:
2453        response = await self._request(
2454            _POST,
2455            "Destiny2/Actions/Loadouts/UpdateLoadoutIdentifiers/",
2456            auth=access_token,
2457            json={
2458                "colorHash": color_hash,
2459                "iconHash": icon_hash,
2460                "nameHash": name_hash,
2461                "loadoutIndex": loadout_index,
2462                "characterId": character_id,
2463                "membershipType": int(membership_type),
2464            },
2465        )
2466        assert isinstance(response, int)
2467
2468    async def clear_loadout(
2469        self,
2470        access_token: str,
2471        /,
2472        loadout_index: int,
2473        character_id: int,
2474        membership_type: enums.MembershipType | int,
2475    ) -> None:
2476        response = await self._request(
2477            _POST,
2478            "Destiny2/Actions/Loadouts/ClearLoadout/",
2479            json={
2480                "loadoutIndex": loadout_index,
2481                "characterId": character_id,
2482                "membership_type": int(membership_type),
2483            },
2484            auth=access_token,
2485        )
2486        assert isinstance(response, int)
2487
2488    async def force_drops_repair(self, access_token: str, /) -> bool:
2489        response = await self._request(
2490            _POST, "Tokens/Partner/ForceDropsRepair/", auth=access_token
2491        )
2492        assert isinstance(response, bool)
2493        return response
2494
2495    async def claim_partner_offer(
2496        self,
2497        access_token: str,
2498        /,
2499        *,
2500        offer_id: str,
2501        bungie_membership_id: int,
2502        transaction_id: str,
2503    ) -> bool:
2504        response = await self._request(
2505            _POST,
2506            "Tokens/Partner/ClaimOffer/",
2507            json={
2508                "PartnerOfferId": offer_id,
2509                "BungieNetMembershipId": bungie_membership_id,
2510                "TransactionId": transaction_id,
2511            },
2512            auth=access_token,
2513        )
2514        assert isinstance(response, bool)
2515        return response
2516
2517    async def fetch_bungie_rewards_for_user(
2518        self, access_token: str, /, membership_id: int
2519    ) -> typedefs.JSONObject:
2520        response = await self._request(
2521            _GET,
2522            f"Tokens/Rewards/GetRewardsForUser/{membership_id}/",
2523            auth=access_token,
2524        )
2525        assert isinstance(response, dict)
2526        return response
2527
2528    async def fetch_bungie_rewards_for_platform(
2529        self,
2530        access_token: str,
2531        /,
2532        membership_id: int,
2533        membership_type: enums.MembershipType | int,
2534    ) -> typedefs.JSONObject:
2535        response = await self._request(
2536            _GET,
2537            f"Tokens/Rewards/GetRewardsForPlatformUser/{membership_id}/{int(membership_type)}",
2538            auth=access_token,
2539        )
2540        assert isinstance(response, dict)
2541        return response
2542
2543    async def fetch_bungie_rewards(self) -> typedefs.JSONObject:
2544        response = await self._request(_GET, "Tokens/Rewards/BungieRewards/")
2545        assert isinstance(response, dict)
2546        return response
2547
2548    async def fetch_fireteam_listing(self, listing_id: int) -> typedefs.JSONObject:
2549        response = await self._request(_GET, f"FireteamFinder/Listing/{listing_id}/")
2550        assert isinstance(response, dict)
2551        return response

class RESTClient(aiobungie.api.rest.RESTClient): View Source

 430class RESTClient(api.RESTClient):
 431    """A single process REST client implementation.
 432
 433    This client is designed to only make HTTP requests and return raw JSON objects.
 434
 435    Example
 436    -------
 437    ```py
 438    import aiobungie
 439
 440    client = aiobungie.RESTClient("TOKEN")
 441    async with client:
 442        response = await client.fetch_clan_members(4389205)
 443        for member in response['results']:
 444            print(member['destinyUserInfo'])
 445    ```
 446
 447    Parameters
 448    ----------
 449    token : `str`
 450        A valid application token from Bungie's developer portal.
 451
 452    Other Parameters
 453    ----------------
 454    client_secret : `str | None`
 455        An optional application client secret,
 456        This is only needed if you're fetching OAuth2 tokens with this client.
 457    client_id : `int | None`
 458        An optional application client id,
 459        This is only needed if you're fetching OAuth2 tokens with this client.
 460    settings: `aiobungie.builders.Settings | None`
 461        The client settings to use, if `None` the default will be used.
 462    owned_client: `bool`
 463        * If set to `True`, this client will use the provided `client_session` parameter instead,
 464        * If set to `True` and `client_session` is `None`, `ValueError` will be raised.
 465        * If set to `False`, aiobungie will initialize a new client session for you.
 466
 467    client_session: `aiohttp.ClientSession | None`
 468        If provided, this client session will be used to make all the HTTP requests.
 469        The `owned_client` must be set to `True` for this to work.
 470    max_retries : `int`
 471        The max retries number to retry if the request hit a `5xx` status code.
 472    debug : `bool | str`
 473        Whether to enable logging responses or not.
 474
 475    Logging Levels
 476    --------------
 477    * `False`: This will disable logging.
 478    * `True`: This will set the level to `DEBUG` and enable logging minimal information.
 479    * `"TRACE" | aiobungie.TRACE`: This will log the response headers along with the minimal information.
 480    """
 481
 482    __slots__ = (
 483        "_token",
 484        "_session",
 485        "_lock",
 486        "_max_retries",
 487        "_client_secret",
 488        "_client_id",
 489        "_metadata",
 490        "_dumps",
 491        "_loads",
 492        "_owned_client",
 493        "_settings",
 494    )
 495
 496    def __init__(
 497        self,
 498        token: str,
 499        /,
 500        *,
 501        client_secret: str | None = None,
 502        client_id: int | None = None,
 503        settings: builders.Settings | None = None,
 504        owned_client: bool = True,
 505        client_session: aiohttp.ClientSession | None = None,
 506        dumps: typedefs.Dumps = helpers.dumps,
 507        loads: typedefs.Loads = helpers.loads,
 508        max_retries: int = 4,
 509        debug: typing.Literal["TRACE"] | bool | int = False,
 510    ) -> None:
 511        if owned_client is False and client_session is None:
 512            raise ValueError(
 513                "Expected an owned client session, but got `None`, Cannot have `owned_client` set to `False` and `client_session` to `None`"
 514            )
 515
 516        self._settings = settings or builders.Settings()
 517        self._session = client_session
 518        self._owned_client = owned_client
 519        self._lock: asyncio.Lock | None = None
 520        self._client_secret = client_secret
 521        self._client_id = client_id
 522        self._token: str = token
 523        self._max_retries = max_retries
 524        self._dumps = dumps
 525        self._loads = loads
 526        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
 527        self.with_debug(debug)
 528
 529    @property
 530    def client_id(self) -> int | None:
 531        return self._client_id
 532
 533    @property
 534    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
 535        return self._metadata
 536
 537    @property
 538    def is_alive(self) -> bool:
 539        return self._session is not None
 540
 541    @property
 542    def settings(self) -> builders.Settings:
 543        return self._settings
 544
 545    async def close(self) -> None:
 546        if self._session is None:
 547            raise RuntimeError("REST client is not running.")
 548
 549        if self._owned_client:
 550            await self._session.close()
 551            self._session = None
 552
 553    def open(self) -> None:
 554        """Open a new client session. This is called internally with contextmanager usage."""
 555        if self.is_alive and self._owned_client:
 556            raise RuntimeError("Cannot open REST client when it's already open.")
 557
 558        if self._owned_client:
 559            self._session = aiohttp.ClientSession(
 560                connector=aiohttp.TCPConnector(
 561                    use_dns_cache=self._settings.use_dns_cache,
 562                    ttl_dns_cache=self._settings.ttl_dns_cache,
 563                    ssl_context=self._settings.ssl_context,
 564                    ssl=self._settings.ssl,
 565                ),
 566                connector_owner=True,
 567                raise_for_status=False,
 568                timeout=self._settings.http_timeout,
 569                trust_env=self._settings.trust_env,
 570                headers=self._settings.headers,
 571            )
 572
 573    @typing.final
 574    async def static_request(
 575        self,
 576        method: _HTTP_METHOD,
 577        path: str,
 578        *,
 579        auth: str | None = None,
 580        json: collections.Mapping[str, typing.Any] | None = None,
 581        params: collections.Mapping[str, typing.Any] | None = None,
 582    ) -> typedefs.JSONIsh:
 583        return await self._request(method, path, auth=auth, json=json, params=params)
 584
 585    @typing.overload
 586    def build_oauth2_url(self, client_id: int) -> builders.OAuthURL: ...
 587
 588    @typing.overload
 589    def build_oauth2_url(self) -> builders.OAuthURL | None: ...
 590
 591    @typing.final
 592    def build_oauth2_url(
 593        self, client_id: int | None = None
 594    ) -> builders.OAuthURL | None:
 595        client_id = client_id or self._client_id
 596        if client_id is None:
 597            return None
 598
 599        return builders.OAuthURL(client_id=client_id)
 600
 601    @typing.final
 602    async def _request(
 603        self,
 604        method: _HTTP_METHOD,
 605        route: str,
 606        *,
 607        base: bool = False,
 608        oauth2: bool = False,
 609        auth: str | None = None,
 610        unwrap_bytes: bool = False,
 611        json: collections.Mapping[str, typing.Any] | None = None,
 612        data: collections.Mapping[str, typing.Any] | None = None,
 613        params: collections.Mapping[str, typing.Any] | None = None,
 614    ) -> typedefs.JSONIsh:
 615        # This is not None when opening the client.
 616        assert self._session is not None, (
 617            "This client hasn't been opened yet. Use `async with client` or `async with client.rest` "
 618            "before performing any request."
 619        )
 620
 621        retries: int = 0
 622        headers: collections.MutableMapping[str, typing.Any] = {}
 623
 624        headers[_USER_AGENT_HEADERS] = _USER_AGENT
 625        headers["X-API-KEY"] = self._token
 626
 627        if auth is not None:
 628            headers[_AUTH_HEADER] = f"Bearer {auth}"
 629
 630        # Handling endpoints
 631        endpoint = url.BASE
 632
 633        if not base:
 634            endpoint = endpoint + url.REST_EP
 635
 636        if oauth2:
 637            assert self._client_id, "Client ID is required to make authorized requests."
 638            assert self._client_secret, (
 639                "Client secret is required to make authorized requests."
 640            )
 641            headers["client_secret"] = self._client_secret
 642
 643            headers["Content-Type"] = "application/x-www-form-urlencoded"
 644            endpoint = endpoint + url.TOKEN_EP
 645
 646        if self._lock is None:
 647            self._lock = asyncio.Lock()
 648
 649        if json:
 650            headers["Content-Type"] = _APP_JSON
 651
 652        stack = contextlib.AsyncExitStack()
 653        while True:
 654            try:
 655                await stack.enter_async_context(self._lock)
 656
 657                # We make the request here.
 658                taken_time = time.monotonic()
 659                response = await self._session.request(
 660                    method=method,
 661                    url=f"{endpoint}/{route}",
 662                    headers=headers,
 663                    data=_JSONPayload(json) if json else data,
 664                    params=params,
 665                )
 666                response_time = (time.monotonic() - taken_time) * 1_000
 667
 668                _LOGGER.debug(
 669                    "METHOD: %s ROUTE: %s STATUS: %i ELAPSED: %.4fms",
 670                    method,
 671                    f"{endpoint}/{route}",
 672                    response.status,
 673                    response_time,
 674                )
 675
 676                await self._handle_ratelimit(response, method, route)
 677
 678            except aiohttp.ClientConnectionError as exc:
 679                if retries >= self._max_retries:
 680                    raise error.HTTPError(
 681                        str(exc),
 682                        http.HTTPStatus.SERVICE_UNAVAILABLE,
 683                    )
 684                backoff_ = backoff.ExponentialBackOff(maximum=8)
 685
 686                timer = next(backoff_)
 687                _LOGGER.warning(
 688                    "Client received a connection error <%s> Retrying in %.2fs. Remaining retries: %s",
 689                    type(exc).__qualname__,
 690                    timer,
 691                    self._max_retries - retries,
 692                )
 693                retries += 1
 694                await asyncio.sleep(timer)
 695                continue
 696
 697            finally:
 698                await stack.aclose()
 699
 700            if response.status == http.HTTPStatus.NO_CONTENT:
 701                return None
 702
 703            # Handle the successful response.
 704            if 300 > response.status >= 200:
 705                if unwrap_bytes:
 706                    # We need to read the bytes for the manifest response.
 707                    return await response.read()
 708
 709                # Bungie get funky and return HTML instead of JSON when making an authorized
 710                # request with a dummy access token. We could technically read the page content
 711                # but that's Bungie's fault for not returning a JSON response.
 712                if response.content_type != _APP_JSON:
 713                    raise error.HTTPError(
 714                        message=f"Expected JSON response, Got {response.content_type}, "
 715                        f"{response.real_url.human_repr()}",
 716                        http_status=http.HTTPStatus(response.status),
 717                    )
 718
 719                json_data = self._loads(await response.read())
 720
 721                if _LOGGER.isEnabledFor(TRACE):
 722                    _LOGGER.log(
 723                        TRACE,
 724                        "%s",
 725                        error.stringify_headers(dict(response.headers)),
 726                    )
 727
 728                    details: collections.MutableMapping[str, typing.Any] = {}
 729                    if json:
 730                        details["json"] = error.filtered_headers(json)
 731
 732                    if data:
 733                        details["data"] = error.filtered_headers(data)
 734
 735                    if params:
 736                        details["params"] = error.filtered_headers(params)
 737
 738                    if details:
 739                        _LOGGER.log(TRACE, "%s", error.stringify_headers(details))
 740
 741                # Return the response.
 742                # auth responses are not inside a Response object.
 743                if oauth2:
 744                    return json_data
 745
 746                # The reason we have a type ignore is because the actual response type
 747                # is within this `Response` key.
 748                return json_data["Response"]  # type: ignore
 749
 750            if (
 751                response.status in _RETRY_5XX and retries < self._max_retries  # noqa: W503
 752            ):
 753                backoff_ = backoff.ExponentialBackOff(maximum=6)
 754                sleep_time = next(backoff_)
 755                _LOGGER.warning(
 756                    "Got %i - %s. Sleeping for %.2f seconds. Remaining retries: %i",
 757                    response.status,
 758                    response.reason,
 759                    sleep_time,
 760                    self._max_retries - retries,
 761                )
 762
 763                retries += 1
 764                await asyncio.sleep(sleep_time)
 765                continue
 766
 767            raise await error.panic(response)
 768
 769    async def __aenter__(self) -> RESTClient:
 770        self.open()
 771        return self
 772
 773    async def __aexit__(
 774        self,
 775        exception_type: type[BaseException] | None,
 776        exception: BaseException | None,
 777        exception_traceback: types.TracebackType | None,
 778    ) -> None:
 779        await self.close()
 780
 781    # We don't want this to be super complicated.
 782    async def _handle_ratelimit(
 783        self,
 784        response: aiohttp.ClientResponse,
 785        method: str,
 786        route: str,
 787    ) -> None:
 788        if response.status != http.HTTPStatus.TOO_MANY_REQUESTS:
 789            return
 790
 791        if response.content_type != _APP_JSON:
 792            raise error.HTTPError(
 793                f"Being ratelimited on non JSON request, {response.content_type}.",
 794                http.HTTPStatus.TOO_MANY_REQUESTS,
 795            )
 796
 797        # The reason we have a type ignore here is that we guaranteed the content type is JSON above.
 798        json: typedefs.JSONObject = self._loads(await response.read())  # type: ignore
 799        retry_after = float(json.get("ThrottleSeconds", 15.0)) + 0.1
 800        max_calls: int = 0
 801
 802        while True:
 803            if max_calls == 10:
 804                # Max retries by default. We raise an error here.
 805                raise error.RateLimitedError(
 806                    body=json,
 807                    url=str(response.real_url),
 808                    retry_after=retry_after,
 809                )
 810
 811            # We sleep for a little bit to avoid funky behavior.
 812            _LOGGER.warning(
 813                "We're being ratelimited, Method %s Route %s. Sleeping for %.2fs.",
 814                method,
 815                route,
 816                retry_after,
 817            )
 818            await asyncio.sleep(retry_after)
 819            max_calls += 1
 820            continue
 821
 822    async def fetch_oauth2_tokens(self, code: str, /) -> builders.OAuth2Response:
 823        data = {
 824            "grant_type": "authorization_code",
 825            "code": code,
 826            "client_id": self._client_id,
 827            "client_secret": self._client_secret,
 828        }
 829
 830        response = await self._request(_POST, "", data=data, oauth2=True)
 831        assert isinstance(response, dict)
 832        return builders.OAuth2Response.build_response(response)
 833
 834    async def refresh_access_token(
 835        self, refresh_token: str, /
 836    ) -> builders.OAuth2Response:
 837        data = {
 838            "grant_type": "refresh_token",
 839            "refresh_token": refresh_token,
 840            "client_id": self._client_id,
 841            "client_secret": self._client_secret,
 842        }
 843
 844        response = await self._request(_POST, "", data=data, oauth2=True)
 845        assert isinstance(response, dict)
 846        return builders.OAuth2Response.build_response(response)
 847
 848    async def fetch_bungie_user(self, id: int) -> typedefs.JSONObject:
 849        resp = await self._request(_GET, f"User/GetBungieNetUserById/{id}/")
 850        assert isinstance(resp, dict)
 851        return resp
 852
 853    async def fetch_user_themes(self) -> typedefs.JSONArray:
 854        resp = await self._request(_GET, "User/GetAvailableThemes/")
 855        assert isinstance(resp, list)
 856        return resp
 857
 858    async def fetch_membership_from_id(
 859        self,
 860        id: int,
 861        type: enums.MembershipType | int = enums.MembershipType.NONE,
 862        /,
 863    ) -> typedefs.JSONObject:
 864        resp = await self._request(_GET, f"User/GetMembershipsById/{id}/{int(type)}")
 865        assert isinstance(resp, dict)
 866        return resp
 867
 868    async def fetch_membership(
 869        self,
 870        name: str,
 871        code: int,
 872        type: enums.MembershipType | int = enums.MembershipType.ALL,
 873        /,
 874    ) -> typedefs.JSONArray:
 875        resp = await self._request(
 876            _POST,
 877            f"Destiny2/SearchDestinyPlayerByBungieName/{int(type)}",
 878            json={"displayName": name, "displayNameCode": code},
 879        )
 880        assert isinstance(resp, list)
 881        return resp
 882
 883    async def fetch_sanitized_membership(
 884        self, membership_id: int, /
 885    ) -> typedefs.JSONObject:
 886        response = await self._request(
 887            _GET, f"User/GetSanitizedPlatformDisplayNames/{membership_id}/"
 888        )
 889        assert isinstance(response, dict)
 890        return response
 891
 892    async def search_users(self, name: str, /) -> typedefs.JSONObject:
 893        resp = await self._request(
 894            _POST,
 895            "User/Search/GlobalName/0",
 896            json={"displayNamePrefix": name},
 897        )
 898        assert isinstance(resp, dict)
 899        return resp
 900
 901    async def fetch_clan_from_id(
 902        self, id: int, /, access_token: str | None = None
 903    ) -> typedefs.JSONObject:
 904        resp = await self._request(_GET, f"GroupV2/{id}", auth=access_token)
 905        assert isinstance(resp, dict)
 906        return resp
 907
 908    async def fetch_clan(
 909        self,
 910        name: str,
 911        /,
 912        access_token: str | None = None,
 913        *,
 914        type: enums.GroupType | int = enums.GroupType.CLAN,
 915    ) -> typedefs.JSONObject:
 916        resp = await self._request(
 917            _GET, f"GroupV2/Name/{name}/{int(type)}", auth=access_token
 918        )
 919        assert isinstance(resp, dict)
 920        return resp
 921
 922    async def search_group(
 923        self,
 924        name: str,
 925        group_type: enums.GroupType | int = enums.GroupType.CLAN,
 926        *,
 927        creation_date: clans.GroupDate | int = 0,
 928        sort_by: int | None = None,
 929        group_member_count_filter: typing.Literal[0, 1, 2, 3] | None = None,
 930        locale_filter: str | None = None,
 931        tag_text: str | None = None,
 932        items_per_page: int | None = None,
 933        current_page: int | None = None,
 934        request_token: str | None = None,
 935    ) -> typedefs.JSONObject:
 936        payload: collections.MutableMapping[str, typing.Any] = {"name": name}
 937
 938        # as the official documentation says, you're not allowed to use those fields
 939        # on a clan search. it is safe to send the request with them being `null` but not filled with a value.
 940        if (
 941            group_type == enums.GroupType.CLAN
 942            and group_member_count_filter is not None
 943            and locale_filter
 944            and tag_text
 945        ):
 946            raise ValueError(
 947                "If you're searching for clans, (group_member_count_filter, locale_filter, tag_text) must be None."
 948            )
 949
 950        payload["groupType"] = int(group_type)
 951        payload["creationDate"] = int(creation_date)
 952        payload["sortBy"] = sort_by
 953        payload["groupMemberCount"] = group_member_count_filter
 954        payload["locale"] = locale_filter
 955        payload["tagText"] = tag_text
 956        payload["itemsPerPage"] = items_per_page
 957        payload["currentPage"] = current_page
 958        payload["requestToken"] = request_token
 959        payload["requestContinuationToken"] = request_token
 960
 961        resp = await self._request(_POST, "GroupV2/Search/", json=payload)
 962        assert isinstance(resp, dict)
 963        return resp
 964
 965    async def fetch_clan_admins(self, clan_id: int, /) -> typedefs.JSONObject:
 966        resp = await self._request(_GET, f"GroupV2/{clan_id}/AdminsAndFounder/")
 967        assert isinstance(resp, dict)
 968        return resp
 969
 970    async def fetch_clan_conversations(self, clan_id: int, /) -> typedefs.JSONArray:
 971        resp = await self._request(_GET, f"GroupV2/{clan_id}/OptionalConversations/")
 972        assert isinstance(resp, list)
 973        return resp
 974
 975    async def fetch_application(self, appid: int, /) -> typedefs.JSONObject:
 976        resp = await self._request(_GET, f"App/Application/{appid}")
 977        assert isinstance(resp, dict)
 978        return resp
 979
 980    async def fetch_character(
 981        self,
 982        member_id: int,
 983        membership_type: enums.MembershipType | int,
 984        character_id: int,
 985        components: collections.Sequence[enums.ComponentType],
 986        auth: str | None = None,
 987    ) -> typedefs.JSONObject:
 988        collector = _collect_components(components)
 989        response = await self._request(
 990            _GET,
 991            f"Destiny2/{int(membership_type)}/Profile/{member_id}/"
 992            f"Character/{character_id}/?components={collector}",
 993            auth=auth,
 994        )
 995        assert isinstance(response, dict)
 996        return response
 997
 998    async def fetch_activities(
 999        self,
1000        member_id: int,
1001        character_id: int,
1002        mode: enums.GameMode | int,
1003        membership_type: enums.MembershipType | int = enums.MembershipType.ALL,
1004        *,
1005        page: int = 0,
1006        limit: int = 1,
1007    ) -> typedefs.JSONObject:
1008        resp = await self._request(
1009            _GET,
1010            f"Destiny2/{int(membership_type)}/Account/"
1011            f"{member_id}/Character/{character_id}/Stats/Activities"
1012            f"/?mode={int(mode)}&count={limit}&page={page}",
1013        )
1014        assert isinstance(resp, dict)
1015        return resp
1016
1017    async def fetch_vendor_sales(self) -> typedefs.JSONObject:
1018        resp = await self._request(
1019            _GET,
1020            f"Destiny2/Vendors/?components={int(enums.ComponentType.VENDOR_SALES)}",
1021        )
1022        assert isinstance(resp, dict)
1023        return resp
1024
1025    async def fetch_profile(
1026        self,
1027        membership_id: int,
1028        type: enums.MembershipType | int,
1029        components: collections.Sequence[enums.ComponentType],
1030        auth: str | None = None,
1031    ) -> typedefs.JSONObject:
1032        collector = _collect_components(components)
1033        response = await self._request(
1034            _GET,
1035            f"Destiny2/{int(type)}/Profile/{membership_id}/?components={collector}",
1036            auth=auth,
1037        )
1038        assert isinstance(response, dict)
1039        return response
1040
1041    async def fetch_entity(self, type: str, hash: int) -> typedefs.JSONObject:
1042        response = await self._request(_GET, route=f"Destiny2/Manifest/{type}/{hash}")
1043        assert isinstance(response, dict)
1044        return response
1045
1046    async def fetch_inventory_item(self, hash: int, /) -> typedefs.JSONObject:
1047        resp = await self.fetch_entity("DestinyInventoryItemDefinition", hash)
1048        assert isinstance(resp, dict)
1049        return resp
1050
1051    async def fetch_objective_entity(self, hash: int, /) -> typedefs.JSONObject:
1052        resp = await self.fetch_entity("DestinyObjectiveDefinition", hash)
1053        assert isinstance(resp, dict)
1054        return resp
1055
1056    async def fetch_groups_for_member(
1057        self,
1058        member_id: int,
1059        member_type: enums.MembershipType | int,
1060        /,
1061        *,
1062        filter: int = 0,
1063        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1064    ) -> typedefs.JSONObject:
1065        resp = await self._request(
1066            _GET,
1067            f"GroupV2/User/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1068        )
1069        assert isinstance(resp, dict)
1070        return resp
1071
1072    async def fetch_potential_groups_for_member(
1073        self,
1074        member_id: int,
1075        member_type: enums.MembershipType | int,
1076        /,
1077        *,
1078        filter: int = 0,
1079        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1080    ) -> typedefs.JSONObject:
1081        resp = await self._request(
1082            _GET,
1083            f"GroupV2/User/Potential/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1084        )
1085        assert isinstance(resp, dict)
1086        return resp
1087
1088    async def fetch_clan_members(
1089        self,
1090        clan_id: int,
1091        /,
1092        *,
1093        name: str | None = None,
1094        type: enums.MembershipType | int = enums.MembershipType.NONE,
1095    ) -> typedefs.JSONObject:
1096        resp = await self._request(
1097            _GET,
1098            f"/GroupV2/{clan_id}/Members/?memberType={int(type)}&nameSearch={name if name else ''}&currentpage=1",
1099        )
1100        assert isinstance(resp, dict)
1101        return resp
1102
1103    async def fetch_hardlinked_credentials(
1104        self,
1105        credential: int,
1106        type: enums.CredentialType | int = enums.CredentialType.STEAMID,
1107        /,
1108    ) -> typedefs.JSONObject:
1109        resp = await self._request(
1110            _GET,
1111            f"User/GetMembershipFromHardLinkedCredential/{int(type)}/{credential}/",
1112        )
1113        assert isinstance(resp, dict)
1114        return resp
1115
1116    async def fetch_user_credentials(
1117        self, access_token: str, membership_id: int, /
1118    ) -> typedefs.JSONArray:
1119        resp = await self._request(
1120            _GET,
1121            f"User/GetCredentialTypesForTargetAccount/{membership_id}",
1122            auth=access_token,
1123        )
1124        assert isinstance(resp, list)
1125        return resp
1126
1127    async def insert_socket_plug(
1128        self,
1129        action_token: str,
1130        /,
1131        instance_id: int,
1132        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1133        character_id: int,
1134        membership_type: enums.MembershipType | int,
1135    ) -> typedefs.JSONObject:
1136        if isinstance(plug, builders.PlugSocketBuilder):
1137            plug = plug.collect()
1138
1139        body = {
1140            "actionToken": action_token,
1141            "itemInstanceId": instance_id,
1142            "plug": plug,
1143            "characterId": character_id,
1144            "membershipType": int(membership_type),
1145        }
1146        resp = await self._request(
1147            _POST, "Destiny2/Actions/Items/InsertSocketPlug", json=body
1148        )
1149        assert isinstance(resp, dict)
1150        return resp
1151
1152    async def insert_socket_plug_free(
1153        self,
1154        access_token: str,
1155        /,
1156        instance_id: int,
1157        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1158        character_id: int,
1159        membership_type: enums.MembershipType | int,
1160    ) -> typedefs.JSONObject:
1161        if isinstance(plug, builders.PlugSocketBuilder):
1162            plug = plug.collect()
1163
1164        body = {
1165            "itemInstanceId": instance_id,
1166            "plug": plug,
1167            "characterId": character_id,
1168            "membershipType": int(membership_type),
1169        }
1170        resp = await self._request(
1171            _POST,
1172            "Destiny2/Actions/Items/InsertSocketPlugFree",
1173            json=body,
1174            auth=access_token,
1175        )
1176        assert isinstance(resp, dict)
1177        return resp
1178
1179    @helpers.unstable
1180    async def set_item_lock_state(
1181        self,
1182        access_token: str,
1183        state: bool,
1184        /,
1185        item_id: int,
1186        character_id: int,
1187        membership_type: enums.MembershipType | int,
1188    ) -> int:
1189        body = {
1190            "state": state,
1191            "itemId": item_id,
1192            "characterId": character_id,
1193            "membershipType": int(membership_type),
1194        }
1195        response = await self._request(
1196            _POST,
1197            "Destiny2/Actions/Items/SetLockState",
1198            json=body,
1199            auth=access_token,
1200        )
1201        assert isinstance(response, int)
1202        return response
1203
1204    async def set_quest_track_state(
1205        self,
1206        access_token: str,
1207        state: bool,
1208        /,
1209        item_id: int,
1210        character_id: int,
1211        membership_type: enums.MembershipType | int,
1212    ) -> int:
1213        body = {
1214            "state": state,
1215            "itemId": item_id,
1216            "characterId": character_id,
1217            "membership_type": int(membership_type),
1218        }
1219        response = await self._request(
1220            _POST,
1221            "Destiny2/Actions/Items/SetTrackedState",
1222            json=body,
1223            auth=access_token,
1224        )
1225        assert isinstance(response, int)
1226        return response
1227
1228    async def fetch_manifest_path(self) -> typedefs.JSONObject:
1229        path = await self._request(_GET, "Destiny2/Manifest")
1230        assert isinstance(path, dict)
1231        return path
1232
1233    async def read_manifest_bytes(self, language: _ALLOWED_LANGS = "en", /) -> bytes:
1234        _ensure_manifest_language(language)
1235
1236        content = await self.fetch_manifest_path()
1237        resp = await self._request(
1238            _GET,
1239            content["mobileWorldContentPaths"][language],
1240            unwrap_bytes=True,
1241            base=True,
1242        )
1243        assert isinstance(resp, bytes)
1244        return resp
1245
1246    async def download_sqlite_manifest(
1247        self,
1248        language: _ALLOWED_LANGS = "en",
1249        name: str = "manifest",
1250        path: pathlib.Path | str = ".",
1251        *,
1252        force: bool = False,
1253        executor: concurrent.futures.Executor | None = None,
1254    ) -> pathlib.Path:
1255        complete_path = _get_path(name, path, sql=True)
1256
1257        if complete_path.exists():
1258            if force:
1259                _LOGGER.info(
1260                    f"Found manifest in {complete_path!s}. Forcing to Re-Download."
1261                )
1262                complete_path.unlink(missing_ok=True)
1263
1264                return await self.download_sqlite_manifest(
1265                    language, name, path, force=force
1266                )
1267
1268            else:
1269                raise FileExistsError(
1270                    "Manifest file already exists, "
1271                    "To force download, set the `force` parameter to `True`."
1272                )
1273
1274        _LOGGER.info(f"Downloading manifest. Location: {complete_path!s}")
1275        data_bytes = await self.read_manifest_bytes(language)
1276        await asyncio.get_running_loop().run_in_executor(
1277            executor, _write_sqlite_bytes, data_bytes, path, name
1278        )
1279        _LOGGER.info("Finished downloading manifest.")
1280        return _get_path(name, path, sql=True)
1281
1282    async def download_json_manifest(
1283        self,
1284        file_name: str = "manifest",
1285        path: str | pathlib.Path = ".",
1286        *,
1287        language: _ALLOWED_LANGS = "en",
1288        executor: concurrent.futures.Executor | None = None,
1289    ) -> pathlib.Path:
1290        _ensure_manifest_language(language)
1291        full_path = _get_path(file_name, path)
1292        _LOGGER.info(f"Downloading manifest JSON to {full_path!r}...")
1293
1294        content = await self.fetch_manifest_path()
1295        json_bytes = await self._request(
1296            _GET,
1297            content["jsonWorldContentPaths"][language],
1298            unwrap_bytes=True,
1299            base=True,
1300        )
1301
1302        assert isinstance(json_bytes, bytes)
1303        await asyncio.get_running_loop().run_in_executor(
1304            executor, _write_json_bytes, json_bytes, file_name, path
1305        )
1306        _LOGGER.info("Finished downloading manifest JSON.")
1307        return full_path
1308
1309    async def fetch_manifest_version(self) -> str:
1310        # This is guaranteed str.
1311        return (await self.fetch_manifest_path())["version"]
1312
1313    async def fetch_linked_profiles(
1314        self,
1315        member_id: int,
1316        member_type: enums.MembershipType | int,
1317        /,
1318        *,
1319        all: bool = False,
1320    ) -> typedefs.JSONObject:
1321        resp = await self._request(
1322            _GET,
1323            f"Destiny2/{int(member_type)}/Profile/{member_id}/LinkedProfiles/?getAllMemberships={all}",
1324        )
1325        assert isinstance(resp, dict)
1326        return resp
1327
1328    async def fetch_clan_banners(self) -> typedefs.JSONObject:
1329        resp = await self._request(_GET, "Destiny2/Clan/ClanBannerDictionary/")
1330        assert isinstance(resp, dict)
1331        return resp
1332
1333    async def fetch_public_milestones(self) -> typedefs.JSONObject:
1334        resp = await self._request(_GET, "Destiny2/Milestones/")
1335        assert isinstance(resp, dict)
1336        return resp
1337
1338    async def fetch_public_milestone_content(
1339        self, milestone_hash: int, /
1340    ) -> typedefs.JSONObject:
1341        resp = await self._request(
1342            _GET, f"Destiny2/Milestones/{milestone_hash}/Content/"
1343        )
1344        assert isinstance(resp, dict)
1345        return resp
1346
1347    async def fetch_current_user_memberships(
1348        self, access_token: str, /
1349    ) -> typedefs.JSONObject:
1350        resp = await self._request(
1351            _GET,
1352            "User/GetMembershipsForCurrentUser/",
1353            auth=access_token,
1354        )
1355        assert isinstance(resp, dict)
1356        return resp
1357
1358    async def equip_item(
1359        self,
1360        access_token: str,
1361        /,
1362        item_id: int,
1363        character_id: int,
1364        membership_type: enums.MembershipType | int,
1365    ) -> None:
1366        payload = {
1367            "itemId": item_id,
1368            "characterId": character_id,
1369            "membershipType": int(membership_type),
1370        }
1371
1372        await self._request(
1373            _POST,
1374            "Destiny2/Actions/Items/EquipItem/",
1375            json=payload,
1376            auth=access_token,
1377        )
1378
1379    async def equip_items(
1380        self,
1381        access_token: str,
1382        /,
1383        item_ids: collections.Sequence[int],
1384        character_id: int,
1385        membership_type: enums.MembershipType | int,
1386    ) -> None:
1387        payload = {
1388            "itemIds": item_ids,
1389            "characterId": character_id,
1390            "membershipType": int(membership_type),
1391        }
1392        await self._request(
1393            _POST,
1394            "Destiny2/Actions/Items/EquipItems/",
1395            json=payload,
1396            auth=access_token,
1397        )
1398
1399    async def ban_clan_member(
1400        self,
1401        access_token: str,
1402        /,
1403        group_id: int,
1404        membership_id: int,
1405        membership_type: enums.MembershipType | int,
1406        *,
1407        length: int = 0,
1408        comment: str | None = None,
1409    ) -> None:
1410        payload = {"comment": str(comment), "length": length}
1411        await self._request(
1412            _POST,
1413            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Ban/",
1414            json=payload,
1415            auth=access_token,
1416        )
1417
1418    async def unban_clan_member(
1419        self,
1420        access_token: str,
1421        /,
1422        group_id: int,
1423        membership_id: int,
1424        membership_type: enums.MembershipType | int,
1425    ) -> None:
1426        await self._request(
1427            _POST,
1428            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Unban/",
1429            auth=access_token,
1430        )
1431
1432    async def kick_clan_member(
1433        self,
1434        access_token: str,
1435        /,
1436        group_id: int,
1437        membership_id: int,
1438        membership_type: enums.MembershipType | int,
1439    ) -> typedefs.JSONObject:
1440        resp = await self._request(
1441            _POST,
1442            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Kick/",
1443            auth=access_token,
1444        )
1445        assert isinstance(resp, dict)
1446        return resp
1447
1448    async def edit_clan(
1449        self,
1450        access_token: str,
1451        /,
1452        group_id: int,
1453        *,
1454        name: str | None = None,
1455        about: str | None = None,
1456        motto: str | None = None,
1457        theme: str | None = None,
1458        tags: collections.Sequence[str] | None = None,
1459        is_public: bool | None = None,
1460        locale: str | None = None,
1461        avatar_image_index: int | None = None,
1462        membership_option: enums.MembershipOption | int | None = None,
1463        allow_chat: bool | None = None,
1464        chat_security: typing.Literal[0, 1] | None = None,
1465        call_sign: str | None = None,
1466        homepage: typing.Literal[0, 1, 2] | None = None,
1467        enable_invite_messaging_for_admins: bool | None = None,
1468        default_publicity: typing.Literal[0, 1, 2] | None = None,
1469        is_public_topic_admin: bool | None = None,
1470    ) -> None:
1471        payload = {
1472            "name": name,
1473            "about": about,
1474            "motto": motto,
1475            "theme": theme,
1476            "tags": tags,
1477            "isPublic": is_public,
1478            "avatarImageIndex": avatar_image_index,
1479            "isPublicTopicAdminOnly": is_public_topic_admin,
1480            "allowChat": allow_chat,
1481            "chatSecurity": chat_security,
1482            "callsign": call_sign,
1483            "homepage": homepage,
1484            "enableInvitationMessagingForAdmins": enable_invite_messaging_for_admins,
1485            "defaultPublicity": default_publicity,
1486            "locale": locale,
1487        }
1488        if membership_option is not None:
1489            payload["membershipOption"] = int(membership_option)
1490
1491        await self._request(
1492            _POST,
1493            f"GroupV2/{group_id}/Edit",
1494            json=payload,
1495            auth=access_token,
1496        )
1497
1498    async def edit_clan_options(
1499        self,
1500        access_token: str,
1501        /,
1502        group_id: int,
1503        *,
1504        invite_permissions_override: bool | None = None,
1505        update_culture_permissionOverride: bool | None = None,
1506        host_guided_game_permission_override: typing.Literal[0, 1, 2] | None = None,
1507        update_banner_permission_override: bool | None = None,
1508        join_level: enums.ClanMemberType | int | None = None,
1509    ) -> None:
1510        payload = {
1511            "InvitePermissionOverride": invite_permissions_override,
1512            "UpdateCulturePermissionOverride": update_culture_permissionOverride,
1513            "HostGuidedGamePermissionOverride": host_guided_game_permission_override,
1514            "UpdateBannerPermissionOverride": update_banner_permission_override,
1515            "JoinLevel": int(join_level) if join_level else None,
1516        }
1517
1518        await self._request(
1519            _POST,
1520            f"GroupV2/{group_id}/EditFounderOptions",
1521            json=payload,
1522            auth=access_token,
1523        )
1524
1525    async def report_player(
1526        self,
1527        access_token: str,
1528        /,
1529        activity_id: int,
1530        character_id: int,
1531        reason_hashes: collections.Sequence[int],
1532        reason_category_hashes: collections.Sequence[int],
1533    ) -> None:
1534        await self._request(
1535            _POST,
1536            f"Destiny2/Stats/PostGameCarnageReport/{activity_id}/Report/",
1537            json={
1538                "reasonCategoryHashes": reason_category_hashes,
1539                "reasonHashes": reason_hashes,
1540                "offendingCharacterId": character_id,
1541            },
1542            auth=access_token,
1543        )
1544
1545    async def fetch_friends(self, access_token: str, /) -> typedefs.JSONObject:
1546        resp = await self._request(
1547            _GET,
1548            "Social/Friends/",
1549            auth=access_token,
1550        )
1551        assert isinstance(resp, dict)
1552        return resp
1553
1554    async def fetch_friend_requests(self, access_token: str, /) -> typedefs.JSONObject:
1555        resp = await self._request(
1556            _GET,
1557            "Social/Friends/Requests",
1558            auth=access_token,
1559        )
1560        assert isinstance(resp, dict)
1561        return resp
1562
1563    async def accept_friend_request(self, access_token: str, /, member_id: int) -> None:
1564        await self._request(
1565            _POST,
1566            f"Social/Friends/Requests/Accept/{member_id}",
1567            auth=access_token,
1568        )
1569
1570    async def send_friend_request(self, access_token: str, /, member_id: int) -> None:
1571        await self._request(
1572            _POST,
1573            f"Social/Friends/Add/{member_id}",
1574            auth=access_token,
1575        )
1576
1577    async def decline_friend_request(
1578        self, access_token: str, /, member_id: int
1579    ) -> None:
1580        await self._request(
1581            _POST,
1582            f"Social/Friends/Requests/Decline/{member_id}",
1583            auth=access_token,
1584        )
1585
1586    async def remove_friend(self, access_token: str, /, member_id: int) -> None:
1587        await self._request(
1588            _POST,
1589            f"Social/Friends/Remove/{member_id}",
1590            auth=access_token,
1591        )
1592
1593    async def remove_friend_request(self, access_token: str, /, member_id: int) -> None:
1594        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1595        await self._request(
1596            _POST,
1597            f"Social/Friends/Requests/Remove/{member_id}",
1598            auth=access_token,
1599        )
1600
1601    async def approve_all_pending_group_users(
1602        self,
1603        access_token: str,
1604        /,
1605        group_id: int,
1606        message: str | None = None,
1607    ) -> None:
1608        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1609        await self._request(
1610            _POST,
1611            f"GroupV2/{group_id}/Members/ApproveAll",
1612            auth=access_token,
1613            json={"message": str(message)},
1614        )
1615
1616    async def deny_all_pending_group_users(
1617        self,
1618        access_token: str,
1619        /,
1620        group_id: int,
1621        *,
1622        message: str | None = None,
1623    ) -> None:
1624        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1625        await self._request(
1626            _POST,
1627            f"GroupV2/{group_id}/Members/DenyAll",
1628            auth=access_token,
1629            json={"message": str(message)},
1630        )
1631
1632    async def add_optional_conversation(
1633        self,
1634        access_token: str,
1635        /,
1636        group_id: int,
1637        *,
1638        name: str | None = None,
1639        security: typing.Literal[0, 1] = 0,
1640    ) -> None:
1641        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1642        payload = {"chatName": str(name), "chatSecurity": security}
1643        await self._request(
1644            _POST,
1645            f"GroupV2/{group_id}/OptionalConversations/Add",
1646            json=payload,
1647            auth=access_token,
1648        )
1649
1650    async def edit_optional_conversation(
1651        self,
1652        access_token: str,
1653        /,
1654        group_id: int,
1655        conversation_id: int,
1656        *,
1657        name: str | None = None,
1658        security: typing.Literal[0, 1] = 0,
1659        enable_chat: bool = False,
1660    ) -> None:
1661        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1662        payload = {
1663            "chatEnabled": enable_chat,
1664            "chatName": str(name),
1665            "chatSecurity": security,
1666        }
1667        await self._request(
1668            _POST,
1669            f"GroupV2/{group_id}/OptionalConversations/Edit/{conversation_id}",
1670            json=payload,
1671            auth=access_token,
1672        )
1673
1674    async def transfer_item(
1675        self,
1676        access_token: str,
1677        /,
1678        item_id: int,
1679        item_hash: int,
1680        character_id: int,
1681        member_type: enums.MembershipType | int,
1682        *,
1683        stack_size: int = 1,
1684        vault: bool = False,
1685    ) -> None:
1686        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1687        payload = {
1688            "characterId": character_id,
1689            "membershipType": int(member_type),
1690            "itemId": item_id,
1691            "itemReferenceHash": item_hash,
1692            "stackSize": stack_size,
1693            "transferToVault": vault,
1694        }
1695        await self._request(
1696            _POST,
1697            "Destiny2/Actions/Items/TransferItem",
1698            json=payload,
1699            auth=access_token,
1700        )
1701
1702    async def pull_item(
1703        self,
1704        access_token: str,
1705        /,
1706        item_id: int,
1707        item_hash: int,
1708        character_id: int,
1709        member_type: enums.MembershipType | int,
1710        *,
1711        stack_size: int = 1,
1712        vault: bool = False,
1713    ) -> None:
1714        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1715        payload = {
1716            "characterId": character_id,
1717            "membershipType": int(member_type),
1718            "itemId": item_id,
1719            "itemReferenceHash": item_hash,
1720            "stackSize": stack_size,
1721        }
1722        await self._request(
1723            _POST,
1724            "Destiny2/Actions/Items/PullFromPostmaster",
1725            json=payload,
1726            auth=access_token,
1727        )
1728        if vault:
1729            await self.transfer_item(
1730                access_token,
1731                item_id=item_id,
1732                item_hash=item_hash,
1733                character_id=character_id,
1734                member_type=member_type,
1735                stack_size=stack_size,
1736                vault=True,
1737            )
1738
1739    @helpers.unstable
1740    async def fetch_fireteams(
1741        self,
1742        activity_type: fireteams.FireteamActivity | int,
1743        *,
1744        platform: fireteams.FireteamPlatform | int = fireteams.FireteamPlatform.ANY,
1745        language: fireteams.FireteamLanguage | str = fireteams.FireteamLanguage.ALL,
1746        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1747        page: int = 0,
1748        slots_filter: int = 0,
1749    ) -> typedefs.JSONObject:
1750        resp = await self._request(
1751            _GET,
1752            f"Fireteam/Search/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{page}/?langFilter={str(language)}",  # noqa: E501 Line too long
1753        )
1754        assert isinstance(resp, dict)
1755        return resp
1756
1757    async def fetch_available_clan_fireteams(
1758        self,
1759        access_token: str,
1760        group_id: int,
1761        activity_type: fireteams.FireteamActivity | int,
1762        *,
1763        platform: fireteams.FireteamPlatform | int,
1764        language: fireteams.FireteamLanguage | str,
1765        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1766        page: int = 0,
1767        public_only: bool = False,
1768        slots_filter: int = 0,
1769    ) -> typedefs.JSONObject:
1770        resp = await self._request(
1771            _GET,
1772            f"Fireteam/Clan/{group_id}/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{public_only}/{page}",  # noqa: E501
1773            json={"langFilter": str(language)},
1774            auth=access_token,
1775        )
1776        assert isinstance(resp, dict)
1777        return resp
1778
1779    async def fetch_clan_fireteam(
1780        self, access_token: str, fireteam_id: int, group_id: int
1781    ) -> typedefs.JSONObject:
1782        resp = await self._request(
1783            _GET,
1784            f"Fireteam/Clan/{group_id}/Summary/{fireteam_id}",
1785            auth=access_token,
1786        )
1787        assert isinstance(resp, dict)
1788        return resp
1789
1790    async def fetch_my_clan_fireteams(
1791        self,
1792        access_token: str,
1793        group_id: int,
1794        *,
1795        include_closed: bool = True,
1796        platform: fireteams.FireteamPlatform | int,
1797        language: fireteams.FireteamLanguage | str,
1798        filtered: bool = True,
1799        page: int = 0,
1800    ) -> typedefs.JSONObject:
1801        payload = {"groupFilter": filtered, "langFilter": str(language)}
1802
1803        resp = await self._request(
1804            _GET,
1805            f"Fireteam/Clan/{group_id}/My/{int(platform)}/{include_closed}/{page}",
1806            json=payload,
1807            auth=access_token,
1808        )
1809        assert isinstance(resp, dict)
1810        return resp
1811
1812    async def fetch_private_clan_fireteams(
1813        self, access_token: str, group_id: int, /
1814    ) -> int:
1815        resp = await self._request(
1816            _GET,
1817            f"Fireteam/Clan/{group_id}/ActiveCount",
1818            auth=access_token,
1819        )
1820        assert isinstance(resp, int)
1821        return resp
1822
1823    async def fetch_post_activity(self, instance_id: int, /) -> typedefs.JSONObject:
1824        resp = await self._request(
1825            _GET, f"Destiny2/Stats/PostGameCarnageReport/{instance_id}"
1826        )
1827        assert isinstance(resp, dict)
1828        return resp
1829
1830    @helpers.unstable
1831    async def search_entities(
1832        self, name: str, entity_type: str, *, page: int = 0
1833    ) -> typedefs.JSONObject:
1834        resp = await self._request(
1835            _GET,
1836            f"Destiny2/Armory/Search/{entity_type}/{name}/",
1837            json={"page": page},
1838        )
1839        assert isinstance(resp, dict)
1840        return resp
1841
1842    async def fetch_unique_weapon_history(
1843        self,
1844        membership_id: int,
1845        character_id: int,
1846        membership_type: enums.MembershipType | int,
1847    ) -> typedefs.JSONObject:
1848        resp = await self._request(
1849            _GET,
1850            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/UniqueWeapons/",
1851        )
1852        assert isinstance(resp, dict)
1853        return resp
1854
1855    async def fetch_item(
1856        self,
1857        member_id: int,
1858        item_id: int,
1859        membership_type: enums.MembershipType | int,
1860        components: collections.Sequence[enums.ComponentType],
1861    ) -> typedefs.JSONObject:
1862        collector = _collect_components(components)
1863
1864        resp = await self._request(
1865            _GET,
1866            f"Destiny2/{int(membership_type)}/Profile/{member_id}/Item/{item_id}/?components={collector}",
1867        )
1868        assert isinstance(resp, dict)
1869        return resp
1870
1871    async def fetch_clan_weekly_rewards(self, clan_id: int, /) -> typedefs.JSONObject:
1872        resp = await self._request(_GET, f"Destiny2/Clan/{clan_id}/WeeklyRewardState/")
1873        assert isinstance(resp, dict)
1874        return resp
1875
1876    async def fetch_available_locales(self) -> typedefs.JSONObject:
1877        resp = await self._request(_GET, "Destiny2/Manifest/DestinyLocaleDefinition/")
1878        assert isinstance(resp, dict)
1879        return resp
1880
1881    async def fetch_common_settings(self) -> typedefs.JSONObject:
1882        resp = await self._request(_GET, "Settings")
1883        assert isinstance(resp, dict)
1884        return resp
1885
1886    async def fetch_user_systems_overrides(self) -> typedefs.JSONObject:
1887        resp = await self._request(_GET, "UserSystemOverrides")
1888        assert isinstance(resp, dict)
1889        return resp
1890
1891    async def fetch_global_alerts(
1892        self, *, include_streaming: bool = False
1893    ) -> typedefs.JSONArray:
1894        resp = await self._request(
1895            _GET, f"GlobalAlerts/?includestreaming={include_streaming}"
1896        )
1897        assert isinstance(resp, list)
1898        return resp
1899
1900    async def awainitialize_request(
1901        self,
1902        access_token: str,
1903        type: typing.Literal[0, 1],
1904        membership_type: enums.MembershipType | int,
1905        /,
1906        *,
1907        affected_item_id: int | None = None,
1908        character_id: int | None = None,
1909    ) -> typedefs.JSONObject:
1910        body = {"type": type, "membershipType": int(membership_type)}
1911
1912        if affected_item_id is not None:
1913            body["affectedItemId"] = affected_item_id
1914
1915        if character_id is not None:
1916            body["characterId"] = character_id
1917
1918        resp = await self._request(
1919            _POST, "Destiny2/Awa/Initialize", json=body, auth=access_token
1920        )
1921        assert isinstance(resp, dict)
1922        return resp
1923
1924    async def awaget_action_token(
1925        self, access_token: str, correlation_id: str, /
1926    ) -> typedefs.JSONObject:
1927        resp = await self._request(
1928            _POST,
1929            f"Destiny2/Awa/GetActionToken/{correlation_id}",
1930            auth=access_token,
1931        )
1932        assert isinstance(resp, dict)
1933        return resp
1934
1935    async def awa_provide_authorization_result(
1936        self,
1937        access_token: str,
1938        selection: int,
1939        correlation_id: str,
1940        nonce: collections.MutableSequence[str | bytes],
1941    ) -> int:
1942        body = {"selection": selection, "correlationId": correlation_id, "nonce": nonce}
1943
1944        resp = await self._request(
1945            _POST,
1946            "Destiny2/Awa/AwaProvideAuthorizationResult",
1947            json=body,
1948            auth=access_token,
1949        )
1950        assert isinstance(resp, int)
1951        return resp
1952
1953    async def fetch_vendors(
1954        self,
1955        access_token: str,
1956        character_id: int,
1957        membership_id: int,
1958        membership_type: enums.MembershipType | int,
1959        /,
1960        components: collections.Sequence[enums.ComponentType],
1961        filter: int | None = None,
1962    ) -> typedefs.JSONObject:
1963        components_ = _collect_components(components)
1964        route = (
1965            f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1966            f"/Character/{character_id}/Vendors/?components={components_}"
1967        )
1968
1969        if filter is not None:
1970            route = route + f"&filter={filter}"
1971
1972        resp = await self._request(
1973            _GET,
1974            route,
1975            auth=access_token,
1976        )
1977        assert isinstance(resp, dict)
1978        return resp
1979
1980    async def fetch_vendor(
1981        self,
1982        access_token: str,
1983        character_id: int,
1984        membership_id: int,
1985        membership_type: enums.MembershipType | int,
1986        vendor_hash: int,
1987        /,
1988        components: collections.Sequence[enums.ComponentType],
1989    ) -> typedefs.JSONObject:
1990        components_ = _collect_components(components)
1991        resp = await self._request(
1992            _GET,
1993            (
1994                f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1995                f"/Character/{character_id}/Vendors/{vendor_hash}/?components={components_}"
1996            ),
1997            auth=access_token,
1998        )
1999        assert isinstance(resp, dict)
2000        return resp
2001
2002    async def fetch_application_api_usage(
2003        self,
2004        access_token: str,
2005        application_id: int,
2006        /,
2007        *,
2008        start: datetime.datetime | None = None,
2009        end: datetime.datetime | None = None,
2010    ) -> typedefs.JSONObject:
2011        end_date, start_date = time.parse_date_range(end, start)
2012        resp = await self._request(
2013            _GET,
2014            f"App/ApiUsage/{application_id}/?end={end_date}&start={start_date}",
2015            auth=access_token,
2016        )
2017        assert isinstance(resp, dict)
2018        return resp
2019
2020    async def fetch_bungie_applications(self) -> typedefs.JSONArray:
2021        resp = await self._request(_GET, "App/FirstParty")
2022        assert isinstance(resp, list)
2023        return resp
2024
2025    async def fetch_content_type(self, type: str, /) -> typedefs.JSONObject:
2026        resp = await self._request(_GET, f"Content/GetContentType/{type}/")
2027        assert isinstance(resp, dict)
2028        return resp
2029
2030    async def fetch_content_by_id(
2031        self, id: int, locale: str, /, *, head: bool = False
2032    ) -> typedefs.JSONObject:
2033        resp = await self._request(
2034            _GET,
2035            f"Content/GetContentById/{id}/{locale}/",
2036            json={"head": head},
2037        )
2038        assert isinstance(resp, dict)
2039        return resp
2040
2041    async def fetch_content_by_tag_and_type(
2042        self, locale: str, tag: str, type: str, *, head: bool = False
2043    ) -> typedefs.JSONObject:
2044        resp = await self._request(
2045            _GET,
2046            f"Content/GetContentByTagAndType/{tag}/{type}/{locale}/",
2047            json={"head": head},
2048        )
2049        assert isinstance(resp, dict)
2050        return resp
2051
2052    async def search_content_with_text(
2053        self,
2054        locale: str,
2055        /,
2056        content_type: str,
2057        search_text: str,
2058        tag: str,
2059        *,
2060        page: int | None = None,
2061        source: str | None = None,
2062    ) -> typedefs.JSONObject:
2063        body: typedefs.JSONObject = {
2064            "locale": locale,
2065            "currentpage": page or 1,
2066            "ctype": content_type,
2067            "searchtxt": search_text,
2068            "searchtext": search_text,
2069            "tag": tag,
2070            "source": source,
2071        }
2072
2073        resp = await self._request(_GET, "Content/Search", params=body)
2074        assert isinstance(resp, dict)
2075        return resp
2076
2077    async def search_content_by_tag_and_type(
2078        self,
2079        locale: str,
2080        tag: str,
2081        type: str,
2082        *,
2083        page: int | None = None,
2084    ) -> typedefs.JSONObject:
2085        body: typedefs.JSONObject = {"currentpage": page or 1}
2086
2087        resp = await self._request(
2088            _GET,
2089            f"Content/SearchContentByTagAndType/{tag}/{type}/{locale}/",
2090            params=body,
2091        )
2092        assert isinstance(resp, dict)
2093        return resp
2094
2095    async def search_help_articles(
2096        self, text: str, size: str, /
2097    ) -> typedefs.JSONObject:
2098        resp = await self._request(_GET, f"Content/SearchHelpArticles/{text}/{size}/")
2099        assert isinstance(resp, dict)
2100        return resp
2101
2102    async def fetch_topics_page(
2103        self,
2104        category_filter: int,
2105        group: int,
2106        date_filter: int,
2107        sort: str | bytes,
2108        *,
2109        page: int | None = None,
2110        locales: collections.Iterable[str] | None = None,
2111        tag_filter: str | None = None,
2112    ) -> typedefs.JSONObject:
2113        params = {
2114            "locales": ",".join(locales) if locales is not None else "en",
2115        }
2116        if tag_filter:
2117            params["tagstring"] = tag_filter
2118
2119        resp = await self._request(
2120            _GET,
2121            f"Forum/GetTopicsPaged/{page or 0}/0/{group}/{sort!s}/{date_filter}/{category_filter}/",
2122            params=params,
2123        )
2124        assert isinstance(resp, dict)
2125        return resp
2126
2127    async def fetch_core_topics_page(
2128        self,
2129        category_filter: int,
2130        date_filter: int,
2131        sort: str | bytes,
2132        *,
2133        page: int | None = None,
2134        locales: collections.Iterable[str] | None = None,
2135    ) -> typedefs.JSONObject:
2136        resp = await self._request(
2137            _GET,
2138            f"Forum/GetCoreTopicsPaged/{page or 0}"
2139            f"/{sort!s}/{date_filter}/{category_filter}/?locales={','.join(locales) if locales else 'en'}",
2140        )
2141        assert isinstance(resp, dict)
2142        return resp
2143
2144    async def fetch_posts_threaded_page(
2145        self,
2146        parent_post: bool,
2147        page: int,
2148        page_size: int,
2149        parent_post_id: int,
2150        reply_size: int,
2151        root_thread_mode: bool,
2152        sort_mode: int,
2153        show_banned: str | None = None,
2154    ) -> typedefs.JSONObject:
2155        resp = await self._request(
2156            _GET,
2157            f"Forum/GetPostsThreadedPaged/{parent_post}/{page}/"
2158            f"{page_size}/{reply_size}/{parent_post_id}/{root_thread_mode}/{sort_mode}/",
2159            json={"showbanned": show_banned},
2160        )
2161        assert isinstance(resp, dict)
2162        return resp
2163
2164    async def fetch_posts_threaded_page_from_child(
2165        self,
2166        child_id: bool,
2167        page: int,
2168        page_size: int,
2169        reply_size: int,
2170        root_thread_mode: bool,
2171        sort_mode: int,
2172        show_banned: str | None = None,
2173    ) -> typedefs.JSONObject:
2174        resp = await self._request(
2175            _GET,
2176            f"Forum/GetPostsThreadedPagedFromChild/{child_id}/"
2177            f"{page}/{page_size}/{reply_size}/{root_thread_mode}/{sort_mode}/",
2178            json={"showbanned": show_banned},
2179        )
2180        assert isinstance(resp, dict)
2181        return resp
2182
2183    async def fetch_post_and_parent(
2184        self, child_id: int, /, *, show_banned: str | None = None
2185    ) -> typedefs.JSONObject:
2186        resp = await self._request(
2187            _GET,
2188            f"Forum/GetPostAndParent/{child_id}/",
2189            json={"showbanned": show_banned},
2190        )
2191        assert isinstance(resp, dict)
2192        return resp
2193
2194    async def fetch_posts_and_parent_awaiting(
2195        self, child_id: int, /, *, show_banned: str | None = None
2196    ) -> typedefs.JSONObject:
2197        resp = await self._request(
2198            _GET,
2199            f"Forum/GetPostAndParentAwaitingApproval/{child_id}/",
2200            json={"showbanned": show_banned},
2201        )
2202        assert isinstance(resp, dict)
2203        return resp
2204
2205    async def fetch_topic_for_content(self, content_id: int, /) -> int:
2206        resp = await self._request(_GET, f"Forum/GetTopicForContent/{content_id}/")
2207        assert isinstance(resp, int)
2208        return resp
2209
2210    async def fetch_forum_tag_suggestions(
2211        self, partial_tag: str, /
2212    ) -> typedefs.JSONObject:
2213        resp = await self._request(
2214            _GET,
2215            "Forum/GetForumTagSuggestions/",
2216            json={"partialtag": partial_tag},
2217        )
2218        assert isinstance(resp, dict)
2219        return resp
2220
2221    async def fetch_poll(self, topic_id: int, /) -> typedefs.JSONObject:
2222        resp = await self._request(_GET, f"Forum/Poll/{topic_id}/")
2223        assert isinstance(resp, dict)
2224        return resp
2225
2226    async def fetch_recruitment_thread_summaries(self) -> typedefs.JSONArray:
2227        resp = await self._request(_POST, "Forum/Recruit/Summaries/")
2228        assert isinstance(resp, list)
2229        return resp
2230
2231    async def fetch_recommended_groups(
2232        self,
2233        access_token: str,
2234        /,
2235        *,
2236        date_range: int = 0,
2237        group_type: enums.GroupType | int = enums.GroupType.CLAN,
2238    ) -> typedefs.JSONArray:
2239        resp = await self._request(
2240            _POST,
2241            f"GroupV2/Recommended/{int(group_type)}/{date_range}/",
2242            auth=access_token,
2243        )
2244        assert isinstance(resp, list)
2245        return resp
2246
2247    async def fetch_available_avatars(self) -> collections.Mapping[str, int]:
2248        resp = await self._request(_GET, "GroupV2/GetAvailableAvatars/")
2249        assert isinstance(resp, dict)
2250        return resp
2251
2252    async def fetch_user_clan_invite_setting(
2253        self,
2254        access_token: str,
2255        /,
2256        membership_type: enums.MembershipType | int,
2257    ) -> bool:
2258        resp = await self._request(
2259            _GET,
2260            f"GroupV2/GetUserClanInviteSetting/{int(membership_type)}/",
2261            auth=access_token,
2262        )
2263        assert isinstance(resp, bool)
2264        return resp
2265
2266    async def fetch_banned_group_members(
2267        self, access_token: str, group_id: int, /, *, page: int = 1
2268    ) -> typedefs.JSONObject:
2269        resp = await self._request(
2270            _GET,
2271            f"GroupV2/{group_id}/Banned/?currentpage={page}",
2272            auth=access_token,
2273        )
2274        assert isinstance(resp, dict)
2275        return resp
2276
2277    async def fetch_pending_group_memberships(
2278        self, access_token: str, group_id: int, /, *, current_page: int = 1
2279    ) -> typedefs.JSONObject:
2280        resp = await self._request(
2281            _GET,
2282            f"GroupV2/{group_id}/Members/Pending/?currentpage={current_page}",
2283            auth=access_token,
2284        )
2285        assert isinstance(resp, dict)
2286        return resp
2287
2288    async def fetch_invited_group_memberships(
2289        self, access_token: str, group_id: int, /, *, current_page: int = 1
2290    ) -> typedefs.JSONObject:
2291        resp = await self._request(
2292            _GET,
2293            f"GroupV2/{group_id}/Members/InvitedIndividuals/?currentpage={current_page}",
2294            auth=access_token,
2295        )
2296        assert isinstance(resp, dict)
2297        return resp
2298
2299    async def invite_member_to_group(
2300        self,
2301        access_token: str,
2302        /,
2303        group_id: int,
2304        membership_id: int,
2305        membership_type: enums.MembershipType | int,
2306        *,
2307        message: str | None = None,
2308    ) -> typedefs.JSONObject:
2309        resp = await self._request(
2310            _POST,
2311            f"GroupV2/{group_id}/Members/IndividualInvite/{int(membership_type)}/{membership_id}/",
2312            auth=access_token,
2313            json={"message": str(message)},
2314        )
2315        assert isinstance(resp, dict)
2316        return resp
2317
2318    async def cancel_group_member_invite(
2319        self,
2320        access_token: str,
2321        /,
2322        group_id: int,
2323        membership_id: int,
2324        membership_type: enums.MembershipType | int,
2325    ) -> typedefs.JSONObject:
2326        resp = await self._request(
2327            _POST,
2328            f"GroupV2/{group_id}/Members/IndividualInviteCancel/{int(membership_type)}/{membership_id}/",
2329            auth=access_token,
2330        )
2331        assert isinstance(resp, dict)
2332        return resp
2333
2334    async def fetch_historical_definition(self) -> typedefs.JSONObject:
2335        resp = await self._request(_GET, "Destiny2/Stats/Definition/")
2336        assert isinstance(resp, dict)
2337        return resp
2338
2339    async def fetch_historical_stats(
2340        self,
2341        character_id: int,
2342        membership_id: int,
2343        membership_type: enums.MembershipType | int,
2344        day_start: datetime.datetime,
2345        day_end: datetime.datetime,
2346        groups: collections.Sequence[enums.StatsGroupType | int],
2347        modes: collections.Sequence[enums.GameMode | int],
2348        *,
2349        period_type: enums.PeriodType = enums.PeriodType.ALL_TIME,
2350    ) -> typedefs.JSONObject:
2351        end, start = time.parse_date_range(day_end, day_start)
2352        resp = await self._request(
2353            _GET,
2354            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/",
2355            json={
2356                "dayend": end,
2357                "daystart": start,
2358                "groups": [str(int(group)) for group in groups],
2359                "modes": [str(int(mode)) for mode in modes],
2360                "periodType": int(period_type),
2361            },
2362        )
2363        assert isinstance(resp, dict)
2364        return resp
2365
2366    async def fetch_historical_stats_for_account(
2367        self,
2368        membership_id: int,
2369        membership_type: enums.MembershipType | int,
2370        groups: collections.Sequence[enums.StatsGroupType | int],
2371    ) -> typedefs.JSONObject:
2372        resp = await self._request(
2373            _GET,
2374            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Stats/",
2375            json={"groups": [str(int(group)) for group in groups]},
2376        )
2377        assert isinstance(resp, dict)
2378        return resp
2379
2380    async def fetch_aggregated_activity_stats(
2381        self,
2382        character_id: int,
2383        membership_id: int,
2384        membership_type: enums.MembershipType | int,
2385        /,
2386    ) -> typedefs.JSONObject:
2387        resp = await self._request(
2388            _GET,
2389            f"Destiny2/{int(membership_type)}/Account/{membership_id}/"
2390            f"Character/{character_id}/Stats/AggregateActivityStats/",
2391        )
2392        assert isinstance(resp, dict)
2393        return resp
2394
2395    async def equip_loadout(
2396        self,
2397        access_token: str,
2398        /,
2399        loadout_index: int,
2400        character_id: int,
2401        membership_type: enums.MembershipType | int,
2402    ) -> None:
2403        response = await self._request(
2404            _POST,
2405            "Destiny2/Actions/Loadouts/EquipLoadout/",
2406            json={
2407                "loadoutIndex": loadout_index,
2408                "characterId": character_id,
2409                "membership_type": int(membership_type),
2410            },
2411            auth=access_token,
2412        )
2413        assert isinstance(response, int)
2414
2415    async def snapshot_loadout(
2416        self,
2417        access_token: str,
2418        /,
2419        loadout_index: int,
2420        character_id: int,
2421        membership_type: enums.MembershipType | int,
2422        *,
2423        color_hash: int | None = None,
2424        icon_hash: int | None = None,
2425        name_hash: int | None = None,
2426    ) -> None:
2427        response = await self._request(
2428            _POST,
2429            "Destiny2/Actions/Loadouts/SnapshotLoadout/",
2430            auth=access_token,
2431            json={
2432                "colorHash": color_hash,
2433                "iconHash": icon_hash,
2434                "nameHash": name_hash,
2435                "loadoutIndex": loadout_index,
2436                "characterId": character_id,
2437                "membershipType": int(membership_type),
2438            },
2439        )
2440        assert isinstance(response, int)
2441
2442    async def update_loadout(
2443        self,
2444        access_token: str,
2445        /,
2446        loadout_index: int,
2447        character_id: int,
2448        membership_type: enums.MembershipType | int,
2449        *,
2450        color_hash: int | None = None,
2451        icon_hash: int | None = None,
2452        name_hash: int | None = None,
2453    ) -> None:
2454        response = await self._request(
2455            _POST,
2456            "Destiny2/Actions/Loadouts/UpdateLoadoutIdentifiers/",
2457            auth=access_token,
2458            json={
2459                "colorHash": color_hash,
2460                "iconHash": icon_hash,
2461                "nameHash": name_hash,
2462                "loadoutIndex": loadout_index,
2463                "characterId": character_id,
2464                "membershipType": int(membership_type),
2465            },
2466        )
2467        assert isinstance(response, int)
2468
2469    async def clear_loadout(
2470        self,
2471        access_token: str,
2472        /,
2473        loadout_index: int,
2474        character_id: int,
2475        membership_type: enums.MembershipType | int,
2476    ) -> None:
2477        response = await self._request(
2478            _POST,
2479            "Destiny2/Actions/Loadouts/ClearLoadout/",
2480            json={
2481                "loadoutIndex": loadout_index,
2482                "characterId": character_id,
2483                "membership_type": int(membership_type),
2484            },
2485            auth=access_token,
2486        )
2487        assert isinstance(response, int)
2488
2489    async def force_drops_repair(self, access_token: str, /) -> bool:
2490        response = await self._request(
2491            _POST, "Tokens/Partner/ForceDropsRepair/", auth=access_token
2492        )
2493        assert isinstance(response, bool)
2494        return response
2495
2496    async def claim_partner_offer(
2497        self,
2498        access_token: str,
2499        /,
2500        *,
2501        offer_id: str,
2502        bungie_membership_id: int,
2503        transaction_id: str,
2504    ) -> bool:
2505        response = await self._request(
2506            _POST,
2507            "Tokens/Partner/ClaimOffer/",
2508            json={
2509                "PartnerOfferId": offer_id,
2510                "BungieNetMembershipId": bungie_membership_id,
2511                "TransactionId": transaction_id,
2512            },
2513            auth=access_token,
2514        )
2515        assert isinstance(response, bool)
2516        return response
2517
2518    async def fetch_bungie_rewards_for_user(
2519        self, access_token: str, /, membership_id: int
2520    ) -> typedefs.JSONObject:
2521        response = await self._request(
2522            _GET,
2523            f"Tokens/Rewards/GetRewardsForUser/{membership_id}/",
2524            auth=access_token,
2525        )
2526        assert isinstance(response, dict)
2527        return response
2528
2529    async def fetch_bungie_rewards_for_platform(
2530        self,
2531        access_token: str,
2532        /,
2533        membership_id: int,
2534        membership_type: enums.MembershipType | int,
2535    ) -> typedefs.JSONObject:
2536        response = await self._request(
2537            _GET,
2538            f"Tokens/Rewards/GetRewardsForPlatformUser/{membership_id}/{int(membership_type)}",
2539            auth=access_token,
2540        )
2541        assert isinstance(response, dict)
2542        return response
2543
2544    async def fetch_bungie_rewards(self) -> typedefs.JSONObject:
2545        response = await self._request(_GET, "Tokens/Rewards/BungieRewards/")
2546        assert isinstance(response, dict)
2547        return response
2548
2549    async def fetch_fireteam_listing(self, listing_id: int) -> typedefs.JSONObject:
2550        response = await self._request(_GET, f"FireteamFinder/Listing/{listing_id}/")
2551        assert isinstance(response, dict)
2552        return response

A single process REST client implementation.

This client is designed to only make HTTP requests and return raw JSON objects.

Example

import aiobungie

client = aiobungie.RESTClient("TOKEN")
async with client:
    response = await client.fetch_clan_members(4389205)
    for member in response['results']:
        print(member['destinyUserInfo'])

Parameters

token (str): A valid application token from Bungie's developer portal.

Other Parameters

client_secret (str | None): An optional application client secret, This is only needed if you're fetching OAuth2 tokens with this client.
client_id (int | None): An optional application client id, This is only needed if you're fetching OAuth2 tokens with this client.
settings (aiobungie.builders.Settings | None): The client settings to use, if None the default will be used.
owned_client (bool):
- If set to True, this client will use the provided client_session parameter instead,
- If set to True and client_session is None, ValueError will be raised.
- If set to False, aiobungie will initialize a new client session for you.
client_session (aiohttp.ClientSession | None): If provided, this client session will be used to make all the HTTP requests. The owned_client must be set to True for this to work.
max_retries (int): The max retries number to retry if the request hit a 5xx status code.
debug (bool | str): Whether to enable logging responses or not.

Logging Levels

False: This will disable logging.
True: This will set the level to DEBUG and enable logging minimal information.
"TRACE" | aiobungie.TRACE: This will log the response headers along with the minimal information.

RESTClient( token: 'str', /, *, client_secret: 'str | None' = None, client_id: 'int | None' = None, settings: 'builders.Settings | None' = None, owned_client: 'bool' = True, client_session: 'aiohttp.ClientSession | None' = None, dumps: 'typedefs.Dumps' = <function dumps>, loads: 'typedefs.Loads' = <function loads>, max_retries: 'int' = 4, debug: "typing.Literal['TRACE'] | bool | int" = False) View Source

496    def __init__(
497        self,
498        token: str,
499        /,
500        *,
501        client_secret: str | None = None,
502        client_id: int | None = None,
503        settings: builders.Settings | None = None,
504        owned_client: bool = True,
505        client_session: aiohttp.ClientSession | None = None,
506        dumps: typedefs.Dumps = helpers.dumps,
507        loads: typedefs.Loads = helpers.loads,
508        max_retries: int = 4,
509        debug: typing.Literal["TRACE"] | bool | int = False,
510    ) -> None:
511        if owned_client is False and client_session is None:
512            raise ValueError(
513                "Expected an owned client session, but got `None`, Cannot have `owned_client` set to `False` and `client_session` to `None`"
514            )
515
516        self._settings = settings or builders.Settings()
517        self._session = client_session
518        self._owned_client = owned_client
519        self._lock: asyncio.Lock | None = None
520        self._client_secret = client_secret
521        self._client_id = client_id
522        self._token: str = token
523        self._max_retries = max_retries
524        self._dumps = dumps
525        self._loads = loads
526        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
527        self.with_debug(debug)

client_id: 'int | None' View Source

529    @property
530    def client_id(self) -> int | None:
531        return self._client_id

Return the client id of this REST client if provided, Otherwise None.

metadata: 'collections.MutableMapping[typing.Any, typing.Any]' View Source

533    @property
534    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
535        return self._metadata

A mutable mapping storage for the user's needs.

This mapping is useful for storing any kind of data that the user may need to access later from a different process.

Example

import aiobungie

client = aiobungie.RESTClient(…)

async with client:
    # Fetch auth tokens and store them
    client.metadata["tokens"] = await client.fetch_access_token("code")

# Some other time.
async with client:
    # Retrieve the tokens
    tokens: aiobungie.OAuth2Response = client.metadata["tokens"]

    # Use them to fetch your user.
    user = await client.fetch_current_user_memberships(tokens.access_token)

is_alive: 'bool' View Source

537    @property
538    def is_alive(self) -> bool:
539        return self._session is not None

Returns True if the REST client is alive and False otherwise.

settings: 'builders.Settings' View Source

541    @property
542    def settings(self) -> builders.Settings:
543        return self._settings

Internal client settings used within the HTTP client session.

async def close(self) -> 'None': View Source

545    async def close(self) -> None:
546        if self._session is None:
547            raise RuntimeError("REST client is not running.")
548
549        if self._owned_client:
550            await self._session.close()
551            self._session = None

Close this REST client session if it was acquired.

This method is automatically called when using async with contextmanager.

Raises

RuntimeError: If the client is already closed.

def open(self) -> 'None': View Source

553    def open(self) -> None:
554        """Open a new client session. This is called internally with contextmanager usage."""
555        if self.is_alive and self._owned_client:
556            raise RuntimeError("Cannot open REST client when it's already open.")
557
558        if self._owned_client:
559            self._session = aiohttp.ClientSession(
560                connector=aiohttp.TCPConnector(
561                    use_dns_cache=self._settings.use_dns_cache,
562                    ttl_dns_cache=self._settings.ttl_dns_cache,
563                    ssl_context=self._settings.ssl_context,
564                    ssl=self._settings.ssl,
565                ),
566                connector_owner=True,
567                raise_for_status=False,
568                timeout=self._settings.http_timeout,
569                trust_env=self._settings.trust_env,
570                headers=self._settings.headers,
571            )

Open a new client session. This is called internally with contextmanager usage.

@typing.final

async def static_request( self, method: '_HTTP_METHOD', path: 'str', *, auth: 'str | None' = None, json: 'collections.Mapping[str, typing.Any] | None' = None, params: 'collections.Mapping[str, typing.Any] | None' = None) -> 'typedefs.JSONIsh': View Source

573    @typing.final
574    async def static_request(
575        self,
576        method: _HTTP_METHOD,
577        path: str,
578        *,
579        auth: str | None = None,
580        json: collections.Mapping[str, typing.Any] | None = None,
581        params: collections.Mapping[str, typing.Any] | None = None,
582    ) -> typedefs.JSONIsh:
583        return await self._request(method, path, auth=auth, json=json, params=params)

Perform an HTTP request given a valid Bungie endpoint.

This method allows you to freely perform HTTP requests to Bungie's API. It provides authentication support, JSON bodies, URL parameters and out of the box exception handling.

This method is useful for testing routes by yourself. or even calling routes that aiobungie doesn't support yet.

Parameters

method (str): The request method, This may be GET, POST, PUT, etc.
path (str): The Bungie endpoint or path. A path must look something like this Destiny2/3/Profile/46111239123/...

Other Parameters

auth (str | None): An optional bearer token for methods that requires OAuth2 Authorization header.
json (MutableMapping[str, typing.Any] | None): An optional JSON mapping to include in the request.
params (MutableMapping[str, typing.Any] | None): An optional URL query parameters mapping to include in the request.

Returns

aiobungie.typedefs.JSONIsh: The response payload.

@typing.final

def build_oauth2_url(self, client_id: 'int | None' = None) -> 'builders.OAuthURL | None': View Source

591    @typing.final
592    def build_oauth2_url(
593        self, client_id: int | None = None
594    ) -> builders.OAuthURL | None:
595        client_id = client_id or self._client_id
596        if client_id is None:
597            return None
598
599        return builders.OAuthURL(client_id=client_id)

Construct a new OAuthURL url object.

You can get the complete string representation of the url by calling .compile() on it.

Parameters

client_id (int | None): An optional client id to provide, If left None it will roll back to the id passed to the RESTClient, If both is None this method will return None.

Returns

aiobungie.builders.OAuthURL | None: * If client_id was provided as a parameter, It guarantees to return a complete OAuthURL object
- If client_id is set to aiobungie.RESTClient will be.
- If both are None this method will return `None.

async def fetch_oauth2_tokens(self, code: 'str', /) -> 'builders.OAuth2Response': View Source

822    async def fetch_oauth2_tokens(self, code: str, /) -> builders.OAuth2Response:
823        data = {
824            "grant_type": "authorization_code",
825            "code": code,
826            "client_id": self._client_id,
827            "client_secret": self._client_secret,
828        }
829
830        response = await self._request(_POST, "", data=data, oauth2=True)
831        assert isinstance(response, dict)
832        return builders.OAuth2Response.build_response(response)

Makes a POST request and fetch the OAuth2 access_token and refresh token.

Parameters

code (str): The Authorization code received from the authorization endpoint found in the URL parameters.

Returns

aiobungie.builders.OAuth2Response: An OAuth2 deserialized response.

Raises

aiobungie.error.Unauthorized: The passed code was invalid.

async def refresh_access_token(self, refresh_token: 'str', /) -> 'builders.OAuth2Response': View Source

834    async def refresh_access_token(
835        self, refresh_token: str, /
836    ) -> builders.OAuth2Response:
837        data = {
838            "grant_type": "refresh_token",
839            "refresh_token": refresh_token,
840            "client_id": self._client_id,
841            "client_secret": self._client_secret,
842        }
843
844        response = await self._request(_POST, "", data=data, oauth2=True)
845        assert isinstance(response, dict)
846        return builders.OAuth2Response.build_response(response)

Refresh OAuth2 access token given its refresh token.

Parameters

refresh_token (str): The refresh token.

Returns

aiobungie.builders.OAuth2Response: An OAuth2 deserialized response.

async def fetch_bungie_user(self, id: 'int') -> 'typedefs.JSONObject': View Source

848    async def fetch_bungie_user(self, id: int) -> typedefs.JSONObject:
849        resp = await self._request(_GET, f"User/GetBungieNetUserById/{id}/")
850        assert isinstance(resp, dict)
851        return resp

Fetch a Bungie user by their id.

Parameters

id (int): The user id.

Returns

aiobungie.typedefs.JSONObject: A JSON object of users objects.

Raises

aiobungie.error.NotFound: The user was not found.

async def fetch_user_themes(self) -> 'typedefs.JSONArray': View Source

853    async def fetch_user_themes(self) -> typedefs.JSONArray:
854        resp = await self._request(_GET, "User/GetAvailableThemes/")
855        assert isinstance(resp, list)
856        return resp

Fetch all available user themes.

Returns

aiobungie.typedefs.JSONArray: A JSON array of user themes.

async def fetch_membership_from_id( self, id: 'int', type: 'enums.MembershipType | int' = <MembershipType.NONE: 0>, /) -> 'typedefs.JSONObject': View Source

858    async def fetch_membership_from_id(
859        self,
860        id: int,
861        type: enums.MembershipType | int = enums.MembershipType.NONE,
862        /,
863    ) -> typedefs.JSONObject:
864        resp = await self._request(_GET, f"User/GetMembershipsById/{id}/{int(type)}")
865        assert isinstance(resp, dict)
866        return resp

Fetch Bungie user's memberships from their id.

Parameters

id (int): The user's id.
type (aiobungie.aiobungie.MembershipType | int): The user's membership type.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the found user.

Raises

aiobungie.NotFound: The requested user was not found.

async def fetch_membership( self, name: 'str', code: 'int', type: 'enums.MembershipType | int' = <MembershipType.ALL: -1>, /) -> 'typedefs.JSONArray': View Source

868    async def fetch_membership(
869        self,
870        name: str,
871        code: int,
872        type: enums.MembershipType | int = enums.MembershipType.ALL,
873        /,
874    ) -> typedefs.JSONArray:
875        resp = await self._request(
876            _POST,
877            f"Destiny2/SearchDestinyPlayerByBungieName/{int(type)}",
878            json={"displayName": name, "displayNameCode": code},
879        )
880        assert isinstance(resp, list)
881        return resp

Fetch a Destiny 2 Player.

Parameters

name (str): The unique Bungie player name.
code (int): The unique Bungie display name code.
type (aiobungie.aiobungie.MembershipType | int): The player's membership type, e,g. XBOX, STEAM, PSN

Returns

aiobungie.typedefs.JSONArray: A JSON array of the found player's memberships.

Raises

aiobungie.NotFound: The player was not found.
aiobungie.MembershipTypeError: The provided membership type was invalid.

async def fetch_sanitized_membership(self, membership_id: 'int', /) -> 'typedefs.JSONObject': View Source

883    async def fetch_sanitized_membership(
884        self, membership_id: int, /
885    ) -> typedefs.JSONObject:
886        response = await self._request(
887            _GET, f"User/GetSanitizedPlatformDisplayNames/{membership_id}/"
888        )
889        assert isinstance(response, dict)
890        return response

Fetch a list of all display names linked to membership_id, Which is profanity filtered.

Parameters

membership_id (int): The membership ID to fetch

Returns

aiobungie.typedefs.JSONObject: A JSON object contains all the available display names.

async def search_users(self, name: 'str', /) -> 'typedefs.JSONObject': View Source

892    async def search_users(self, name: str, /) -> typedefs.JSONObject:
893        resp = await self._request(
894            _POST,
895            "User/Search/GlobalName/0",
896            json={"displayNamePrefix": name},
897        )
898        assert isinstance(resp, dict)
899        return resp

Search for users by their global name and return all users who share this name.

Parameters

name (str): The user name.

Returns

aiobungie.typedefs.JSONObject: A JSON object of an array of the found users.

Raises

aiobungie.NotFound: The user(s) was not found.

async def fetch_clan_from_id( self, id: 'int', /, access_token: 'str | None' = None) -> 'typedefs.JSONObject': View Source

901    async def fetch_clan_from_id(
902        self, id: int, /, access_token: str | None = None
903    ) -> typedefs.JSONObject:
904        resp = await self._request(_GET, f"GroupV2/{id}", auth=access_token)
905        assert isinstance(resp, dict)
906        return resp

Fetch a Bungie Clan by its id.

Parameters

id (int): The clan id.

Other Parameters

access_token (str | None): An optional access token to make the request with.

If the token was bound to a member of the clan, This field aiobungie.crates.Clan.current_user_membership will be available and will return the membership of the user who made this request.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the clan.

Raises

aiobungie.NotFound: The clan was not found.

async def fetch_clan( self, name: 'str', /, access_token: 'str | None' = None, *, type: 'enums.GroupType | int' = <GroupType.CLAN: 1>) -> 'typedefs.JSONObject': View Source

908    async def fetch_clan(
909        self,
910        name: str,
911        /,
912        access_token: str | None = None,
913        *,
914        type: enums.GroupType | int = enums.GroupType.CLAN,
915    ) -> typedefs.JSONObject:
916        resp = await self._request(
917            _GET, f"GroupV2/Name/{name}/{int(type)}", auth=access_token
918        )
919        assert isinstance(resp, dict)
920        return resp

Fetch a Clan by its name. This method will return the first clan found with given name name.

Parameters

name (str): The clan name.

Other Parameters

access_token (str | None): An optional access token to make the request with.

If the token was bound to a member of the clan, This field aiobungie.crates.Clan.current_user_membership will be available and will return the membership of the user who made this request.
type (aiobungie.aiobungie.GroupType | int): The group type, Default is one.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the clan.

Raises

aiobungie.NotFound: The clan was not found.

async def search_group( self, name: 'str', group_type: 'enums.GroupType | int' = <GroupType.CLAN: 1>, *, creation_date: 'clans.GroupDate | int' = 0, sort_by: 'int | None' = None, group_member_count_filter: 'typing.Literal[0, 1, 2, 3] | None' = None, locale_filter: 'str | None' = None, tag_text: 'str | None' = None, items_per_page: 'int | None' = None, current_page: 'int | None' = None, request_token: 'str | None' = None) -> 'typedefs.JSONObject': View Source

922    async def search_group(
923        self,
924        name: str,
925        group_type: enums.GroupType | int = enums.GroupType.CLAN,
926        *,
927        creation_date: clans.GroupDate | int = 0,
928        sort_by: int | None = None,
929        group_member_count_filter: typing.Literal[0, 1, 2, 3] | None = None,
930        locale_filter: str | None = None,
931        tag_text: str | None = None,
932        items_per_page: int | None = None,
933        current_page: int | None = None,
934        request_token: str | None = None,
935    ) -> typedefs.JSONObject:
936        payload: collections.MutableMapping[str, typing.Any] = {"name": name}
937
938        # as the official documentation says, you're not allowed to use those fields
939        # on a clan search. it is safe to send the request with them being `null` but not filled with a value.
940        if (
941            group_type == enums.GroupType.CLAN
942            and group_member_count_filter is not None
943            and locale_filter
944            and tag_text
945        ):
946            raise ValueError(
947                "If you're searching for clans, (group_member_count_filter, locale_filter, tag_text) must be None."
948            )
949
950        payload["groupType"] = int(group_type)
951        payload["creationDate"] = int(creation_date)
952        payload["sortBy"] = sort_by
953        payload["groupMemberCount"] = group_member_count_filter
954        payload["locale"] = locale_filter
955        payload["tagText"] = tag_text
956        payload["itemsPerPage"] = items_per_page
957        payload["currentPage"] = current_page
958        payload["requestToken"] = request_token
959        payload["requestContinuationToken"] = request_token
960
961        resp = await self._request(_POST, "GroupV2/Search/", json=payload)
962        assert isinstance(resp, dict)
963        return resp

Search for groups.

If the group type is set to CLAN, then parameters group_member_count_filter, locale_filter and tag_text must be None, otherwise ValueError will be raised.

Parameters

name (str): The group name.
group_type (aiobungie.GroupType | int): The group type that's being searched for.

Other Parameters

creation_date (aiobungie.GroupDate | int): The creation date of the group. Defaults to 0 which is all time.
sort_by (int | None): ...
group_member_count_filter (int | None): ...
locale_filter (str | None): ...
tag_text (str | None): ...
items_per_page (int | None): ...
current_page (int | None): ...
request_token (str | None): ...

Returns

aiobungie.typedefs.JSONObject: A JSON object of the search results.

Raises

ValueError: If the group type is aiobungie.GroupType.CLAN and group_member_count_filter, locale_filter and tag_text are not None.

async def fetch_clan_admins(self, clan_id: 'int', /) -> 'typedefs.JSONObject': View Source

965    async def fetch_clan_admins(self, clan_id: int, /) -> typedefs.JSONObject:
966        resp = await self._request(_GET, f"GroupV2/{clan_id}/AdminsAndFounder/")
967        assert isinstance(resp, dict)
968        return resp

Fetch the admins and founder members of the clan.

Parameters

clan_id (int): The clan id.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the clan admins and founder members.

Raises

aiobungie.NotFound: The clan was not found.

async def fetch_clan_conversations(self, clan_id: 'int', /) -> 'typedefs.JSONArray': View Source

970    async def fetch_clan_conversations(self, clan_id: int, /) -> typedefs.JSONArray:
971        resp = await self._request(_GET, f"GroupV2/{clan_id}/OptionalConversations/")
972        assert isinstance(resp, list)
973        return resp

Fetch a clan's conversations.

Parameters

clan_id (int): The clan's id.

Returns

aiobungie.typedefs.JSONArray: A JSON array of the conversations.

async def fetch_application(self, appid: 'int', /) -> 'typedefs.JSONObject': View Source

975    async def fetch_application(self, appid: int, /) -> typedefs.JSONObject:
976        resp = await self._request(_GET, f"App/Application/{appid}")
977        assert isinstance(resp, dict)
978        return resp

Fetch a Bungie Application.

Parameters

appid (int): The application id.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the application.

async def fetch_character( self, member_id: 'int', membership_type: 'enums.MembershipType | int', character_id: 'int', components: 'collections.Sequence[enums.ComponentType]', auth: 'str | None' = None) -> 'typedefs.JSONObject': View Source

980    async def fetch_character(
981        self,
982        member_id: int,
983        membership_type: enums.MembershipType | int,
984        character_id: int,
985        components: collections.Sequence[enums.ComponentType],
986        auth: str | None = None,
987    ) -> typedefs.JSONObject:
988        collector = _collect_components(components)
989        response = await self._request(
990            _GET,
991            f"Destiny2/{int(membership_type)}/Profile/{member_id}/"
992            f"Character/{character_id}/?components={collector}",
993            auth=auth,
994        )
995        assert isinstance(response, dict)
996        return response

Fetch a Destiny 2 player's characters.

Parameters

member_id (int): A valid bungie member id.
membership_type (aiobungie.aiobungie.internal.enums.MembershipType | int): The member's membership type.
character_id (int): The character id to return.
components (collections.Sequence[aiobungie.ComponentType]): A list of character components to collect and return.

Other Parameters

auth (str | None): A bearer access_token to make the request with. This is optional and limited to components that only requires an Authorization token.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the requested character.

Raises

aiobungie.MembershipTypeError: The provided membership type was invalid.

async def fetch_activities( self, member_id: 'int', character_id: 'int', mode: 'enums.GameMode | int', membership_type: 'enums.MembershipType | int' = <MembershipType.ALL: -1>, *, page: 'int' = 0, limit: 'int' = 1) -> 'typedefs.JSONObject': View Source

 998    async def fetch_activities(
 999        self,
1000        member_id: int,
1001        character_id: int,
1002        mode: enums.GameMode | int,
1003        membership_type: enums.MembershipType | int = enums.MembershipType.ALL,
1004        *,
1005        page: int = 0,
1006        limit: int = 1,
1007    ) -> typedefs.JSONObject:
1008        resp = await self._request(
1009            _GET,
1010            f"Destiny2/{int(membership_type)}/Account/"
1011            f"{member_id}/Character/{character_id}/Stats/Activities"
1012            f"/?mode={int(mode)}&count={limit}&page={page}",
1013        )
1014        assert isinstance(resp, dict)
1015        return resp

Fetch a Destiny 2 activity for the specified user id and character.

Parameters

member_id (int): The user id that starts with 4611.
character_id (int): The id of the character to retrieve.
mode (aiobungie.aiobungie.GameMode | int): This parameter filters the game mode, Nightfall, Strike, Iron Banner, etc.
membership_type (aiobungie.aiobungie.internal.enums.MembershipType | int): The Destiny 2 membership type.

Other Parameters

page (int): The page number. Default to 1
limit (int): Limit the returned result. Default to 1

Returns

aiobungie.typedefs.JSONObject: A JSON object of the player's activities.

Raises

aiobungie.error.NotFound: The activity was not found.
aiobungie.MembershipTypeError: The provided membership type was invalid.

async def fetch_vendor_sales(self) -> 'typedefs.JSONObject': View Source

1017    async def fetch_vendor_sales(self) -> typedefs.JSONObject:
1018        resp = await self._request(
1019            _GET,
1020            f"Destiny2/Vendors/?components={int(enums.ComponentType.VENDOR_SALES)}",
1021        )
1022        assert isinstance(resp, dict)
1023        return resp

async def fetch_profile( self, membership_id: 'int', type: 'enums.MembershipType | int', components: 'collections.Sequence[enums.ComponentType]', auth: 'str | None' = None) -> 'typedefs.JSONObject': View Source

1025    async def fetch_profile(
1026        self,
1027        membership_id: int,
1028        type: enums.MembershipType | int,
1029        components: collections.Sequence[enums.ComponentType],
1030        auth: str | None = None,
1031    ) -> typedefs.JSONObject:
1032        collector = _collect_components(components)
1033        response = await self._request(
1034            _GET,
1035            f"Destiny2/{int(type)}/Profile/{membership_id}/?components={collector}",
1036            auth=auth,
1037        )
1038        assert isinstance(response, dict)
1039        return response

Fetch a bungie profile.

Parameters

membership_id (int): The member's id.
type (aiobungie.aiobungie.MembershipType | int): A valid membership type.
components (collections.Sequence[aiobungie.ComponentType]): A sequence of profile components to collect and return.

Other Parameters

auth (str | None): A bearer access_token to make the request with. This is optional and limited to components that only requires an Authorization token.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the found profile.

Raises

aiobungie.MembershipTypeError: The provided membership type was invalid.

async def fetch_entity(self, type: 'str', hash: 'int') -> 'typedefs.JSONObject': View Source

1041    async def fetch_entity(self, type: str, hash: int) -> typedefs.JSONObject:
1042        response = await self._request(_GET, route=f"Destiny2/Manifest/{type}/{hash}")
1043        assert isinstance(response, dict)
1044        return response

Fetch a Destiny definition item given its type and hash.

Parameters

type (str): Entity's type definition.
hash (int): Entity's hash.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the definition data.

async def fetch_inventory_item(self, hash: 'int', /) -> 'typedefs.JSONObject': View Source

1046    async def fetch_inventory_item(self, hash: int, /) -> typedefs.JSONObject:
1047        resp = await self.fetch_entity("DestinyInventoryItemDefinition", hash)
1048        assert isinstance(resp, dict)
1049        return resp

Fetch a Destiny inventory item entity given a its hash.

Parameters

hash (int): Entity's hash.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the inventory item.

async def fetch_objective_entity(self, hash: 'int', /) -> 'typedefs.JSONObject': View Source

1051    async def fetch_objective_entity(self, hash: int, /) -> typedefs.JSONObject:
1052        resp = await self.fetch_entity("DestinyObjectiveDefinition", hash)
1053        assert isinstance(resp, dict)
1054        return resp

Fetch a Destiny objective entity given a its hash.

Parameters

hash (int): objective's hash.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the objective data.

async def fetch_groups_for_member( self, member_id: 'int', member_type: 'enums.MembershipType | int', /, *, filter: 'int' = 0, group_type: 'enums.GroupType | int' = <GroupType.CLAN: 1>) -> 'typedefs.JSONObject': View Source

1056    async def fetch_groups_for_member(
1057        self,
1058        member_id: int,
1059        member_type: enums.MembershipType | int,
1060        /,
1061        *,
1062        filter: int = 0,
1063        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1064    ) -> typedefs.JSONObject:
1065        resp = await self._request(
1066            _GET,
1067            f"GroupV2/User/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1068        )
1069        assert isinstance(resp, dict)
1070        return resp

Fetch the information about the groups for a member.

Parameters

member_id (int): The member's id
member_type (aiobungie.aiobungie.MembershipType | int): The member's membership type.

Other Parameters

filter (int): Filter apply to list of joined groups. This Default to 0
group_type (aiobungie.aiobungie.GroupType | int): The group's type. This is always set to aiobungie.GroupType.CLAN and should not be changed.

Returns

aiobungie.typedefs.JSONObject: A JSON object of an array of the member's membership data and groups data.

async def fetch_potential_groups_for_member( self, member_id: 'int', member_type: 'enums.MembershipType | int', /, *, filter: 'int' = 0, group_type: 'enums.GroupType | int' = <GroupType.CLAN: 1>) -> 'typedefs.JSONObject': View Source

1072    async def fetch_potential_groups_for_member(
1073        self,
1074        member_id: int,
1075        member_type: enums.MembershipType | int,
1076        /,
1077        *,
1078        filter: int = 0,
1079        group_type: enums.GroupType | int = enums.GroupType.CLAN,
1080    ) -> typedefs.JSONObject:
1081        resp = await self._request(
1082            _GET,
1083            f"GroupV2/User/Potential/{int(member_type)}/{member_id}/{filter}/{int(group_type)}/",
1084        )
1085        assert isinstance(resp, dict)
1086        return resp

Get information about the groups that a given member has applied to or been invited to.

Parameters

member_id (int): The member's id
member_type (aiobungie.aiobungie.MembershipType | int): The member's membership type.

Other Parameters

filter (int): Filter apply to list of joined groups. This Default to 0
group_type (aiobungie.aiobungie.GroupType | int): The group's type. This is always set to aiobungie.GroupType.CLAN and should not be changed.

Returns

aiobungie.typedefs.JSONObject: A JSON object of an array of the member's membership data and groups data.

async def fetch_clan_members( self, clan_id: 'int', /, *, name: 'str | None' = None, type: 'enums.MembershipType | int' = <MembershipType.NONE: 0>) -> 'typedefs.JSONObject': View Source

1088    async def fetch_clan_members(
1089        self,
1090        clan_id: int,
1091        /,
1092        *,
1093        name: str | None = None,
1094        type: enums.MembershipType | int = enums.MembershipType.NONE,
1095    ) -> typedefs.JSONObject:
1096        resp = await self._request(
1097            _GET,
1098            f"/GroupV2/{clan_id}/Members/?memberType={int(type)}&nameSearch={name if name else ''}&currentpage=1",
1099        )
1100        assert isinstance(resp, dict)
1101        return resp

Fetch all Bungie Clan members.

Parameters

clan_id (int): The clans id

Other Parameters

name (str | None): If provided, Only players matching this name will be returned.
type (aiobungie.aiobungie.MembershipType | int): An optional clan member's membership type. Default is set to aiobungie.MembershipType.NONE Which returns the first matched clan member by their name.

Returns

aiobungie.typedefs.JSONObject: A JSON object of an array of clan members.

Raises

aiobungie.NotFound: The clan was not found.

async def fetch_hardlinked_credentials( self, credential: 'int', type: 'enums.CredentialType | int' = <CredentialType.STEAMID: 12>, /) -> 'typedefs.JSONObject': View Source

1103    async def fetch_hardlinked_credentials(
1104        self,
1105        credential: int,
1106        type: enums.CredentialType | int = enums.CredentialType.STEAMID,
1107        /,
1108    ) -> typedefs.JSONObject:
1109        resp = await self._request(
1110            _GET,
1111            f"User/GetMembershipFromHardLinkedCredential/{int(type)}/{credential}/",
1112        )
1113        assert isinstance(resp, dict)
1114        return resp

Gets any hard linked membership given a credential.

Only works for credentials that are public just aiobungie.CredentialType.STEAMID right now. Cross Save aware.

Parameters

credential (int): A valid SteamID64
type (aiobungie.aiobungie.CredentialType | int): The credential type. This must not be changed Since its only credential that works "currently"

Returns

aiobungie.typedefs.JSONObject: A JSON object of the found user hard linked types.

async def fetch_user_credentials( self, access_token: 'str', membership_id: 'int', /) -> 'typedefs.JSONArray': View Source

1116    async def fetch_user_credentials(
1117        self, access_token: str, membership_id: int, /
1118    ) -> typedefs.JSONArray:
1119        resp = await self._request(
1120            _GET,
1121            f"User/GetCredentialTypesForTargetAccount/{membership_id}",
1122            auth=access_token,
1123        )
1124        assert isinstance(resp, list)
1125        return resp

Fetch an array of credential types attached to the requested account.

This method require OAuth2 Bearer access token.

Parameters

access_token (str): The bearer access token associated with the bungie account.
membership_id (int): The id of the membership to return.

Returns

aiobungie.typedefs.JSONArray: A JSON array of the returned credentials.

Raises

aiobungie.Unauthorized: The access token was wrong or no access token passed.

async def insert_socket_plug( self, action_token: 'str', /, instance_id: 'int', plug: 'builders.PlugSocketBuilder | collections.Mapping[str, int]', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

1127    async def insert_socket_plug(
1128        self,
1129        action_token: str,
1130        /,
1131        instance_id: int,
1132        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1133        character_id: int,
1134        membership_type: enums.MembershipType | int,
1135    ) -> typedefs.JSONObject:
1136        if isinstance(plug, builders.PlugSocketBuilder):
1137            plug = plug.collect()
1138
1139        body = {
1140            "actionToken": action_token,
1141            "itemInstanceId": instance_id,
1142            "plug": plug,
1143            "characterId": character_id,
1144            "membershipType": int(membership_type),
1145        }
1146        resp = await self._request(
1147            _POST, "Destiny2/Actions/Items/InsertSocketPlug", json=body
1148        )
1149        assert isinstance(resp, dict)
1150        return resp

Insert a plug into a socketed item.

OAuth2: AdvancedWriteActions scope is required

Parameters

action_token (str): Action token provided by the AwaGetActionToken API call.
instance_id (int): The item instance id that's plug inserted.
plug (aiobungie.builders.PlugSocketBuilder | Mapping[str, int]): Either a PlugSocketBuilder object or a raw dict contains key, value for the plug entries.

Example

plug = (
    aiobungie.PlugSocketBuilder()
    .set_socket_array(0)
    .set_socket_index(0)
    .set_plug_item(3023847)
    .collect()
)
await insert_socket_plug_free(..., plug=plug)

character_id : int The character's id. membership_type : aiobungie.aiobungie.MembershipType | int The membership type.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains the changed item details.

Raises

aiobungie.Unauthorized: The access token was wrong or no access token passed.

async def insert_socket_plug_free( self, access_token: 'str', /, instance_id: 'int', plug: 'builders.PlugSocketBuilder | collections.Mapping[str, int]', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

1152    async def insert_socket_plug_free(
1153        self,
1154        access_token: str,
1155        /,
1156        instance_id: int,
1157        plug: builders.PlugSocketBuilder | collections.Mapping[str, int],
1158        character_id: int,
1159        membership_type: enums.MembershipType | int,
1160    ) -> typedefs.JSONObject:
1161        if isinstance(plug, builders.PlugSocketBuilder):
1162            plug = plug.collect()
1163
1164        body = {
1165            "itemInstanceId": instance_id,
1166            "plug": plug,
1167            "characterId": character_id,
1168            "membershipType": int(membership_type),
1169        }
1170        resp = await self._request(
1171            _POST,
1172            "Destiny2/Actions/Items/InsertSocketPlugFree",
1173            json=body,
1174            auth=access_token,
1175        )
1176        assert isinstance(resp, dict)
1177        return resp

Insert a plug into a socketed item. This doesn't require an Action token.

OAuth2: MoveEquipDestinyItems scope is required

Parameters

instance_id (int): The item instance id that's plug inserted.
plug (aiobungie.builders.PlugSocketBuilder | Mapping[str, int]): Either a PlugSocketBuilder object or a raw dict contains key, value for the plug entries.

Example

plug = (
    aiobungie.PlugSocketBuilder()
    .set_socket_array(0)
    .set_socket_index(0)
    .set_plug_item(3023847)
    .collect()
)
await insert_socket_plug_free(..., plug=plug)

character_id : int The character's id. membership_type : aiobungie.aiobungie.MembershipType | int The membership type.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains the changed item details.

Raises

aiobungie.Unauthorized: The access token was wrong or no access token passed.

@helpers.unstable

async def set_item_lock_state( self, access_token: 'str', state: 'bool', /, item_id: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'int': View Source

1179    @helpers.unstable
1180    async def set_item_lock_state(
1181        self,
1182        access_token: str,
1183        state: bool,
1184        /,
1185        item_id: int,
1186        character_id: int,
1187        membership_type: enums.MembershipType | int,
1188    ) -> int:
1189        body = {
1190            "state": state,
1191            "itemId": item_id,
1192            "characterId": character_id,
1193            "membershipType": int(membership_type),
1194        }
1195        response = await self._request(
1196            _POST,
1197            "Destiny2/Actions/Items/SetLockState",
1198            json=body,
1199            auth=access_token,
1200        )
1201        assert isinstance(response, int)
1202        return response

Set the Lock State for an instanced item.

OAuth2: MoveEquipDestinyItems scope is required

Parameters

access_token (str): The bearer access token associated with the bungie account.
state (bool): If True, The item will be locked, If False, The item will be unlocked.
item_id (int): The item id.
character_id (int): The character id.
membership_type (aiobungie.aiobungie.MembershipType | int): The membership type for the associated account.

Returns

int: An integer represents whether the request was successful or failed.

Raises

aiobungie.Unauthorized: - The access token was wrong
- No access token passed.
- Other authorization causes.

async def set_quest_track_state( self, access_token: 'str', state: 'bool', /, item_id: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'int': View Source

1204    async def set_quest_track_state(
1205        self,
1206        access_token: str,
1207        state: bool,
1208        /,
1209        item_id: int,
1210        character_id: int,
1211        membership_type: enums.MembershipType | int,
1212    ) -> int:
1213        body = {
1214            "state": state,
1215            "itemId": item_id,
1216            "characterId": character_id,
1217            "membership_type": int(membership_type),
1218        }
1219        response = await self._request(
1220            _POST,
1221            "Destiny2/Actions/Items/SetTrackedState",
1222            json=body,
1223            auth=access_token,
1224        )
1225        assert isinstance(response, int)
1226        return response

Set the Tracking State for an instanced Quest or Bounty.

OAuth2: MoveEquipDestinyItems scope is required

Parameters

access_token (str): The bearer access token associated with the bungie account.
state (bool): If True, The item will be locked, If False, The item will be unlocked.
item_id (int): The item id.
character_id (int): The character id.
membership_type (aiobungie.aiobungie.MembershipType | int): The membership type for the associated account.

Returns

int: An integer represents whether the request was successful or failed.

Raises

aiobungie.Unauthorized: - The access token was wrong
- No access token passed.
- Other authorization causes.

async def fetch_manifest_path(self) -> 'typedefs.JSONObject': View Source

1228    async def fetch_manifest_path(self) -> typedefs.JSONObject:
1229        path = await self._request(_GET, "Destiny2/Manifest")
1230        assert isinstance(path, dict)
1231        return path

Fetch the manifest JSON paths.

Returns

typedefs.JSONObject: The manifest JSON paths.

async def read_manifest_bytes(self, language: '_ALLOWED_LANGS' = 'en', /) -> 'bytes': View Source

1233    async def read_manifest_bytes(self, language: _ALLOWED_LANGS = "en", /) -> bytes:
1234        _ensure_manifest_language(language)
1235
1236        content = await self.fetch_manifest_path()
1237        resp = await self._request(
1238            _GET,
1239            content["mobileWorldContentPaths"][language],
1240            unwrap_bytes=True,
1241            base=True,
1242        )
1243        assert isinstance(resp, bytes)
1244        return resp

Read raw manifest SQLite database bytes response.

This method can be used to write the bytes to zipped file and then extract it to get the manifest content.

Parameters

language (str): The manifest database language bytes to get.

Returns

bytes: The bytes to read and write the manifest database.

async def download_sqlite_manifest( self, language: '_ALLOWED_LANGS' = 'en', name: 'str' = 'manifest', path: 'pathlib.Path | str' = '.', *, force: 'bool' = False, executor: 'concurrent.futures.Executor | None' = None) -> 'pathlib.Path': View Source

1246    async def download_sqlite_manifest(
1247        self,
1248        language: _ALLOWED_LANGS = "en",
1249        name: str = "manifest",
1250        path: pathlib.Path | str = ".",
1251        *,
1252        force: bool = False,
1253        executor: concurrent.futures.Executor | None = None,
1254    ) -> pathlib.Path:
1255        complete_path = _get_path(name, path, sql=True)
1256
1257        if complete_path.exists():
1258            if force:
1259                _LOGGER.info(
1260                    f"Found manifest in {complete_path!s}. Forcing to Re-Download."
1261                )
1262                complete_path.unlink(missing_ok=True)
1263
1264                return await self.download_sqlite_manifest(
1265                    language, name, path, force=force
1266                )
1267
1268            else:
1269                raise FileExistsError(
1270                    "Manifest file already exists, "
1271                    "To force download, set the `force` parameter to `True`."
1272                )
1273
1274        _LOGGER.info(f"Downloading manifest. Location: {complete_path!s}")
1275        data_bytes = await self.read_manifest_bytes(language)
1276        await asyncio.get_running_loop().run_in_executor(
1277            executor, _write_sqlite_bytes, data_bytes, path, name
1278        )
1279        _LOGGER.info("Finished downloading manifest.")
1280        return _get_path(name, path, sql=True)

Downloads the SQLite version of Destiny2's Manifest.

Example

manifest = await rest.download_sqlite_manifest()
with sqlite3.connect(manifest) as conn:
    ...

Parameters

language (str): The manifest language to download, Default is English.
path (str | pathlib.Path): The path to download this manifest. Example "/tmp/databases/", Default is the current directory.
name (str): The manifest database file name. Default is manifest
force (bool): Whether to force the download. Default is False. However if set to true the old file will get unlinked and a new one will begin to download.
executor (concurrent.futures.Executor | None): An optional executor which will be used to write the bytes of the manifest.

Returns

pathlib.Path: A pathlib object of the .sqlite file.

Raises

FileExistsError: If the manifest file exists and force is False.
ValueError: If the provided language was not recognized.

async def download_json_manifest( self, file_name: 'str' = 'manifest', path: 'str | pathlib.Path' = '.', *, language: '_ALLOWED_LANGS' = 'en', executor: 'concurrent.futures.Executor | None' = None) -> 'pathlib.Path': View Source

1282    async def download_json_manifest(
1283        self,
1284        file_name: str = "manifest",
1285        path: str | pathlib.Path = ".",
1286        *,
1287        language: _ALLOWED_LANGS = "en",
1288        executor: concurrent.futures.Executor | None = None,
1289    ) -> pathlib.Path:
1290        _ensure_manifest_language(language)
1291        full_path = _get_path(file_name, path)
1292        _LOGGER.info(f"Downloading manifest JSON to {full_path!r}...")
1293
1294        content = await self.fetch_manifest_path()
1295        json_bytes = await self._request(
1296            _GET,
1297            content["jsonWorldContentPaths"][language],
1298            unwrap_bytes=True,
1299            base=True,
1300        )
1301
1302        assert isinstance(json_bytes, bytes)
1303        await asyncio.get_running_loop().run_in_executor(
1304            executor, _write_json_bytes, json_bytes, file_name, path
1305        )
1306        _LOGGER.info("Finished downloading manifest JSON.")
1307        return full_path

Download the Bungie manifest json file.

Example

manifest = await rest.download_json_manifest()
with open(manifest, "r") as f:
    to_dict = json.loads(f.read())
    item_definitions = to_dict['DestinyInventoryItemDefinition']

Parameters

file_name (str): The file name to save the manifest json file. Default is manifest.
path (str | pathlib.Path): The path to save the manifest json file. Default is the current directory. Example "D:/"
language (str): The manifest database language bytes to get. Default is English.
executor (concurrent.futures.Executor | None): An optional executor which will be used to write the bytes of the manifest.

Returns

pathlib.Path: The path of this JSON manifest.

async def fetch_manifest_version(self) -> 'str': View Source

1309    async def fetch_manifest_version(self) -> str:
1310        # This is guaranteed str.
1311        return (await self.fetch_manifest_path())["version"]

Fetch the manifest version.

Returns

str: The manifest version.

async def fetch_linked_profiles( self, member_id: 'int', member_type: 'enums.MembershipType | int', /, *, all: 'bool' = False) -> 'typedefs.JSONObject': View Source

1313    async def fetch_linked_profiles(
1314        self,
1315        member_id: int,
1316        member_type: enums.MembershipType | int,
1317        /,
1318        *,
1319        all: bool = False,
1320    ) -> typedefs.JSONObject:
1321        resp = await self._request(
1322            _GET,
1323            f"Destiny2/{int(member_type)}/Profile/{member_id}/LinkedProfiles/?getAllMemberships={all}",
1324        )
1325        assert isinstance(resp, dict)
1326        return resp

Returns a summary information about all profiles linked to the requested member.

The passed membership id/type maybe a Bungie.Net membership or a Destiny memberships.

It will only return linked accounts whose linkages you are allowed to view.

Parameters

member_id (int): The ID of the membership. This must be a valid Bungie.Net or PSN or Xbox ID.
member_type (aiobungie.aiobungie.MembershipType | int): The type for the membership whose linked Destiny account you want to return.

Other Parameters

all (bool): If provided and set to True, All memberships regardless of whether they're obscured by overrides will be returned,

If provided and set to False, Only available memberships will be returned. The default for this is False.

Returns

aiobungie.typedefs.JSONObject
A JSON object which contains an Array of profiles, an Array of profiles with errors and Bungie.Net membership

async def fetch_clan_banners(self) -> 'typedefs.JSONObject': View Source

1328    async def fetch_clan_banners(self) -> typedefs.JSONObject:
1329        resp = await self._request(_GET, "Destiny2/Clan/ClanBannerDictionary/")
1330        assert isinstance(resp, dict)
1331        return resp

Fetch the values of the clan banners.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the clan banners.

async def fetch_public_milestones(self) -> 'typedefs.JSONObject': View Source

1333    async def fetch_public_milestones(self) -> typedefs.JSONObject:
1334        resp = await self._request(_GET, "Destiny2/Milestones/")
1335        assert isinstance(resp, dict)
1336        return resp

Fetch the available milestones.

Returns

aiobungie.typedefs.JSONObject: A JSON object of information about the milestones.

async def fetch_public_milestone_content(self, milestone_hash: 'int', /) -> 'typedefs.JSONObject': View Source

1338    async def fetch_public_milestone_content(
1339        self, milestone_hash: int, /
1340    ) -> typedefs.JSONObject:
1341        resp = await self._request(
1342            _GET, f"Destiny2/Milestones/{milestone_hash}/Content/"
1343        )
1344        assert isinstance(resp, dict)
1345        return resp

Fetch the milestone content given its hash.

Parameters

milestone_hash (int): The milestone hash.

Returns

aiobungie.typedefs.JSONObject: A JSON object of information related to the fetched milestone.

async def fetch_current_user_memberships(self, access_token: 'str', /) -> 'typedefs.JSONObject': View Source

1347    async def fetch_current_user_memberships(
1348        self, access_token: str, /
1349    ) -> typedefs.JSONObject:
1350        resp = await self._request(
1351            _GET,
1352            "User/GetMembershipsForCurrentUser/",
1353            auth=access_token,
1354        )
1355        assert isinstance(resp, dict)
1356        return resp

Fetch a bungie user's accounts with the signed in user. This GET method requires a Bearer access token for the authorization.

This requires OAuth2 scope enabled and the valid Bearer access_token.

Parameters

access_token (str): The bearer access token associated with the bungie account.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the bungie net user and destiny memberships of this account.

async def equip_item( self, access_token: 'str', /, item_id: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'None': View Source

1358    async def equip_item(
1359        self,
1360        access_token: str,
1361        /,
1362        item_id: int,
1363        character_id: int,
1364        membership_type: enums.MembershipType | int,
1365    ) -> None:
1366        payload = {
1367            "itemId": item_id,
1368            "characterId": character_id,
1369            "membershipType": int(membership_type),
1370        }
1371
1372        await self._request(
1373            _POST,
1374            "Destiny2/Actions/Items/EquipItem/",
1375            json=payload,
1376            auth=access_token,
1377        )

Equip an item to a character.

This requires the OAuth2: MoveEquipDestinyItems scope. Also You must have a valid Destiny account, and either be in a social space, in orbit or offline.

Parameters

access_token (str): The bearer access token associated with the bungie account.
item_id (int): The item id.
character_id (int): The character's id to equip the item to.
membership_type (aiobungie.aiobungie.MembershipType | int): The membership type associated with this player.

async def equip_items( self, access_token: 'str', /, item_ids: 'collections.Sequence[int]', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'None': View Source

1379    async def equip_items(
1380        self,
1381        access_token: str,
1382        /,
1383        item_ids: collections.Sequence[int],
1384        character_id: int,
1385        membership_type: enums.MembershipType | int,
1386    ) -> None:
1387        payload = {
1388            "itemIds": item_ids,
1389            "characterId": character_id,
1390            "membershipType": int(membership_type),
1391        }
1392        await self._request(
1393            _POST,
1394            "Destiny2/Actions/Items/EquipItems/",
1395            json=payload,
1396            auth=access_token,
1397        )

Equip multiple items to a character.

This requires the OAuth2: MoveEquipDestinyItems scope. Also You must have a valid Destiny account, and either be in a social space, in orbit or offline.

Parameters

access_token (str): The bearer access token associated with the bungie account.
item_ids (Sequence[int]): A sequence of item ids.
character_id (int): The character's id to equip the item to.
membership_type (aiobungie.aiobungie.MembershipType | int): The membership type associated with this player.

async def ban_clan_member( self, access_token: 'str', /, group_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', *, length: 'int' = 0, comment: 'str | None' = None) -> 'None': View Source

1399    async def ban_clan_member(
1400        self,
1401        access_token: str,
1402        /,
1403        group_id: int,
1404        membership_id: int,
1405        membership_type: enums.MembershipType | int,
1406        *,
1407        length: int = 0,
1408        comment: str | None = None,
1409    ) -> None:
1410        payload = {"comment": str(comment), "length": length}
1411        await self._request(
1412            _POST,
1413            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Ban/",
1414            json=payload,
1415            auth=access_token,
1416        )

Bans a member from the clan.

This request requires OAuth2: oauth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group id.
membership_id (int): The member id to ban.
membership_type (aiobungie.aiobungie.MembershipType | int): The member's membership type.

Other Parameters

length (int): An optional ban length.
comment (aiobungie.UndefinedOr[str]): An optional comment to this ban. Default is UNDEFINED

async def unban_clan_member( self, access_token: 'str', /, group_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int') -> 'None': View Source

1418    async def unban_clan_member(
1419        self,
1420        access_token: str,
1421        /,
1422        group_id: int,
1423        membership_id: int,
1424        membership_type: enums.MembershipType | int,
1425    ) -> None:
1426        await self._request(
1427            _POST,
1428            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Unban/",
1429            auth=access_token,
1430        )

Unban a member from the clan.

This request requires OAuth2: oauth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group id.
membership_id (int): The member id to unban.
membership_type (aiobungie.aiobungie.MembershipType | int): The member's membership type.

async def kick_clan_member( self, access_token: 'str', /, group_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

1432    async def kick_clan_member(
1433        self,
1434        access_token: str,
1435        /,
1436        group_id: int,
1437        membership_id: int,
1438        membership_type: enums.MembershipType | int,
1439    ) -> typedefs.JSONObject:
1440        resp = await self._request(
1441            _POST,
1442            f"GroupV2/{group_id}/Members/{int(membership_type)}/{membership_id}/Kick/",
1443            auth=access_token,
1444        )
1445        assert isinstance(resp, dict)
1446        return resp

Kick a member from the clan.

This request requires OAuth2: oauth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group id.
membership_id (int): The member id to kick.
membership_type (aiobungie.aiobungie.MembershipType | int): The member's membership type.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the group that the member has been kicked from.

1448    async def edit_clan(
1449        self,
1450        access_token: str,
1451        /,
1452        group_id: int,
1453        *,
1454        name: str | None = None,
1455        about: str | None = None,
1456        motto: str | None = None,
1457        theme: str | None = None,
1458        tags: collections.Sequence[str] | None = None,
1459        is_public: bool | None = None,
1460        locale: str | None = None,
1461        avatar_image_index: int | None = None,
1462        membership_option: enums.MembershipOption | int | None = None,
1463        allow_chat: bool | None = None,
1464        chat_security: typing.Literal[0, 1] | None = None,
1465        call_sign: str | None = None,
1466        homepage: typing.Literal[0, 1, 2] | None = None,
1467        enable_invite_messaging_for_admins: bool | None = None,
1468        default_publicity: typing.Literal[0, 1, 2] | None = None,
1469        is_public_topic_admin: bool | None = None,
1470    ) -> None:
1471        payload = {
1472            "name": name,
1473            "about": about,
1474            "motto": motto,
1475            "theme": theme,
1476            "tags": tags,
1477            "isPublic": is_public,
1478            "avatarImageIndex": avatar_image_index,
1479            "isPublicTopicAdminOnly": is_public_topic_admin,
1480            "allowChat": allow_chat,
1481            "chatSecurity": chat_security,
1482            "callsign": call_sign,
1483            "homepage": homepage,
1484            "enableInvitationMessagingForAdmins": enable_invite_messaging_for_admins,
1485            "defaultPublicity": default_publicity,
1486            "locale": locale,
1487        }
1488        if membership_option is not None:
1489            payload["membershipOption"] = int(membership_option)
1490
1491        await self._request(
1492            _POST,
1493            f"GroupV2/{group_id}/Edit",
1494            json=payload,
1495            auth=access_token,
1496        )

Edit a clan.

Notes

This request requires OAuth2: oauth2: AdminGroups scope.
All arguments will default to None if not provided. This does not include access_token and group_id

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group id to edit.

Other Parameters

name (str | None): The name to edit the clan with.
about (str | None): The about section to edit the clan with.
motto (str | None): The motto section to edit the clan with.
theme (str | None): The theme name to edit the clan with.
tags (collections.Sequence[str] | None): A sequence of strings to replace the clan tags with.
is_public (bool | None): If provided and set to True, The clan will set to private. If provided and set to False, The clan will set to public whether it was or not.
locale (str | None): The locale section to edit the clan with.
avatar_image_index (int | None): The clan avatar image index to edit the clan with.
membership_option : aiobungie.typedefs.NoneOr[aiobungie.aiobungie.MembershipOption | int] # noqa (E501 # Line too long): The clan membership option to edit it with.
allow_chat (bool | None): If provided and set to True, The clan members will be allowed to chat. If provided and set to False, The clan members will not be allowed to chat.
chat_security (aiobungie.typedefs.NoneOr[typing.Literal[0, 1]]): If provided and set to 0, The clan chat security will be edited to Group only. If provided and set to 1, The clan chat security will be edited to Admin only.
call_sign (str | None): The clan call sign to edit it with.
homepage (aiobungie.typedefs.NoneOr[typing.Literal[0, 1, 2]]): If provided and set to 0, The clan chat homepage will be edited to Wall. If provided and set to 1, The clan chat homepage will be edited to Forum. If provided and set to 0, The clan chat homepage will be edited to AllianceForum.
enable_invite_messaging_for_admins (bool | None): ???
default_publicity (aiobungie.typedefs.NoneOr[typing.Literal[0, 1, 2]]): If provided and set to 0, The clan chat publicity will be edited to Public. If provided and set to 1, The clan chat publicity will be edited to Alliance. If provided and set to 2, The clan chat publicity will be edited to Private.
is_public_topic_admin (bool | None): ???

async def edit_clan_options( self, access_token: 'str', /, group_id: 'int', *, invite_permissions_override: 'bool | None' = None, update_culture_permissionOverride: 'bool | None' = None, host_guided_game_permission_override: 'typing.Literal[0, 1, 2] | None' = None, update_banner_permission_override: 'bool | None' = None, join_level: 'enums.ClanMemberType | int | None' = None) -> 'None': View Source

1498    async def edit_clan_options(
1499        self,
1500        access_token: str,
1501        /,
1502        group_id: int,
1503        *,
1504        invite_permissions_override: bool | None = None,
1505        update_culture_permissionOverride: bool | None = None,
1506        host_guided_game_permission_override: typing.Literal[0, 1, 2] | None = None,
1507        update_banner_permission_override: bool | None = None,
1508        join_level: enums.ClanMemberType | int | None = None,
1509    ) -> None:
1510        payload = {
1511            "InvitePermissionOverride": invite_permissions_override,
1512            "UpdateCulturePermissionOverride": update_culture_permissionOverride,
1513            "HostGuidedGamePermissionOverride": host_guided_game_permission_override,
1514            "UpdateBannerPermissionOverride": update_banner_permission_override,
1515            "JoinLevel": int(join_level) if join_level else None,
1516        }
1517
1518        await self._request(
1519            _POST,
1520            f"GroupV2/{group_id}/EditFounderOptions",
1521            json=payload,
1522            auth=access_token,
1523        )

Edit the clan options.

Notes

This request requires OAuth2: oauth2: AdminGroups scope.
All arguments will default to None if not provided. This does not include access_token and group_id

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group id.

Other Parameters

invite_permissions_override (bool | None): Minimum Member Level allowed to invite new members to group Always Allowed: Founder, Acting Founder True means admins have this power, false means they don't Default is False for clans, True for groups.
update_culture_permissionOverride (bool | None): Minimum Member Level allowed to update group culture Always Allowed: Founder, Acting Founder True means admins have this power, false means they don't Default is False for clans, True for groups.
host_guided_game_permission_override (aiobungie.typedefs.NoneOr[typing.Literal[0, 1, 2]]): Minimum Member Level allowed to host guided games Always Allowed: Founder, Acting Founder, Admin Allowed Overrides: 0 -> None, 1 -> Beginner 2 -> Member. Default is Member for clans, None for groups, although this means nothing for groups.
update_banner_permission_override (bool | None): Minimum Member Level allowed to update banner Always Allowed: Founder, Acting Founder True means admins have this power, false means they don't Default is False for clans, True for groups.
join_level (aiobungie.ClanMemberType): Level to join a member at when accepting an invite, application, or joining an open clan. Default is aiobungie.ClanMemberType.BEGINNER

async def report_player( self, access_token: 'str', /, activity_id: 'int', character_id: 'int', reason_hashes: 'collections.Sequence[int]', reason_category_hashes: 'collections.Sequence[int]') -> 'None': View Source

1525    async def report_player(
1526        self,
1527        access_token: str,
1528        /,
1529        activity_id: int,
1530        character_id: int,
1531        reason_hashes: collections.Sequence[int],
1532        reason_category_hashes: collections.Sequence[int],
1533    ) -> None:
1534        await self._request(
1535            _POST,
1536            f"Destiny2/Stats/PostGameCarnageReport/{activity_id}/Report/",
1537            json={
1538                "reasonCategoryHashes": reason_category_hashes,
1539                "reasonHashes": reason_hashes,
1540                "offendingCharacterId": character_id,
1541            },
1542            auth=access_token,
1543        )

Report a player that you met in an activity that was engaging in ToS-violating activities.

This method requires BnetWrite OAuth2 scope.

Both you and the offending player must have played in the activity_id passed in. Please use this judiciously and only when you have strong suspicions of violation, pretty please.

Parameters

access_token (str): The bearer access token associated with the bungie account that will be used to make the request with.
activity_id (int): The activity where you ran into the person you're reporting.
character_id (int): The character ID of the person you're reporting.
reason_hashes (Sequence[int]): See
reason_category_hashes (Sequence[int]): See

async def fetch_friends(self, access_token: 'str', /) -> 'typedefs.JSONObject': View Source

1545    async def fetch_friends(self, access_token: str, /) -> typedefs.JSONObject:
1546        resp = await self._request(
1547            _GET,
1548            "Social/Friends/",
1549            auth=access_token,
1550        )
1551        assert isinstance(resp, dict)
1552        return resp

Fetch bungie friend list.

This requests OAuth2: ReadUserData scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.

Returns

aiobungie.typedefs.JSONObject: A JSON object of an array of the bungie friends's data.

async def fetch_friend_requests(self, access_token: 'str', /) -> 'typedefs.JSONObject': View Source

1554    async def fetch_friend_requests(self, access_token: str, /) -> typedefs.JSONObject:
1555        resp = await self._request(
1556            _GET,
1557            "Social/Friends/Requests",
1558            auth=access_token,
1559        )
1560        assert isinstance(resp, dict)
1561        return resp

Fetch pending bungie friend requests queue.

This requests OAuth2: ReadUserData scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.

Returns

aiobungie.typedefs.JSONObject: A JSON object of incoming requests and outgoing requests.

async def accept_friend_request(self, access_token: 'str', /, member_id: 'int') -> 'None': View Source

1563    async def accept_friend_request(self, access_token: str, /, member_id: int) -> None:
1564        await self._request(
1565            _POST,
1566            f"Social/Friends/Requests/Accept/{member_id}",
1567            auth=access_token,
1568        )

Accepts a friend relationship with the target user. The user must be on your incoming friend request list.

This request requires OAuth2: BnetWrite scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
member_id (int): The member's id to accept.

async def send_friend_request(self, access_token: 'str', /, member_id: 'int') -> 'None': View Source

1570    async def send_friend_request(self, access_token: str, /, member_id: int) -> None:
1571        await self._request(
1572            _POST,
1573            f"Social/Friends/Add/{member_id}",
1574            auth=access_token,
1575        )

Requests a friend relationship with the target user.

This request requires OAuth2: BnetWrite scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
member_id (int): The member's id to send the request to.

async def decline_friend_request(self, access_token: 'str', /, member_id: 'int') -> 'None': View Source

1577    async def decline_friend_request(
1578        self, access_token: str, /, member_id: int
1579    ) -> None:
1580        await self._request(
1581            _POST,
1582            f"Social/Friends/Requests/Decline/{member_id}",
1583            auth=access_token,
1584        )

Decline a friend request with the target user. The user must be in your incoming friend request list.

This request requires OAuth2: BnetWrite scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
member_id (int): The member's id to decline.

async def remove_friend(self, access_token: 'str', /, member_id: 'int') -> 'None': View Source

1586    async def remove_friend(self, access_token: str, /, member_id: int) -> None:
1587        await self._request(
1588            _POST,
1589            f"Social/Friends/Remove/{member_id}",
1590            auth=access_token,
1591        )

Removes a friend from your friend list. The user must be in your friend list.

This request requires OAuth2: BnetWrite scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
member_id (int): The member's id to remove.

async def remove_friend_request(self, access_token: 'str', /, member_id: 'int') -> 'None': View Source

1593    async def remove_friend_request(self, access_token: str, /, member_id: int) -> None:
1594        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1595        await self._request(
1596            _POST,
1597            f"Social/Friends/Requests/Remove/{member_id}",
1598            auth=access_token,
1599        )

Removes a friend from your friend list requests. The user must be in your outgoing request list.

.. note : This request requires OAuth2: BnetWrite scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
member_id (int): The member's id to remove from the requested friend list.

async def approve_all_pending_group_users( self, access_token: 'str', /, group_id: 'int', message: 'str | None' = None) -> 'None': View Source

1601    async def approve_all_pending_group_users(
1602        self,
1603        access_token: str,
1604        /,
1605        group_id: int,
1606        message: str | None = None,
1607    ) -> None:
1608        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1609        await self._request(
1610            _POST,
1611            f"GroupV2/{group_id}/Members/ApproveAll",
1612            auth=access_token,
1613            json={"message": str(message)},
1614        )

Approve all pending users for the given group id.

This request requires OAuth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The given group id.

Other Parameters

message (aiobungie.UndefinedOr[str]): An optional message to send with the request. Default is UNDEFINED.

async def deny_all_pending_group_users( self, access_token: 'str', /, group_id: 'int', *, message: 'str | None' = None) -> 'None': View Source

1616    async def deny_all_pending_group_users(
1617        self,
1618        access_token: str,
1619        /,
1620        group_id: int,
1621        *,
1622        message: str | None = None,
1623    ) -> None:
1624        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1625        await self._request(
1626            _POST,
1627            f"GroupV2/{group_id}/Members/DenyAll",
1628            auth=access_token,
1629            json={"message": str(message)},
1630        )

Deny all pending users for the given group id.

This request requires OAuth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The given group id.

Other Parameters

message (aiobungie.UndefinedOr[str]): An optional message to send with the request. Default is UNDEFINED.

async def add_optional_conversation( self, access_token: 'str', /, group_id: 'int', *, name: 'str | None' = None, security: 'typing.Literal[0, 1]' = 0) -> 'None': View Source

1632    async def add_optional_conversation(
1633        self,
1634        access_token: str,
1635        /,
1636        group_id: int,
1637        *,
1638        name: str | None = None,
1639        security: typing.Literal[0, 1] = 0,
1640    ) -> None:
1641        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1642        payload = {"chatName": str(name), "chatSecurity": security}
1643        await self._request(
1644            _POST,
1645            f"GroupV2/{group_id}/OptionalConversations/Add",
1646            json=payload,
1647            auth=access_token,
1648        )

Add a new chat channel to a group.

This request requires OAuth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The given group id.

Other parameters

name: aiobungie.UndefinedOr[str] The chat name. Default to UNDEFINED security: typing.Literal[0, 1] The security level of the chat.

If provided and set to 0, It will be to `Group` only.
If provided and set to 1, It will be `Admins` only.
Default is `0`

async def edit_optional_conversation( self, access_token: 'str', /, group_id: 'int', conversation_id: 'int', *, name: 'str | None' = None, security: 'typing.Literal[0, 1]' = 0, enable_chat: 'bool' = False) -> 'None': View Source

1650    async def edit_optional_conversation(
1651        self,
1652        access_token: str,
1653        /,
1654        group_id: int,
1655        conversation_id: int,
1656        *,
1657        name: str | None = None,
1658        security: typing.Literal[0, 1] = 0,
1659        enable_chat: bool = False,
1660    ) -> None:
1661        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1662        payload = {
1663            "chatEnabled": enable_chat,
1664            "chatName": str(name),
1665            "chatSecurity": security,
1666        }
1667        await self._request(
1668            _POST,
1669            f"GroupV2/{group_id}/OptionalConversations/Edit/{conversation_id}",
1670            json=payload,
1671            auth=access_token,
1672        )

Edit the settings of this chat channel.

This request requires OAuth2: AdminGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The given group id.
conversation_id (int): The conversation/chat id.

Other parameters

name: aiobungie.UndefinedOr[str] The new chat name. Default to UNDEFINED security: typing.Literal[0, 1] The new security level of the chat.

If provided and set to 0, It will be to `Group` only.
If provided and set to 1, It will be `Admins` only.
Default is `0`

enable_chat : bool Whether to enable chatting or not. If set to True then chatting will be enabled. Otherwise it will be disabled.

async def transfer_item( self, access_token: 'str', /, item_id: 'int', item_hash: 'int', character_id: 'int', member_type: 'enums.MembershipType | int', *, stack_size: 'int' = 1, vault: 'bool' = False) -> 'None': View Source

1674    async def transfer_item(
1675        self,
1676        access_token: str,
1677        /,
1678        item_id: int,
1679        item_hash: int,
1680        character_id: int,
1681        member_type: enums.MembershipType | int,
1682        *,
1683        stack_size: int = 1,
1684        vault: bool = False,
1685    ) -> None:
1686        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1687        payload = {
1688            "characterId": character_id,
1689            "membershipType": int(member_type),
1690            "itemId": item_id,
1691            "itemReferenceHash": item_hash,
1692            "stackSize": stack_size,
1693            "transferToVault": vault,
1694        }
1695        await self._request(
1696            _POST,
1697            "Destiny2/Actions/Items/TransferItem",
1698            json=payload,
1699            auth=access_token,
1700        )

Transfer an item from / to your vault.

Notes

This method requires OAuth2: MoveEquipDestinyItems scope.
This method requires both item id and hash.

Parameters

access_token (str): The bearer access token associated with the bungie account.
item_id (int): The item instance id you to transfer.
item_hash (int): The item hash.
character_id (int): The character id to transfer the item from/to.
member_type (aiobungie.aiobungie.MembershipType | int): The user membership type.

Other Parameters

stack_size (int): The item stack size.
vault (bool): Whether to transfer this item to your vault or not. Defaults to False.

async def pull_item( self, access_token: 'str', /, item_id: 'int', item_hash: 'int', character_id: 'int', member_type: 'enums.MembershipType | int', *, stack_size: 'int' = 1, vault: 'bool' = False) -> 'None': View Source

1702    async def pull_item(
1703        self,
1704        access_token: str,
1705        /,
1706        item_id: int,
1707        item_hash: int,
1708        character_id: int,
1709        member_type: enums.MembershipType | int,
1710        *,
1711        stack_size: int = 1,
1712        vault: bool = False,
1713    ) -> None:
1714        # <<inherited docstring from aiobungie.interfaces.rest.RESTInterface>>
1715        payload = {
1716            "characterId": character_id,
1717            "membershipType": int(member_type),
1718            "itemId": item_id,
1719            "itemReferenceHash": item_hash,
1720            "stackSize": stack_size,
1721        }
1722        await self._request(
1723            _POST,
1724            "Destiny2/Actions/Items/PullFromPostmaster",
1725            json=payload,
1726            auth=access_token,
1727        )
1728        if vault:
1729            await self.transfer_item(
1730                access_token,
1731                item_id=item_id,
1732                item_hash=item_hash,
1733                character_id=character_id,
1734                member_type=member_type,
1735                stack_size=stack_size,
1736                vault=True,
1737            )

pull an item from the postmaster.

Notes

This method requires OAuth2: MoveEquipDestinyItems scope.
This method requires both item id and hash.

Parameters

access_token (str): The bearer access token associated with the bungie account.
item_id (int): The item instance id to pull.
item_hash (int): The item hash.
character_id (int): The character id to pull the item to.
member_type (aiobungie.aiobungie.MembershipType | int): The user membership type.

Other Parameters

stack_size (int): The item stack size.
vault (bool): If True, an extra HTTP call will be performed to transfer this item to the vault, Defaults to False.

@helpers.unstable

async def fetch_fireteams( self, activity_type: 'fireteams.FireteamActivity | int', *, platform: 'fireteams.FireteamPlatform | int' = <FireteamPlatform.ANY: 0>, language: 'fireteams.FireteamLanguage | str' = <FireteamLanguage.ALL: >, date_range: 'fireteams.FireteamDate | int' = <FireteamDate.ALL: 0>, page: 'int' = 0, slots_filter: 'int' = 0) -> 'typedefs.JSONObject': View Source

1739    @helpers.unstable
1740    async def fetch_fireteams(
1741        self,
1742        activity_type: fireteams.FireteamActivity | int,
1743        *,
1744        platform: fireteams.FireteamPlatform | int = fireteams.FireteamPlatform.ANY,
1745        language: fireteams.FireteamLanguage | str = fireteams.FireteamLanguage.ALL,
1746        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1747        page: int = 0,
1748        slots_filter: int = 0,
1749    ) -> typedefs.JSONObject:
1750        resp = await self._request(
1751            _GET,
1752            f"Fireteam/Search/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{page}/?langFilter={str(language)}",  # noqa: E501 Line too long
1753        )
1754        assert isinstance(resp, dict)
1755        return resp

Fetch public Bungie fireteams with open slots.

Parameters

activity_type (aiobungie.aiobungie.crates.FireteamActivity | int): The fireteam activity type.

Other Parameters

platform (aiobungie.aiobungie.crates.fireteams.FireteamPlatform | int): If this is provided. Then the results will be filtered with the given platform. Defaults to aiobungie.crates.FireteamPlatform.ANY which returns all platforms.
language (aiobungie.crates.fireteams.FireteamLanguage | str): A locale language to filter the used language in that fireteam. Defaults to aiobungie.crates.FireteamLanguage.ALL
date_range (aiobungie.aiobungie.FireteamDate | int): An integer to filter the date range of the returned fireteams. Defaults to aiobungie.FireteamDate.ALL.
page (int): The page number. By default its 0 which returns all available activities.
slots_filter (int): Filter the returned fireteams based on available slots. Default is 0

Returns

aiobungie.typedefs.JSONObject: A JSON object of the fireteam details.

async def fetch_available_clan_fireteams( self, access_token: 'str', group_id: 'int', activity_type: 'fireteams.FireteamActivity | int', *, platform: 'fireteams.FireteamPlatform | int', language: 'fireteams.FireteamLanguage | str', date_range: 'fireteams.FireteamDate | int' = <FireteamDate.ALL: 0>, page: 'int' = 0, public_only: 'bool' = False, slots_filter: 'int' = 0) -> 'typedefs.JSONObject': View Source

1757    async def fetch_available_clan_fireteams(
1758        self,
1759        access_token: str,
1760        group_id: int,
1761        activity_type: fireteams.FireteamActivity | int,
1762        *,
1763        platform: fireteams.FireteamPlatform | int,
1764        language: fireteams.FireteamLanguage | str,
1765        date_range: fireteams.FireteamDate | int = fireteams.FireteamDate.ALL,
1766        page: int = 0,
1767        public_only: bool = False,
1768        slots_filter: int = 0,
1769    ) -> typedefs.JSONObject:
1770        resp = await self._request(
1771            _GET,
1772            f"Fireteam/Clan/{group_id}/Available/{int(platform)}/{int(activity_type)}/{int(date_range)}/{slots_filter}/{public_only}/{page}",  # noqa: E501
1773            json={"langFilter": str(language)},
1774            auth=access_token,
1775        )
1776        assert isinstance(resp, dict)
1777        return resp

Fetch a clan's fireteams with open slots.

This method requires OAuth2: ReadGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group/clan id of the fireteam.
activity_type (aiobungie.aiobungie.crates.FireteamActivity | int): The fireteam activity type.

Other Parameters

platform (aiobungie.aiobungie.crates.fireteams.FireteamPlatform | int): If this is provided. Then the results will be filtered with the given platform. Defaults to aiobungie.crates.FireteamPlatform.ANY which returns all platforms.
language (aiobungie.crates.fireteams.FireteamLanguage | str): A locale language to filter the used language in that fireteam. Defaults to aiobungie.crates.FireteamLanguage.ALL
date_range (aiobungie.aiobungie.FireteamDate | int): An integer to filter the date range of the returned fireteams. Defaults to aiobungie.FireteamDate.ALL.
page (int): The page number. By default its 0 which returns all available activities.
public_only (bool): If set to True, Then only public fireteams will be returned.
slots_filter (int): Filter the returned fireteams based on available slots. Default is 0

Returns

aiobungie.typedefs.JSONObject: A JSON object of the fireteams detail.

async def fetch_clan_fireteam( self, access_token: 'str', fireteam_id: 'int', group_id: 'int') -> 'typedefs.JSONObject': View Source

1779    async def fetch_clan_fireteam(
1780        self, access_token: str, fireteam_id: int, group_id: int
1781    ) -> typedefs.JSONObject:
1782        resp = await self._request(
1783            _GET,
1784            f"Fireteam/Clan/{group_id}/Summary/{fireteam_id}",
1785            auth=access_token,
1786        )
1787        assert isinstance(resp, dict)
1788        return resp

Fetch a specific clan fireteam.

This method requires OAuth2: ReadGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group/clan id to fetch the fireteam from.
fireteam_id (int): The fireteam id to fetch.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the fireteam details.

async def fetch_my_clan_fireteams( self, access_token: 'str', group_id: 'int', *, include_closed: 'bool' = True, platform: 'fireteams.FireteamPlatform | int', language: 'fireteams.FireteamLanguage | str', filtered: 'bool' = True, page: 'int' = 0) -> 'typedefs.JSONObject': View Source

1790    async def fetch_my_clan_fireteams(
1791        self,
1792        access_token: str,
1793        group_id: int,
1794        *,
1795        include_closed: bool = True,
1796        platform: fireteams.FireteamPlatform | int,
1797        language: fireteams.FireteamLanguage | str,
1798        filtered: bool = True,
1799        page: int = 0,
1800    ) -> typedefs.JSONObject:
1801        payload = {"groupFilter": filtered, "langFilter": str(language)}
1802
1803        resp = await self._request(
1804            _GET,
1805            f"Fireteam/Clan/{group_id}/My/{int(platform)}/{include_closed}/{page}",
1806            json=payload,
1807            auth=access_token,
1808        )
1809        assert isinstance(resp, dict)
1810        return resp

Fetch a clan's fireteams with open slots.

This method requires OAuth2: ReadGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group/clan id to fetch.

Other Parameters

include_closed (bool): If provided and set to True, It will also return closed fireteams. If provided and set to False, It will only return public fireteams. Default is True.
platform (aiobungie.aiobungie.crates.fireteams.FireteamPlatform | int): If this is provided. Then the results will be filtered with the given platform. Defaults to aiobungie.crates.FireteamPlatform.ANY which returns all platforms.
language (aiobungie.crates.fireteams.FireteamLanguage | str): A locale language to filter the used language in that fireteam. Defaults to aiobungie.crates.FireteamLanguage.ALL
filtered (bool): If set to True, it will filter by clan. Otherwise not. Default is True.
page (int): The page number. By default its 0 which returns all available activities.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the fireteams detail.

async def fetch_private_clan_fireteams(self, access_token: 'str', group_id: 'int', /) -> 'int': View Source

1812    async def fetch_private_clan_fireteams(
1813        self, access_token: str, group_id: int, /
1814    ) -> int:
1815        resp = await self._request(
1816            _GET,
1817            f"Fireteam/Clan/{group_id}/ActiveCount",
1818            auth=access_token,
1819        )
1820        assert isinstance(resp, int)
1821        return resp

Fetch the active count of the clan fireteams that are only private.

This method requires OAuth2: ReadGroups scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
group_id (int): The group/clan id.

Returns

int: The active fireteams count. Max value returned is 25.

async def fetch_post_activity(self, instance_id: 'int', /) -> 'typedefs.JSONObject': View Source

1823    async def fetch_post_activity(self, instance_id: int, /) -> typedefs.JSONObject:
1824        resp = await self._request(
1825            _GET, f"Destiny2/Stats/PostGameCarnageReport/{instance_id}"
1826        )
1827        assert isinstance(resp, dict)
1828        return resp

Fetch a post activity details.

Parameters

instance_id (int): The activity instance id.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the post activity.

@helpers.unstable

async def search_entities( self, name: 'str', entity_type: 'str', *, page: 'int' = 0) -> 'typedefs.JSONObject': View Source

1830    @helpers.unstable
1831    async def search_entities(
1832        self, name: str, entity_type: str, *, page: int = 0
1833    ) -> typedefs.JSONObject:
1834        resp = await self._request(
1835            _GET,
1836            f"Destiny2/Armory/Search/{entity_type}/{name}/",
1837            json={"page": page},
1838        )
1839        assert isinstance(resp, dict)
1840        return resp

Search for Destiny2 entities given a name and its type.

Parameters

name (str): The name of the entity, i.e., Thunderlord, One thousand voices.
entity_type (str): The type of the entity, AKA Definition, For an example DestinyInventoryItemDefinition

Other Parameters

page (int): An optional page to return. Default to 0.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains details about the searched term.

async def fetch_unique_weapon_history( self, membership_id: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

1842    async def fetch_unique_weapon_history(
1843        self,
1844        membership_id: int,
1845        character_id: int,
1846        membership_type: enums.MembershipType | int,
1847    ) -> typedefs.JSONObject:
1848        resp = await self._request(
1849            _GET,
1850            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/UniqueWeapons/",
1851        )
1852        assert isinstance(resp, dict)
1853        return resp

Fetch details about unique weapon usage for a character. Includes all exotics.

Parameters

membership_id (int): The Destiny user membership id.
character_id (int): The character id to retrieve.
membership_type (aiobungie.aiobungie.MembershipType | int): The Destiny user's membership type.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains details about the returned weapons.

async def fetch_item( self, member_id: 'int', item_id: 'int', membership_type: 'enums.MembershipType | int', components: 'collections.Sequence[enums.ComponentType]') -> 'typedefs.JSONObject': View Source

1855    async def fetch_item(
1856        self,
1857        member_id: int,
1858        item_id: int,
1859        membership_type: enums.MembershipType | int,
1860        components: collections.Sequence[enums.ComponentType],
1861    ) -> typedefs.JSONObject:
1862        collector = _collect_components(components)
1863
1864        resp = await self._request(
1865            _GET,
1866            f"Destiny2/{int(membership_type)}/Profile/{member_id}/Item/{item_id}/?components={collector}",
1867        )
1868        assert isinstance(resp, dict)
1869        return resp

Fetch an instanced Destiny 2 item's details.

Parameters

member_id (int): The membership id of the Destiny 2 player.
item_id (int): The instance id of the item.
membership_type (aiobungie.aiobungie.MembershipType | int): The membership type of the Destiny 2 player.
components (collections.Sequence[aiobungie.ComponentType]): A list of components to retrieve.

Returns

aiobungie.typedefs.JSONObject: A JSON object response contains the fetched item with its components.

async def fetch_clan_weekly_rewards(self, clan_id: 'int', /) -> 'typedefs.JSONObject': View Source

1871    async def fetch_clan_weekly_rewards(self, clan_id: int, /) -> typedefs.JSONObject:
1872        resp = await self._request(_GET, f"Destiny2/Clan/{clan_id}/WeeklyRewardState/")
1873        assert isinstance(resp, dict)
1874        return resp

Fetch the weekly reward state for a clan.

Parameters

clan_id (int): The clan id.

Returns

aiobungie.typedefs.JSONObject: A JSON response of the clan rewards state.

async def fetch_available_locales(self) -> 'typedefs.JSONObject': View Source

1876    async def fetch_available_locales(self) -> typedefs.JSONObject:
1877        resp = await self._request(_GET, "Destiny2/Manifest/DestinyLocaleDefinition/")
1878        assert isinstance(resp, dict)
1879        return resp

Fetch available locales at Bungie.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains a list of all available localization cultures.

async def fetch_common_settings(self) -> 'typedefs.JSONObject': View Source

1881    async def fetch_common_settings(self) -> typedefs.JSONObject:
1882        resp = await self._request(_GET, "Settings")
1883        assert isinstance(resp, dict)
1884        return resp

Fetch the common settings used by Bungie's environment.

Returns

aiobungie.typedefs.JSONObject: The common settings JSON object.

async def fetch_user_systems_overrides(self) -> 'typedefs.JSONObject': View Source

1886    async def fetch_user_systems_overrides(self) -> typedefs.JSONObject:
1887        resp = await self._request(_GET, "UserSystemOverrides")
1888        assert isinstance(resp, dict)
1889        return resp

Fetch a user's specific system overrides.

Returns

aiobungie.typedefs.JSONObject: The system overrides JSON object.

async def fetch_global_alerts(self, *, include_streaming: 'bool' = False) -> 'typedefs.JSONArray': View Source

1891    async def fetch_global_alerts(
1892        self, *, include_streaming: bool = False
1893    ) -> typedefs.JSONArray:
1894        resp = await self._request(
1895            _GET, f"GlobalAlerts/?includestreaming={include_streaming}"
1896        )
1897        assert isinstance(resp, list)
1898        return resp

Fetch any active global alerts.

Parameters

include_streaming (bool): If True, the returned results will include streaming alerts. Default is False.

Returns

aiobungie.typedefs.JSONArray: A JSON array of the global alerts objects.

async def awainitialize_request( self, access_token: 'str', type: 'typing.Literal[0, 1]', membership_type: 'enums.MembershipType | int', /, *, affected_item_id: 'int | None' = None, character_id: 'int | None' = None) -> 'typedefs.JSONObject': View Source

1900    async def awainitialize_request(
1901        self,
1902        access_token: str,
1903        type: typing.Literal[0, 1],
1904        membership_type: enums.MembershipType | int,
1905        /,
1906        *,
1907        affected_item_id: int | None = None,
1908        character_id: int | None = None,
1909    ) -> typedefs.JSONObject:
1910        body = {"type": type, "membershipType": int(membership_type)}
1911
1912        if affected_item_id is not None:
1913            body["affectedItemId"] = affected_item_id
1914
1915        if character_id is not None:
1916            body["characterId"] = character_id
1917
1918        resp = await self._request(
1919            _POST, "Destiny2/Awa/Initialize", json=body, auth=access_token
1920        )
1921        assert isinstance(resp, dict)
1922        return resp

Initialize a request to perform an advanced write action.

OAuth2: AdvancedWriteActions application scope is required to perform this request.

Parameters

access_token (str): The bearer access token associated with the bungie account.
type (typing.Literal[0, 1]): Type of the advanced write action. Its either 0 or 1. If set to 0 that means it None. Otherwise if 1 that means its insert plugs.
membership_type (aiobungie.aiobungie.MembershipType | int): The Destiny membership type of the account to modify.

Other Parameters

affected_item_id (int | None): Item instance ID the action shall be applied to. This is optional for all but a new AwaType values.
character_id (int | None): The Destiny character ID to perform this action on.

Returns

aiobungie.typedefs.JSONObject: A JSON object response.

async def awaget_action_token( self, access_token: 'str', correlation_id: 'str', /) -> 'typedefs.JSONObject': View Source

1924    async def awaget_action_token(
1925        self, access_token: str, correlation_id: str, /
1926    ) -> typedefs.JSONObject:
1927        resp = await self._request(
1928            _POST,
1929            f"Destiny2/Awa/GetActionToken/{correlation_id}",
1930            auth=access_token,
1931        )
1932        assert isinstance(resp, dict)
1933        return resp

Returns the action token if user approves the request.

OAuth2: AdvancedWriteActions application scope is required to perform this request.

Parameters

access_token (str): The bearer access token associated with the bungie account.
correlation_id (str): The identifier for the advanced write action request.

Returns

aiobungie.typedefs.JSONObject: A JSON object response.

async def awa_provide_authorization_result( self, access_token: 'str', selection: 'int', correlation_id: 'str', nonce: 'collections.MutableSequence[str | bytes]') -> 'int': View Source

1935    async def awa_provide_authorization_result(
1936        self,
1937        access_token: str,
1938        selection: int,
1939        correlation_id: str,
1940        nonce: collections.MutableSequence[str | bytes],
1941    ) -> int:
1942        body = {"selection": selection, "correlationId": correlation_id, "nonce": nonce}
1943
1944        resp = await self._request(
1945            _POST,
1946            "Destiny2/Awa/AwaProvideAuthorizationResult",
1947            json=body,
1948            auth=access_token,
1949        )
1950        assert isinstance(resp, int)
1951        return resp

Provide the result of the user interaction. Called by the Bungie Destiny App to approve or reject a request.

OAuth2: AdvancedWriteActions application scope is required to perform this request.

Parameters

access_token (str): The bearer access token associated with the bungie account.
selection (int): Indication of the selection the user has made (Approving or rejecting the action)
correlation_id (str): Correlation ID of the request.
nonce (collections.MutableSequence[str | bytes]): Secret nonce received via the PUSH notification.

Returns

int: ...

async def fetch_vendors( self, access_token: 'str', character_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', /, components: 'collections.Sequence[enums.ComponentType]', filter: 'int | None' = None) -> 'typedefs.JSONObject': View Source

1953    async def fetch_vendors(
1954        self,
1955        access_token: str,
1956        character_id: int,
1957        membership_id: int,
1958        membership_type: enums.MembershipType | int,
1959        /,
1960        components: collections.Sequence[enums.ComponentType],
1961        filter: int | None = None,
1962    ) -> typedefs.JSONObject:
1963        components_ = _collect_components(components)
1964        route = (
1965            f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1966            f"/Character/{character_id}/Vendors/?components={components_}"
1967        )
1968
1969        if filter is not None:
1970            route = route + f"&filter={filter}"
1971
1972        resp = await self._request(
1973            _GET,
1974            route,
1975            auth=access_token,
1976        )
1977        assert isinstance(resp, dict)
1978        return resp

Get currently available vendors from the list of vendors that can possibly have rotating inventory.

Parameters

access_token (str): The bearer access token associated with the bungie account.
character_id (int): The character ID to return the vendor info for.
membership_id (int): The Destiny membership id to return the vendor info for.
membership_type (aiobungie.aiobungie.MembershipType | int): The Destiny membership type to return the vendor info for.
components (collections.Sequence[aiobungie.ComponentType]): A list of vendor components to collect and return.

Other Parameters

filter (int): Filters the type of items returned from the vendor. This can be left to None.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the vendor response.

async def fetch_vendor( self, access_token: 'str', character_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', vendor_hash: 'int', /, components: 'collections.Sequence[enums.ComponentType]') -> 'typedefs.JSONObject': View Source

1980    async def fetch_vendor(
1981        self,
1982        access_token: str,
1983        character_id: int,
1984        membership_id: int,
1985        membership_type: enums.MembershipType | int,
1986        vendor_hash: int,
1987        /,
1988        components: collections.Sequence[enums.ComponentType],
1989    ) -> typedefs.JSONObject:
1990        components_ = _collect_components(components)
1991        resp = await self._request(
1992            _GET,
1993            (
1994                f"Destiny2/{int(membership_type)}/Profile/{membership_id}"
1995                f"/Character/{character_id}/Vendors/{vendor_hash}/?components={components_}"
1996            ),
1997            auth=access_token,
1998        )
1999        assert isinstance(resp, dict)
2000        return resp

Fetch details for a specific vendor.

Parameters

access_token (str): The bearer access token associated with the bungie account.
character_id (int): The character ID to return the vendor info for.
membership_id (int): The Destiny membership id to return the vendor info for.
membership_type (aiobungie.aiobungie.MembershipType | int): The Destiny membership type to return the vendor info for.
vendor_hash (int): The vendor hash to return the details for.
components (collections.Sequence[aiobungie.ComponentType]): A list of vendor components to collect and return.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the vendor response.

async def fetch_application_api_usage( self, access_token: 'str', application_id: 'int', /, *, start: 'datetime.datetime | None' = None, end: 'datetime.datetime | None' = None) -> 'typedefs.JSONObject': View Source

2002    async def fetch_application_api_usage(
2003        self,
2004        access_token: str,
2005        application_id: int,
2006        /,
2007        *,
2008        start: datetime.datetime | None = None,
2009        end: datetime.datetime | None = None,
2010    ) -> typedefs.JSONObject:
2011        end_date, start_date = time.parse_date_range(end, start)
2012        resp = await self._request(
2013            _GET,
2014            f"App/ApiUsage/{application_id}/?end={end_date}&start={start_date}",
2015            auth=access_token,
2016        )
2017        assert isinstance(resp, dict)
2018        return resp

Fetch a Bungie application's API usage.

Parameters

access_token (str): The bearer access token associated with the bungie account.
application_id (int): The application id to get.

Other Parameters

start (datetime.datetime | None): A datetime object can be used to collect the start of the application usage. This is limited and can go back to 30 days maximum.

If this is left to None. It will return the last 24 hours.
end (datetime.datetime | None): A datetime object can be used to collect the end of the application usage.

If this is left to None. It will return now.

Example

import datetime

# Fetch data from 2021 Dec 10th to 2021 Dec 20th
await fetch_application_api_usage(
    start=datetime.datetime(2021, 12, 10),
    end=datetime.datetime(2021, 12, 20)
)

Returns

aiobungie.typedefs.JSONObject: A JSON object of the application usage details.

async def fetch_bungie_applications(self) -> 'typedefs.JSONArray': View Source

2020    async def fetch_bungie_applications(self) -> typedefs.JSONArray:
2021        resp = await self._request(_GET, "App/FirstParty")
2022        assert isinstance(resp, list)
2023        return resp

Fetch details for applications created by Bungie.

Returns

aiobungie.typedefs.JSONArray: An array of Bungie created applications.

async def fetch_content_type(self, type: 'str', /) -> 'typedefs.JSONObject': View Source

2025    async def fetch_content_type(self, type: str, /) -> typedefs.JSONObject:
2026        resp = await self._request(_GET, f"Content/GetContentType/{type}/")
2027        assert isinstance(resp, dict)
2028        return resp

async def fetch_content_by_id( self, id: 'int', locale: 'str', /, *, head: 'bool' = False) -> 'typedefs.JSONObject': View Source

2030    async def fetch_content_by_id(
2031        self, id: int, locale: str, /, *, head: bool = False
2032    ) -> typedefs.JSONObject:
2033        resp = await self._request(
2034            _GET,
2035            f"Content/GetContentById/{id}/{locale}/",
2036            json={"head": head},
2037        )
2038        assert isinstance(resp, dict)
2039        return resp

async def fetch_content_by_tag_and_type( self, locale: 'str', tag: 'str', type: 'str', *, head: 'bool' = False) -> 'typedefs.JSONObject': View Source

2041    async def fetch_content_by_tag_and_type(
2042        self, locale: str, tag: str, type: str, *, head: bool = False
2043    ) -> typedefs.JSONObject:
2044        resp = await self._request(
2045            _GET,
2046            f"Content/GetContentByTagAndType/{tag}/{type}/{locale}/",
2047            json={"head": head},
2048        )
2049        assert isinstance(resp, dict)
2050        return resp

async def search_content_with_text( self, locale: 'str', /, content_type: 'str', search_text: 'str', tag: 'str', *, page: 'int | None' = None, source: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2052    async def search_content_with_text(
2053        self,
2054        locale: str,
2055        /,
2056        content_type: str,
2057        search_text: str,
2058        tag: str,
2059        *,
2060        page: int | None = None,
2061        source: str | None = None,
2062    ) -> typedefs.JSONObject:
2063        body: typedefs.JSONObject = {
2064            "locale": locale,
2065            "currentpage": page or 1,
2066            "ctype": content_type,
2067            "searchtxt": search_text,
2068            "searchtext": search_text,
2069            "tag": tag,
2070            "source": source,
2071        }
2072
2073        resp = await self._request(_GET, "Content/Search", params=body)
2074        assert isinstance(resp, dict)
2075        return resp

async def search_content_by_tag_and_type( self, locale: 'str', tag: 'str', type: 'str', *, page: 'int | None' = None) -> 'typedefs.JSONObject': View Source

2077    async def search_content_by_tag_and_type(
2078        self,
2079        locale: str,
2080        tag: str,
2081        type: str,
2082        *,
2083        page: int | None = None,
2084    ) -> typedefs.JSONObject:
2085        body: typedefs.JSONObject = {"currentpage": page or 1}
2086
2087        resp = await self._request(
2088            _GET,
2089            f"Content/SearchContentByTagAndType/{tag}/{type}/{locale}/",
2090            params=body,
2091        )
2092        assert isinstance(resp, dict)
2093        return resp

async def search_help_articles(self, text: 'str', size: 'str', /) -> 'typedefs.JSONObject': View Source

2095    async def search_help_articles(
2096        self, text: str, size: str, /
2097    ) -> typedefs.JSONObject:
2098        resp = await self._request(_GET, f"Content/SearchHelpArticles/{text}/{size}/")
2099        assert isinstance(resp, dict)
2100        return resp

async def fetch_topics_page( self, category_filter: 'int', group: 'int', date_filter: 'int', sort: 'str | bytes', *, page: 'int | None' = None, locales: 'collections.Iterable[str] | None' = None, tag_filter: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2102    async def fetch_topics_page(
2103        self,
2104        category_filter: int,
2105        group: int,
2106        date_filter: int,
2107        sort: str | bytes,
2108        *,
2109        page: int | None = None,
2110        locales: collections.Iterable[str] | None = None,
2111        tag_filter: str | None = None,
2112    ) -> typedefs.JSONObject:
2113        params = {
2114            "locales": ",".join(locales) if locales is not None else "en",
2115        }
2116        if tag_filter:
2117            params["tagstring"] = tag_filter
2118
2119        resp = await self._request(
2120            _GET,
2121            f"Forum/GetTopicsPaged/{page or 0}/0/{group}/{sort!s}/{date_filter}/{category_filter}/",
2122            params=params,
2123        )
2124        assert isinstance(resp, dict)
2125        return resp

async def fetch_core_topics_page( self, category_filter: 'int', date_filter: 'int', sort: 'str | bytes', *, page: 'int | None' = None, locales: 'collections.Iterable[str] | None' = None) -> 'typedefs.JSONObject': View Source

2127    async def fetch_core_topics_page(
2128        self,
2129        category_filter: int,
2130        date_filter: int,
2131        sort: str | bytes,
2132        *,
2133        page: int | None = None,
2134        locales: collections.Iterable[str] | None = None,
2135    ) -> typedefs.JSONObject:
2136        resp = await self._request(
2137            _GET,
2138            f"Forum/GetCoreTopicsPaged/{page or 0}"
2139            f"/{sort!s}/{date_filter}/{category_filter}/?locales={','.join(locales) if locales else 'en'}",
2140        )
2141        assert isinstance(resp, dict)
2142        return resp

async def fetch_posts_threaded_page( self, parent_post: 'bool', page: 'int', page_size: 'int', parent_post_id: 'int', reply_size: 'int', root_thread_mode: 'bool', sort_mode: 'int', show_banned: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2144    async def fetch_posts_threaded_page(
2145        self,
2146        parent_post: bool,
2147        page: int,
2148        page_size: int,
2149        parent_post_id: int,
2150        reply_size: int,
2151        root_thread_mode: bool,
2152        sort_mode: int,
2153        show_banned: str | None = None,
2154    ) -> typedefs.JSONObject:
2155        resp = await self._request(
2156            _GET,
2157            f"Forum/GetPostsThreadedPaged/{parent_post}/{page}/"
2158            f"{page_size}/{reply_size}/{parent_post_id}/{root_thread_mode}/{sort_mode}/",
2159            json={"showbanned": show_banned},
2160        )
2161        assert isinstance(resp, dict)
2162        return resp

async def fetch_posts_threaded_page_from_child( self, child_id: 'bool', page: 'int', page_size: 'int', reply_size: 'int', root_thread_mode: 'bool', sort_mode: 'int', show_banned: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2164    async def fetch_posts_threaded_page_from_child(
2165        self,
2166        child_id: bool,
2167        page: int,
2168        page_size: int,
2169        reply_size: int,
2170        root_thread_mode: bool,
2171        sort_mode: int,
2172        show_banned: str | None = None,
2173    ) -> typedefs.JSONObject:
2174        resp = await self._request(
2175            _GET,
2176            f"Forum/GetPostsThreadedPagedFromChild/{child_id}/"
2177            f"{page}/{page_size}/{reply_size}/{root_thread_mode}/{sort_mode}/",
2178            json={"showbanned": show_banned},
2179        )
2180        assert isinstance(resp, dict)
2181        return resp

async def fetch_post_and_parent( self, child_id: 'int', /, *, show_banned: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2183    async def fetch_post_and_parent(
2184        self, child_id: int, /, *, show_banned: str | None = None
2185    ) -> typedefs.JSONObject:
2186        resp = await self._request(
2187            _GET,
2188            f"Forum/GetPostAndParent/{child_id}/",
2189            json={"showbanned": show_banned},
2190        )
2191        assert isinstance(resp, dict)
2192        return resp

async def fetch_posts_and_parent_awaiting( self, child_id: 'int', /, *, show_banned: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2194    async def fetch_posts_and_parent_awaiting(
2195        self, child_id: int, /, *, show_banned: str | None = None
2196    ) -> typedefs.JSONObject:
2197        resp = await self._request(
2198            _GET,
2199            f"Forum/GetPostAndParentAwaitingApproval/{child_id}/",
2200            json={"showbanned": show_banned},
2201        )
2202        assert isinstance(resp, dict)
2203        return resp

async def fetch_topic_for_content(self, content_id: 'int', /) -> 'int': View Source

2205    async def fetch_topic_for_content(self, content_id: int, /) -> int:
2206        resp = await self._request(_GET, f"Forum/GetTopicForContent/{content_id}/")
2207        assert isinstance(resp, int)
2208        return resp

async def fetch_forum_tag_suggestions(self, partial_tag: 'str', /) -> 'typedefs.JSONObject': View Source

2210    async def fetch_forum_tag_suggestions(
2211        self, partial_tag: str, /
2212    ) -> typedefs.JSONObject:
2213        resp = await self._request(
2214            _GET,
2215            "Forum/GetForumTagSuggestions/",
2216            json={"partialtag": partial_tag},
2217        )
2218        assert isinstance(resp, dict)
2219        return resp

async def fetch_poll(self, topic_id: 'int', /) -> 'typedefs.JSONObject': View Source

2221    async def fetch_poll(self, topic_id: int, /) -> typedefs.JSONObject:
2222        resp = await self._request(_GET, f"Forum/Poll/{topic_id}/")
2223        assert isinstance(resp, dict)
2224        return resp

async def fetch_recruitment_thread_summaries(self) -> 'typedefs.JSONArray': View Source

2226    async def fetch_recruitment_thread_summaries(self) -> typedefs.JSONArray:
2227        resp = await self._request(_POST, "Forum/Recruit/Summaries/")
2228        assert isinstance(resp, list)
2229        return resp

async def fetch_recommended_groups( self, access_token: 'str', /, *, date_range: 'int' = 0, group_type: 'enums.GroupType | int' = <GroupType.CLAN: 1>) -> 'typedefs.JSONArray': View Source

2231    async def fetch_recommended_groups(
2232        self,
2233        access_token: str,
2234        /,
2235        *,
2236        date_range: int = 0,
2237        group_type: enums.GroupType | int = enums.GroupType.CLAN,
2238    ) -> typedefs.JSONArray:
2239        resp = await self._request(
2240            _POST,
2241            f"GroupV2/Recommended/{int(group_type)}/{date_range}/",
2242            auth=access_token,
2243        )
2244        assert isinstance(resp, list)
2245        return resp

async def fetch_available_avatars(self) -> 'collections.Mapping[str, int]': View Source

2247    async def fetch_available_avatars(self) -> collections.Mapping[str, int]:
2248        resp = await self._request(_GET, "GroupV2/GetAvailableAvatars/")
2249        assert isinstance(resp, dict)
2250        return resp

async def fetch_user_clan_invite_setting( self, access_token: 'str', /, membership_type: 'enums.MembershipType | int') -> 'bool': View Source

2252    async def fetch_user_clan_invite_setting(
2253        self,
2254        access_token: str,
2255        /,
2256        membership_type: enums.MembershipType | int,
2257    ) -> bool:
2258        resp = await self._request(
2259            _GET,
2260            f"GroupV2/GetUserClanInviteSetting/{int(membership_type)}/",
2261            auth=access_token,
2262        )
2263        assert isinstance(resp, bool)
2264        return resp

async def fetch_banned_group_members( self, access_token: 'str', group_id: 'int', /, *, page: 'int' = 1) -> 'typedefs.JSONObject': View Source

2266    async def fetch_banned_group_members(
2267        self, access_token: str, group_id: int, /, *, page: int = 1
2268    ) -> typedefs.JSONObject:
2269        resp = await self._request(
2270            _GET,
2271            f"GroupV2/{group_id}/Banned/?currentpage={page}",
2272            auth=access_token,
2273        )
2274        assert isinstance(resp, dict)
2275        return resp

async def fetch_pending_group_memberships( self, access_token: 'str', group_id: 'int', /, *, current_page: 'int' = 1) -> 'typedefs.JSONObject': View Source

2277    async def fetch_pending_group_memberships(
2278        self, access_token: str, group_id: int, /, *, current_page: int = 1
2279    ) -> typedefs.JSONObject:
2280        resp = await self._request(
2281            _GET,
2282            f"GroupV2/{group_id}/Members/Pending/?currentpage={current_page}",
2283            auth=access_token,
2284        )
2285        assert isinstance(resp, dict)
2286        return resp

async def fetch_invited_group_memberships( self, access_token: 'str', group_id: 'int', /, *, current_page: 'int' = 1) -> 'typedefs.JSONObject': View Source

2288    async def fetch_invited_group_memberships(
2289        self, access_token: str, group_id: int, /, *, current_page: int = 1
2290    ) -> typedefs.JSONObject:
2291        resp = await self._request(
2292            _GET,
2293            f"GroupV2/{group_id}/Members/InvitedIndividuals/?currentpage={current_page}",
2294            auth=access_token,
2295        )
2296        assert isinstance(resp, dict)
2297        return resp

async def invite_member_to_group( self, access_token: 'str', /, group_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', *, message: 'str | None' = None) -> 'typedefs.JSONObject': View Source

2299    async def invite_member_to_group(
2300        self,
2301        access_token: str,
2302        /,
2303        group_id: int,
2304        membership_id: int,
2305        membership_type: enums.MembershipType | int,
2306        *,
2307        message: str | None = None,
2308    ) -> typedefs.JSONObject:
2309        resp = await self._request(
2310            _POST,
2311            f"GroupV2/{group_id}/Members/IndividualInvite/{int(membership_type)}/{membership_id}/",
2312            auth=access_token,
2313            json={"message": str(message)},
2314        )
2315        assert isinstance(resp, dict)
2316        return resp

async def cancel_group_member_invite( self, access_token: 'str', /, group_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

2318    async def cancel_group_member_invite(
2319        self,
2320        access_token: str,
2321        /,
2322        group_id: int,
2323        membership_id: int,
2324        membership_type: enums.MembershipType | int,
2325    ) -> typedefs.JSONObject:
2326        resp = await self._request(
2327            _POST,
2328            f"GroupV2/{group_id}/Members/IndividualInviteCancel/{int(membership_type)}/{membership_id}/",
2329            auth=access_token,
2330        )
2331        assert isinstance(resp, dict)
2332        return resp

async def fetch_historical_definition(self) -> 'typedefs.JSONObject': View Source

2334    async def fetch_historical_definition(self) -> typedefs.JSONObject:
2335        resp = await self._request(_GET, "Destiny2/Stats/Definition/")
2336        assert isinstance(resp, dict)
2337        return resp

async def fetch_historical_stats( self, character_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', day_start: 'datetime.datetime', day_end: 'datetime.datetime', groups: 'collections.Sequence[enums.StatsGroupType | int]', modes: 'collections.Sequence[enums.GameMode | int]', *, period_type: 'enums.PeriodType' = <PeriodType.ALL_TIME: 2>) -> 'typedefs.JSONObject': View Source

2339    async def fetch_historical_stats(
2340        self,
2341        character_id: int,
2342        membership_id: int,
2343        membership_type: enums.MembershipType | int,
2344        day_start: datetime.datetime,
2345        day_end: datetime.datetime,
2346        groups: collections.Sequence[enums.StatsGroupType | int],
2347        modes: collections.Sequence[enums.GameMode | int],
2348        *,
2349        period_type: enums.PeriodType = enums.PeriodType.ALL_TIME,
2350    ) -> typedefs.JSONObject:
2351        end, start = time.parse_date_range(day_end, day_start)
2352        resp = await self._request(
2353            _GET,
2354            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Character/{character_id}/Stats/",
2355            json={
2356                "dayend": end,
2357                "daystart": start,
2358                "groups": [str(int(group)) for group in groups],
2359                "modes": [str(int(mode)) for mode in modes],
2360                "periodType": int(period_type),
2361            },
2362        )
2363        assert isinstance(resp, dict)
2364        return resp

Fetch historical stats for a specific membership character.

Parameters

character_id (int): The character ID to return the stats for.
membership_id (int): The Destiny membership id to return the stats for.
membership_type (aiobungie.MembershipType | int): The Destiny membership type to return the stats for.
day_start (datetime.datetime): The start of the day to return the stats for.
day_end (datetime.datetime): The end of the day to return the stats for.
groups (collections.Sequence[aiobungie.StatsGroupType]): A list of stats groups to return.
modes (collections.Sequence[aiobungie.GameMode | int]): A list of game modes to return.
period_type (aiobungie.enums.PeriodType): The period type to return the stats for. This will return ALL_TIME by default if not modified.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the historical stats.

async def fetch_historical_stats_for_account( self, membership_id: 'int', membership_type: 'enums.MembershipType | int', groups: 'collections.Sequence[enums.StatsGroupType | int]') -> 'typedefs.JSONObject': View Source

2366    async def fetch_historical_stats_for_account(
2367        self,
2368        membership_id: int,
2369        membership_type: enums.MembershipType | int,
2370        groups: collections.Sequence[enums.StatsGroupType | int],
2371    ) -> typedefs.JSONObject:
2372        resp = await self._request(
2373            _GET,
2374            f"Destiny2/{int(membership_type)}/Account/{membership_id}/Stats/",
2375            json={"groups": [str(int(group)) for group in groups]},
2376        )
2377        assert isinstance(resp, dict)
2378        return resp

Fetch historical stats for an account's membership.

Parameters

membership_id (int): The Destiny membership id to return the stats for.
membership_type (aiobungie.MembershipType | int): The Destiny membership type to return the stats for.
groups (collections.Sequence[aiobungie.StatsGroupType]): A list of stats groups to return.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the historical stats for the account. This includes both the character and account stats.

async def fetch_aggregated_activity_stats( self, character_id: 'int', membership_id: 'int', membership_type: 'enums.MembershipType | int', /) -> 'typedefs.JSONObject': View Source

2380    async def fetch_aggregated_activity_stats(
2381        self,
2382        character_id: int,
2383        membership_id: int,
2384        membership_type: enums.MembershipType | int,
2385        /,
2386    ) -> typedefs.JSONObject:
2387        resp = await self._request(
2388            _GET,
2389            f"Destiny2/{int(membership_type)}/Account/{membership_id}/"
2390            f"Character/{character_id}/Stats/AggregateActivityStats/",
2391        )
2392        assert isinstance(resp, dict)
2393        return resp

Fetch aggregated activity stats for a specific membership character.

Parameters

character_id (int): The character ID to return the stats for.
membership_id (int): The Destiny membership id to return the stats for.
membership_type (aiobungie.MembershipType | int): The Destiny membership type to return the stats for.

Returns

aiobungie.typedefs.JSONObject: A JSON object of the aggregated activity stats.

async def equip_loadout( self, access_token: 'str', /, loadout_index: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'None': View Source

2395    async def equip_loadout(
2396        self,
2397        access_token: str,
2398        /,
2399        loadout_index: int,
2400        character_id: int,
2401        membership_type: enums.MembershipType | int,
2402    ) -> None:
2403        response = await self._request(
2404            _POST,
2405            "Destiny2/Actions/Loadouts/EquipLoadout/",
2406            json={
2407                "loadoutIndex": loadout_index,
2408                "characterId": character_id,
2409                "membership_type": int(membership_type),
2410            },
2411            auth=access_token,
2412        )
2413        assert isinstance(response, int)

Equip a loadout. Your character must be in a Social space, Orbit or Offline while performing this operation.

This operation requires MoveEquipDestinyItems OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
loadout_index (int): The index of the loadout to use.
character_id (int): The character ID to equip the loadout to.
membership_type (aiobungie.MembershipType | int): The membership type of the account.

async def snapshot_loadout( self, access_token: 'str', /, loadout_index: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int', *, color_hash: 'int | None' = None, icon_hash: 'int | None' = None, name_hash: 'int | None' = None) -> 'None': View Source

2415    async def snapshot_loadout(
2416        self,
2417        access_token: str,
2418        /,
2419        loadout_index: int,
2420        character_id: int,
2421        membership_type: enums.MembershipType | int,
2422        *,
2423        color_hash: int | None = None,
2424        icon_hash: int | None = None,
2425        name_hash: int | None = None,
2426    ) -> None:
2427        response = await self._request(
2428            _POST,
2429            "Destiny2/Actions/Loadouts/SnapshotLoadout/",
2430            auth=access_token,
2431            json={
2432                "colorHash": color_hash,
2433                "iconHash": icon_hash,
2434                "nameHash": name_hash,
2435                "loadoutIndex": loadout_index,
2436                "characterId": character_id,
2437                "membershipType": int(membership_type),
2438            },
2439        )
2440        assert isinstance(response, int)

Snapshot a loadout with the currently equipped items.

This operation requires MoveEquipDestinyItems OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
loadout_index (int): The index of the loadout to use.
character_id (int): The character ID to equip the loadout to.
membership_type (aiobungie.MembershipType | int): The membership type of the account.

Other Parameters

color_hash (int | None): ...
icon_hash (int | None): ...
name_hash (int | None): ...

async def update_loadout( self, access_token: 'str', /, loadout_index: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int', *, color_hash: 'int | None' = None, icon_hash: 'int | None' = None, name_hash: 'int | None' = None) -> 'None': View Source

2442    async def update_loadout(
2443        self,
2444        access_token: str,
2445        /,
2446        loadout_index: int,
2447        character_id: int,
2448        membership_type: enums.MembershipType | int,
2449        *,
2450        color_hash: int | None = None,
2451        icon_hash: int | None = None,
2452        name_hash: int | None = None,
2453    ) -> None:
2454        response = await self._request(
2455            _POST,
2456            "Destiny2/Actions/Loadouts/UpdateLoadoutIdentifiers/",
2457            auth=access_token,
2458            json={
2459                "colorHash": color_hash,
2460                "iconHash": icon_hash,
2461                "nameHash": name_hash,
2462                "loadoutIndex": loadout_index,
2463                "characterId": character_id,
2464                "membershipType": int(membership_type),
2465            },
2466        )
2467        assert isinstance(response, int)

Update the loadout. Color, Icon and Name.

This operation requires MoveEquipDestinyItems OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
loadout_index (int): The index of the loadout to use.
character_id (int): The character ID to equip the loadout to.
membership_type (aiobungie.MembershipType | int): The membership type of the account.

Other Parameters

color_hash (int | None): The new color hash of the loadout to update.
icon_hash (int | None): The new icon hash of the loadout to update.
name_hash (int | None): The new name hash of the loadout to update.

async def clear_loadout( self, access_token: 'str', /, loadout_index: 'int', character_id: 'int', membership_type: 'enums.MembershipType | int') -> 'None': View Source

2469    async def clear_loadout(
2470        self,
2471        access_token: str,
2472        /,
2473        loadout_index: int,
2474        character_id: int,
2475        membership_type: enums.MembershipType | int,
2476    ) -> None:
2477        response = await self._request(
2478            _POST,
2479            "Destiny2/Actions/Loadouts/ClearLoadout/",
2480            json={
2481                "loadoutIndex": loadout_index,
2482                "characterId": character_id,
2483                "membership_type": int(membership_type),
2484            },
2485            auth=access_token,
2486        )
2487        assert isinstance(response, int)

Clear the identifiers and items of a loadout.

This operation requires MoveEquipDestinyItems OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account.
loadout_index (int): The index of the loadout to use.
character_id (int): The character ID to equip the loadout to.
membership_type (aiobungie.MembershipType | int): The membership type of the account.

async def force_drops_repair(self, access_token: 'str', /) -> 'bool': View Source

2489    async def force_drops_repair(self, access_token: str, /) -> bool:
2490        response = await self._request(
2491            _POST, "Tokens/Partner/ForceDropsRepair/", auth=access_token
2492        )
2493        assert isinstance(response, bool)
2494        return response

Twitch Drops self-repair function - scans twitch for drops not marked as fulfilled and resyncs them.

This operation requires PartnerOfferGrant OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account that will be used to make the request with.

Returns

bool: The nature of this response is a boolean.

async def claim_partner_offer( self, access_token: 'str', /, *, offer_id: 'str', bungie_membership_id: 'int', transaction_id: 'str') -> 'bool': View Source

2496    async def claim_partner_offer(
2497        self,
2498        access_token: str,
2499        /,
2500        *,
2501        offer_id: str,
2502        bungie_membership_id: int,
2503        transaction_id: str,
2504    ) -> bool:
2505        response = await self._request(
2506            _POST,
2507            "Tokens/Partner/ClaimOffer/",
2508            json={
2509                "PartnerOfferId": offer_id,
2510                "BungieNetMembershipId": bungie_membership_id,
2511                "TransactionId": transaction_id,
2512            },
2513            auth=access_token,
2514        )
2515        assert isinstance(response, bool)
2516        return response

Claim a partner offer as the authenticated user.

This operation requires PartnerOfferGrant OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account that will be used to make the request with.
offer_id (str): The partner offer ID
bungie_membership_id (int): The associated Bungie.net membership ID
transaction_id (str): The transaction ID

Returns

bool: The nature of this response is a boolean.

async def fetch_bungie_rewards_for_user( self, access_token: 'str', /, membership_id: 'int') -> 'typedefs.JSONObject': View Source

2518    async def fetch_bungie_rewards_for_user(
2519        self, access_token: str, /, membership_id: int
2520    ) -> typedefs.JSONObject:
2521        response = await self._request(
2522            _GET,
2523            f"Tokens/Rewards/GetRewardsForUser/{membership_id}/",
2524            auth=access_token,
2525        )
2526        assert isinstance(response, dict)
2527        return response

Returns the bungie rewards for the targeted user.

This operation requires ReadAndApplyTokens OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account that will be used to make the request with.
membership_id (int): The associated membership ID to fetch the rewards for.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains a display of the rewards. See

async def fetch_bungie_rewards_for_platform( self, access_token: 'str', /, membership_id: 'int', membership_type: 'enums.MembershipType | int') -> 'typedefs.JSONObject': View Source

2529    async def fetch_bungie_rewards_for_platform(
2530        self,
2531        access_token: str,
2532        /,
2533        membership_id: int,
2534        membership_type: enums.MembershipType | int,
2535    ) -> typedefs.JSONObject:
2536        response = await self._request(
2537            _GET,
2538            f"Tokens/Rewards/GetRewardsForPlatformUser/{membership_id}/{int(membership_type)}",
2539            auth=access_token,
2540        )
2541        assert isinstance(response, dict)
2542        return response

Returns the bungie rewards for the targeted user and membership.

This operation requires ReadAndApplyTokens OAuth2 scope.

Parameters

access_token (str): The bearer access token associated with the bungie account that will be used to make the request with.
membership_id (int): The associated membership ID to fetch the rewards for.
membership_type (aiobungie.MembershipType | int): The associated membership type for the user.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains a display of the rewards. See

async def fetch_bungie_rewards(self) -> 'typedefs.JSONObject': View Source

2544    async def fetch_bungie_rewards(self) -> typedefs.JSONObject:
2545        response = await self._request(_GET, "Tokens/Rewards/BungieRewards/")
2546        assert isinstance(response, dict)
2547        return response

Returns a list of the current bungie rewards.

Returns

aiobungie.typedefs.JSONObject: A JSON object contains a display of the rewards. See

async def fetch_fireteam_listing(self, listing_id: 'int') -> 'typedefs.JSONObject': View Source

2549    async def fetch_fireteam_listing(self, listing_id: int) -> typedefs.JSONObject:
2550        response = await self._request(_GET, f"FireteamFinder/Listing/{listing_id}/")
2551        assert isinstance(response, dict)
2552        return response

Inherited Members

aiobungie.traits.RESTful: with_debug; build_fireteam_finder

class RESTPool: View Source

202class RESTPool:
203    """a Pool of `RESTClient` instances that shares the same TCP client connection.
204
205    This allows you to acquire instances of `RESTClient`s from single settings and credentials.
206
207    Example
208    -------
209    ```py
210    import aiobungie
211    import asyncio
212
213    pool = aiobungie.RESTPool("token")
214
215    async def get() -> None:
216        await pool.start()
217
218        async with pool.acquire() as client:
219            await client.fetch_character(...)
220
221        await pool.stop()
222
223    asyncio.run(get())
224    ```
225
226    Parameters
227    ----------
228    token : `str`
229        A valid application token from Bungie's developer portal.
230
231    Other Parameters
232    ----------------
233    client_secret : `str | None`
234        An optional application client secret,
235        This is only needed if you're fetching OAuth2 tokens with this client.
236    client_id : `int | None`
237        An optional application client id,
238        This is only needed if you're fetching OAuth2 tokens with this client.
239    settings: `aiobungie.builders.Settings | None`
240        The client settings to use, if `None` the default will be used.
241    max_retries : `int`
242        The max retries number to retry if the request hit a `5xx` status code.
243    debug : `bool | str`
244        Whether to enable logging responses or not.
245
246    Logging Levels
247    --------------
248    * `False`: This will disable logging.
249    * `True`: This will set the level to `DEBUG` and enable logging minimal information.
250    Like the response status, route, taken time and so on.
251    * `"TRACE" | aiobungie.TRACE`: This will log the response headers along with the minimal information.
252    """
253
254    __slots__ = (
255        "_token",
256        "_max_retries",
257        "_client_secret",
258        "_client_id",
259        "_metadata",
260        "_enable_debug",
261        "_client_session",
262        "_loads",
263        "_dumps",
264        "_settings",
265    )
266
267    # Looks like mypy doesn't like this.
268    if typing.TYPE_CHECKING:
269        _enable_debug: typing.Literal["TRACE"] | bool | int
270
271    def __init__(
272        self,
273        token: str,
274        /,
275        *,
276        client_secret: str | None = None,
277        client_id: int | None = None,
278        settings: builders.Settings | None = None,
279        dumps: typedefs.Dumps = helpers.dumps,
280        loads: typedefs.Loads = helpers.loads,
281        max_retries: int = 4,
282        debug: typing.Literal["TRACE"] | bool | int = False,
283    ) -> None:
284        self._client_secret = client_secret
285        self._client_id = client_id
286        self._token = token
287        self._max_retries = max_retries
288        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
289        self._enable_debug = debug
290        self._client_session: aiohttp.ClientSession | None = None
291        self._loads = loads
292        self._dumps = dumps
293        self._settings = settings or builders.Settings()
294
295    @property
296    def client_id(self) -> int | None:
297        """Return the client id of this REST client if provided, Otherwise None."""
298        return self._client_id
299
300    @property
301    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
302        """A general-purpose mutable mapping you can use to store data.
303
304        This mapping can be accessed from any process that has a reference to this pool.
305        """
306        return self._metadata
307
308    @property
309    def settings(self) -> builders.Settings:
310        """Internal client settings used within the HTTP client session."""
311        return self._settings
312
313    @typing.overload
314    def build_oauth2_url(self, client_id: int) -> builders.OAuthURL: ...
315
316    @typing.overload
317    def build_oauth2_url(self) -> builders.OAuthURL | None: ...
318
319    @typing.final
320    def build_oauth2_url(
321        self, client_id: int | None = None
322    ) -> builders.OAuthURL | None:
323        """Construct a new `OAuthURL` url object.
324
325        You can get the complete string representation of the url by calling `.compile()` on it.
326
327        Parameters
328        ----------
329        client_id : `int | None`
330            An optional client id to provide, If left `None` it will roll back to the id passed
331            to the `RESTClient`, If both is `None` this method will return `None`.
332
333        Returns
334        -------
335        `aiobungie.builders.OAuthURL | None`
336            * If `client_id` was provided as a parameter, It guarantees to return a complete `OAuthURL` object
337            * If `client_id` is set to `aiobungie.RESTClient` will be.
338            * If both are `None` this method will return `None.
339        """
340        client_id = client_id or self._client_id
341        if client_id is None:
342            return None
343
344        return builders.OAuthURL(client_id=client_id)
345
346    async def start(self) -> None:
347        """Start the TCP connection of this client pool.
348
349        This will raise `RuntimeError` if the connection has already been started.
350
351        Example
352        -------
353        ```py
354        pool = aiobungie.RESTPool(...)
355
356        async def run() -> None:
357            await pool.start()
358            async with pool.acquire() as client:
359                # use client
360
361        async def stop(self) -> None:
362            await pool.close()
363        ```
364        """
365        if self._client_session is not None:
366            raise RuntimeError("<RESTPool> has already been started.") from None
367
368        self._client_session = aiohttp.ClientSession(
369            connector=aiohttp.TCPConnector(
370                use_dns_cache=self._settings.use_dns_cache,
371                ttl_dns_cache=self._settings.ttl_dns_cache,
372                ssl_context=self._settings.ssl_context,
373                ssl=self._settings.ssl,
374            ),
375            connector_owner=True,
376            raise_for_status=False,
377            timeout=self._settings.http_timeout,
378            trust_env=self._settings.trust_env,
379            headers=self._settings.headers,
380        )
381
382    async def stop(self) -> None:
383        """Stop the TCP connection of this client pool.
384
385        This will raise `RuntimeError` if the connection has already been closed.
386
387        Example
388        -------
389        ```py
390        pool = aiobungie.RESTPool(...)
391
392        async def run() -> None:
393            await pool.start()
394            async with pool.acquire() as client:
395                # use client
396
397        async def stop(self) -> None:
398            await pool.close()
399        ```
400        """
401        if self._client_session is None:
402            raise RuntimeError("<RESTPool> is already stopped.")
403
404        await self._client_session.close()
405        self._client_session = None
406
407    @typing.final
408    def acquire(self) -> RESTClient:
409        """Acquires a new `RESTClient` instance from this pool.
410
411        Returns
412        -------
413        `RESTClient`
414            An instance of a `RESTClient`.
415        """
416        return RESTClient(
417            self._token,
418            client_secret=self._client_secret,
419            client_id=self._client_id,
420            loads=self._loads,
421            dumps=self._dumps,
422            max_retries=self._max_retries,
423            debug=self._enable_debug,
424            client_session=self._client_session,
425            owned_client=False,
426            settings=self._settings,
427        )

a Pool of RESTClient instances that shares the same TCP client connection.

This allows you to acquire instances of RESTClients from single settings and credentials.

Example

import aiobungie
import asyncio

pool = aiobungie.RESTPool("token")

async def get() -> None:
    await pool.start()

    async with pool.acquire() as client:
        await client.fetch_character(...)

    await pool.stop()

asyncio.run(get())

Parameters

token (str): A valid application token from Bungie's developer portal.

Other Parameters

client_secret (str | None): An optional application client secret, This is only needed if you're fetching OAuth2 tokens with this client.
client_id (int | None): An optional application client id, This is only needed if you're fetching OAuth2 tokens with this client.
settings (aiobungie.builders.Settings | None): The client settings to use, if None the default will be used.
max_retries (int): The max retries number to retry if the request hit a 5xx status code.
debug (bool | str): Whether to enable logging responses or not.

Logging Levels

False: This will disable logging.
True: This will set the level to DEBUG and enable logging minimal information. Like the response status, route, taken time and so on.
"TRACE" | aiobungie.TRACE: This will log the response headers along with the minimal information.

RESTPool( token: 'str', /, *, client_secret: 'str | None' = None, client_id: 'int | None' = None, settings: 'builders.Settings | None' = None, dumps: 'typedefs.Dumps' = <function dumps>, loads: 'typedefs.Loads' = <function loads>, max_retries: 'int' = 4, debug: "typing.Literal['TRACE'] | bool | int" = False) View Source

271    def __init__(
272        self,
273        token: str,
274        /,
275        *,
276        client_secret: str | None = None,
277        client_id: int | None = None,
278        settings: builders.Settings | None = None,
279        dumps: typedefs.Dumps = helpers.dumps,
280        loads: typedefs.Loads = helpers.loads,
281        max_retries: int = 4,
282        debug: typing.Literal["TRACE"] | bool | int = False,
283    ) -> None:
284        self._client_secret = client_secret
285        self._client_id = client_id
286        self._token = token
287        self._max_retries = max_retries
288        self._metadata: collections.MutableMapping[typing.Any, typing.Any] = {}
289        self._enable_debug = debug
290        self._client_session: aiohttp.ClientSession | None = None
291        self._loads = loads
292        self._dumps = dumps
293        self._settings = settings or builders.Settings()

client_id: 'int | None' View Source

295    @property
296    def client_id(self) -> int | None:
297        """Return the client id of this REST client if provided, Otherwise None."""
298        return self._client_id

Return the client id of this REST client if provided, Otherwise None.

metadata: 'collections.MutableMapping[typing.Any, typing.Any]' View Source

300    @property
301    def metadata(self) -> collections.MutableMapping[typing.Any, typing.Any]:
302        """A general-purpose mutable mapping you can use to store data.
303
304        This mapping can be accessed from any process that has a reference to this pool.
305        """
306        return self._metadata

A general-purpose mutable mapping you can use to store data.

This mapping can be accessed from any process that has a reference to this pool.

settings: 'builders.Settings' View Source

308    @property
309    def settings(self) -> builders.Settings:
310        """Internal client settings used within the HTTP client session."""
311        return self._settings

Internal client settings used within the HTTP client session.

@typing.final

def build_oauth2_url(self, client_id: 'int | None' = None) -> 'builders.OAuthURL | None': View Source

319    @typing.final
320    def build_oauth2_url(
321        self, client_id: int | None = None
322    ) -> builders.OAuthURL | None:
323        """Construct a new `OAuthURL` url object.
324
325        You can get the complete string representation of the url by calling `.compile()` on it.
326
327        Parameters
328        ----------
329        client_id : `int | None`
330            An optional client id to provide, If left `None` it will roll back to the id passed
331            to the `RESTClient`, If both is `None` this method will return `None`.
332
333        Returns
334        -------
335        `aiobungie.builders.OAuthURL | None`
336            * If `client_id` was provided as a parameter, It guarantees to return a complete `OAuthURL` object
337            * If `client_id` is set to `aiobungie.RESTClient` will be.
338            * If both are `None` this method will return `None.
339        """
340        client_id = client_id or self._client_id
341        if client_id is None:
342            return None
343
344        return builders.OAuthURL(client_id=client_id)

Construct a new OAuthURL url object.

You can get the complete string representation of the url by calling .compile() on it.

Parameters

client_id (int | None): An optional client id to provide, If left None it will roll back to the id passed to the RESTClient, If both is None this method will return None.

Returns

aiobungie.builders.OAuthURL | None: * If client_id was provided as a parameter, It guarantees to return a complete OAuthURL object
- If client_id is set to aiobungie.RESTClient will be.
- If both are None this method will return `None.

async def start(self) -> 'None': View Source

346    async def start(self) -> None:
347        """Start the TCP connection of this client pool.
348
349        This will raise `RuntimeError` if the connection has already been started.
350
351        Example
352        -------
353        ```py
354        pool = aiobungie.RESTPool(...)
355
356        async def run() -> None:
357            await pool.start()
358            async with pool.acquire() as client:
359                # use client
360
361        async def stop(self) -> None:
362            await pool.close()
363        ```
364        """
365        if self._client_session is not None:
366            raise RuntimeError("<RESTPool> has already been started.") from None
367
368        self._client_session = aiohttp.ClientSession(
369            connector=aiohttp.TCPConnector(
370                use_dns_cache=self._settings.use_dns_cache,
371                ttl_dns_cache=self._settings.ttl_dns_cache,
372                ssl_context=self._settings.ssl_context,
373                ssl=self._settings.ssl,
374            ),
375            connector_owner=True,
376            raise_for_status=False,
377            timeout=self._settings.http_timeout,
378            trust_env=self._settings.trust_env,
379            headers=self._settings.headers,
380        )

Start the TCP connection of this client pool.

This will raise RuntimeError if the connection has already been started.

Example

pool = aiobungie.RESTPool(...)

async def run() -> None:
    await pool.start()
    async with pool.acquire() as client:
        # use client

async def stop(self) -> None:
    await pool.close()

async def stop(self) -> 'None': View Source

382    async def stop(self) -> None:
383        """Stop the TCP connection of this client pool.
384
385        This will raise `RuntimeError` if the connection has already been closed.
386
387        Example
388        -------
389        ```py
390        pool = aiobungie.RESTPool(...)
391
392        async def run() -> None:
393            await pool.start()
394            async with pool.acquire() as client:
395                # use client
396
397        async def stop(self) -> None:
398            await pool.close()
399        ```
400        """
401        if self._client_session is None:
402            raise RuntimeError("<RESTPool> is already stopped.")
403
404        await self._client_session.close()
405        self._client_session = None

Stop the TCP connection of this client pool.

This will raise RuntimeError if the connection has already been closed.

Example

pool = aiobungie.RESTPool(...)

async def run() -> None:
    await pool.start()
    async with pool.acquire() as client:
        # use client

async def stop(self) -> None:
    await pool.close()

@typing.final

def acquire(self) -> 'RESTClient': View Source

407    @typing.final
408    def acquire(self) -> RESTClient:
409        """Acquires a new `RESTClient` instance from this pool.
410
411        Returns
412        -------
413        `RESTClient`
414            An instance of a `RESTClient`.
415        """
416        return RESTClient(
417            self._token,
418            client_secret=self._client_secret,
419            client_id=self._client_id,
420            loads=self._loads,
421            dumps=self._dumps,
422            max_retries=self._max_retries,
423            debug=self._enable_debug,
424            client_session=self._client_session,
425            owned_client=False,
426            settings=self._settings,
427        )

Acquires a new RESTClient instance from this pool.

Returns

RESTClient: An instance of a RESTClient.

TRACE: 'typing.Final[int]' = 5

The trace logging level for the RESTClient responses.

You can enable this with the following code

>>> import logging
>>> logging.getLogger("aiobungie.rest").setLevel(aiobungie.TRACE)
<h1 id="or">or</h1>

>>> logging.basicConfig(level=aiobungie.TRACE)
<h1 id="or-2">Or</h1>

>>> client = aiobungie.RESTClient(debug="TRACE")
<h1 id="or-if-youre-using-aiobungieclient">Or if you're using <code>aiobungie.Client</code></h1>

>>> client = aiobungie.Client()
>>> client.rest.with_debug(level=aiobungie.TRACE, file="rest_logs.txt")