HTTP Components

Parts required for the http communication to work

RateLimiter

Gives out x tokens for network operations every y seconds

Attributes:

Name	Type	Description
max_tokens	int	How many requests can we save up - bungie limits after 250 in 10s, so the default is 240
seconds	int	In how many seconds those requests are allowed

Source code in src/bungio/http/ratelimiting.py

@custom_define()
class RateLimiter:
    """
    Gives out x tokens for network operations every y seconds

    Attributes:
        max_tokens: How many requests can we save up - bungie limits after 250 in 10s, so the default is 240
        seconds: In how many seconds those requests are allowed
    """

    max_tokens: int = custom_field(default=240)
    seconds: int = custom_field(default=10)

    tokens: float = custom_field(init=False)
    updated_at: float = custom_field(init=False)

    lock: asyncio.Lock = custom_field(init=False, default=asyncio.Lock())

    def __attrs_post_init__(self):
        self.tokens = self.max_tokens

    async def wait_for_token(self):
        """Waits until a token becomes available"""

        async with self.lock:
            if self.tokens == 0:
                current_time = time.time()
                if (missing := current_time - self.updated_at) < self.seconds:
                    await asyncio.sleep(self.seconds - missing)
                self.tokens = self.max_tokens

            if self.tokens == self.max_tokens:
                # the first request is made, start the timer when that should fill back up
                self.updated_at = time.time()

            self.tokens -= 1

wait_for_token() async

Waits until a token becomes available

Source code in src/bungio/http/ratelimiting.py

async def wait_for_token(self):
    """Waits until a token becomes available"""

    async with self.lock:
        if self.tokens == 0:
            current_time = time.time()
            if (missing := current_time - self.updated_at) < self.seconds:
                await asyncio.sleep(self.seconds - missing)
            self.tokens = self.max_tokens

        if self.tokens == self.max_tokens:
            # the first request is made, start the timer when that should fill back up
            self.updated_at = time.time()

        self.tokens -= 1

Route

Bungie http route

Attributes:

Name	Type	Description
path	str	The bungie path addition
method	str	The http method
auth	Optional[AuthData]	Authentication information
data	Optional[dict | list]	Body data to send
**params	Optional[dict | list]	All query parameters

Source code in src/bungio/http/route.py

@custom_define(init=False)
class Route:
    """
    Bungie http route

    Attributes:
        path: The bungie path addition
        method: The http method
        auth: Authentication information
        data: Body data to send
        **params: All query parameters
    """

    path: str
    method: str
    data: Optional[dict | list]
    auth: Optional[AuthData]
    params: dict

    def __init__(
        self, path: str, method: str, data: Optional[dict | list] = None, auth: Optional[AuthData] = None, **params
    ):
        self.data = data
        self.method = method
        self.path = BASE_ROUTE + path

        # pgcr need a different url
        if "PostGameCarnageReport" in self.path:
            self.path = self.path.replace("www", "stats")

        # manifest too
        if "destiny2_content" in self.path:
            self.path = self.path.replace("/Platform", "")

        # only set auth if a token is provided
        if auth is not None and auth.token is None:
            auth = None
        self.auth = auth

        # clean the data
        if isinstance(data, dict):
            cleaned = {}
            for key, value in self.data.items():
                # skip None / MISSING entries
                if value is None or value is MISSING:
                    continue

                cleaned[key] = value
            self.data = cleaned

        # clean the params
        self.params = {}
        for key, value in params.items():
            # skip None / MISSING params
            if value is None or value is MISSING:
                continue

            # convert bools to string
            if isinstance(value, bool):
                value = str(value)

            # lists need to be comma seperated
            if isinstance(value, list):
                value = ",".join([str(v) for v in value])

            # key needs to be name case again
            bungie_key = ""
            for i, k in enumerate(key.split("_")):
                if i != 0:
                    bungie_key += k.capitalize()
                else:
                    bungie_key += k
            self.params[bungie_key] = value