Back to top

hikari.internal.data_binding

Data binding utilities.

View Source
# -*- coding: utf-8 -*-
# cython: language_level=3
# Copyright (c) 2020 Nekokatt
# Copyright (c) 2021-present davfsa
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Data binding utilities."""
from __future__ import annotations

__all__: typing.Sequence[str] = (
    "Headers",
    "Query",
    "JSONObject",
    "JSONArray",
    "JSONish",
    "dump_json",
    "load_json",
    "JSONDecodeError",
    "JSONObjectBuilder",
    "StringMapBuilder",
    "URLEncodedFormBuilder",
)

import typing

import aiohttp
import multidict

from hikari import files
from hikari import snowflakes
from hikari import undefined

if typing.TYPE_CHECKING:
    import concurrent.futures
    import contextlib

    T_co = typing.TypeVar("T_co", covariant=True)

Headers = typing.Mapping[str, str]
"""Type hint for HTTP headers."""

Query = typing.Union[typing.Dict[str, str], multidict.MultiDict[str]]
"""Type hint for HTTP query string."""

# MyPy does not support recursive types yet. This has been ongoing for a long time, unfortunately.
# See https://github.com/python/typing/issues/182

JSONObject = typing.Dict[str, typing.Any]
"""Type hint for a JSON-decoded object representation as a mapping."""

JSONArray = typing.List[typing.Any]
"""Type hint for a JSON-decoded array representation as a sequence."""

JSONish = typing.Union[str, int, float, bool, None, JSONArray, JSONObject]
"""Type hint for any valid JSON-decoded type."""

Stringish = typing.Union[str, int, bool, undefined.UndefinedType, None, snowflakes.Unique]
"""Type hint for any valid that can be put in a StringMapBuilder"""

_StringMapBuilderArg = typing.Union[
    typing.Mapping[str, str],
    typing.Dict[str, str],
    multidict.MultiMapping[str],
    typing.Iterable[typing.Tuple[str, str]],
]

_APPLICATION_OCTET_STREAM: typing.Final[str] = "application/octet-stream"

if typing.TYPE_CHECKING:
    JSONDecodeError: typing.Type[Exception] = Exception
    """Exception raised when loading an invalid JSON string"""

    def dump_json(_: typing.Union[JSONArray, JSONObject], /, *, indent: int = ...) -> str:
        """Convert a Python type to a JSON string."""
        raise NotImplementedError

    def load_json(_: typing.AnyStr, /) -> typing.Union[JSONArray, JSONObject]:
        """Convert a JSON string to a Python type."""
        raise NotImplementedError

else:
    import json

    dump_json = json.dumps
    """Convert a Python type to a JSON string."""

    load_json = json.loads
    """Convert a JSON string to a Python type."""

    JSONDecodeError = json.JSONDecodeError
    """Exception raised when loading an invalid JSON string"""


@typing.final
class URLEncodedFormBuilder:
    """Helper class to generate `aiohttp.FormData`."""

    __slots__: typing.Sequence[str] = ("_executor", "_fields", "_resources")

    def __init__(self, executor: typing.Optional[concurrent.futures.Executor] = None) -> None:
        self._executor = executor
        self._fields: typing.List[typing.Tuple[str, str, typing.Optional[str]]] = []
        self._resources: typing.List[typing.Tuple[str, files.Resource[files.AsyncReader]]] = []

    def add_field(self, name: str, data: str, *, content_type: typing.Optional[str] = None) -> None:
        self._fields.append((name, data, content_type))

    def add_resource(self, name: str, resource: files.Resource[files.AsyncReader]) -> None:
        self._resources.append((name, resource))

    async def build(self, stack: contextlib.AsyncExitStack) -> aiohttp.FormData:
        form = aiohttp.FormData()

        for field in self._fields:
            form.add_field(field[0], field[1], content_type=field[2])

        for name, resource in self._resources:
            stream = await stack.enter_async_context(resource.stream(executor=self._executor))
            mimetype = stream.mimetype or _APPLICATION_OCTET_STREAM
            form.add_field(name, stream, filename=stream.filename, content_type=mimetype)

        return form


@typing.final
class StringMapBuilder(multidict.MultiDict[str]):
    """Helper class used to quickly build query strings or header maps.

    This will consume any items that are not `hikari.undefined.UNDEFINED`.
    If a value _is_ unspecified, it will be ignored when inserting it. This reduces
    the amount of boilerplate needed for generating the headers and query strings for
    low-level HTTP API interaction, amongst other things.

    .. warning::
        Because this subclasses `dict`, you should not use the
        index operator to set items on this object. Doing so will skip any
        form of validation on the type. Use the `put*` methods instead.
    """

    __slots__: typing.Sequence[str] = ()

    def __init__(self, arg: _StringMapBuilderArg = (), **kwargs: str) -> None:
        # We have to allow arguments to be passed to the init here otherwise the inherited copy behaviour from
        # multidict.MultiDict fails.
        super().__init__(arg, **kwargs)

    @typing.overload
    def put(
        self,
        key: str,
        value: Stringish,
        /,
    ) -> None:
        ...

    @typing.overload
    def put(
        self,
        key: str,
        value: undefined.UndefinedOr[T_co],
        /,
        *,
        conversion: typing.Callable[[T_co], Stringish],
    ) -> None:
        ...

    def put(
        self,
        key: str,
        value: undefined.UndefinedOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], typing.Any]] = None,
    ) -> None:
        """Add a key and value to the string map.

        .. note::
            The value will always be cast to a `str` before inserting it.
            `True` will be translated to `"true"`, `False` will be
            translated to `"false"`, and `None` will be translated to
            `"null"`.

        Parameters
        ----------
        key : str
            The string key.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The value to set.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], typing.Any]]
            An optional conversion to perform.
        """
        if value is not undefined.UNDEFINED:
            if conversion is not None:
                value = conversion(value)

            if value is True:
                value = "true"
            elif value is False:
                value = "false"
            elif value is None:
                value = "null"
            elif isinstance(value, snowflakes.Unique):
                value = str(value.id)
            else:
                value = str(value)

            # __setitem__ just overwrites the previous value.
            self.add(key, value)


@typing.final
class JSONObjectBuilder(typing.Dict[str, JSONish]):
    """Helper class used to quickly build JSON objects from various values.

    If provided with any values that are `hikari.undefined.UNDEFINED`,
    then these values will be ignored.

    This speeds up generation of JSON payloads for low level HTTP and websocket
    API interaction.

    .. warning::
        Because this subclasses `dict`, you should not use the
        index operator to set items on this object. Doing so will skip any
        form of validation on the type. Use the `put*` methods instead.
    """

    __slots__: typing.Sequence[str] = ()

    def __init__(self) -> None:
        # Only allow use of empty constructor here.
        super().__init__()

    @typing.overload
    def put(self, key: str, value: undefined.UndefinedNoneOr[JSONish], /) -> None:
        ...

    @typing.overload
    def put(
        self,
        key: str,
        value: undefined.UndefinedNoneOr[T_co],
        /,
        *,
        conversion: typing.Callable[[T_co], JSONish],
    ) -> None:
        ...

    def put(
        self,
        key: str,
        value: undefined.UndefinedNoneOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON value.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The JSON type to put. This may be a non-JSON type if a conversion
            is also specified. This may alternatively be undefined. In the latter
            case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONish]]
            The optional conversion to apply.
        """
        if value is undefined.UNDEFINED:
            return

        if conversion is None or value is None:
            self[key] = value
        else:
            self[key] = conversion(value)

    @typing.overload
    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[JSONish]],
        /,
    ) -> None:
        ...

    @typing.overload
    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[T_co]],
        /,
        *,
        conversion: typing.Callable[[T_co], JSONish],
    ) -> None:
        ...

    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[typing.Any]],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON array.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        If provided, a conversion will be applied to each item.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[T_co]]
            The JSON types to put. This may be an iterable of non-JSON types if
            a conversion is also specified. This may alternatively be undefined.
            In the latter case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONType]]
            The optional conversion to apply.
        """
        if values is not undefined.UNDEFINED:
            if conversion is not None:
                self[key] = [conversion(value) for value in values]
            else:
                self[key] = list(values)

    def put_snowflake(
        self, key: str, value: undefined.UndefinedNoneOr[snowflakes.SnowflakeishOr[snowflakes.Unique]], /
    ) -> None:
        """Put a key with a snowflake value into the builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedNoneOr[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]
            The JSON type to put. This may alternatively be undefined, in this
            case, nothing is performed. This may also be `None`, in this
            case the value isn't cast.
        """
        if value is not undefined.UNDEFINED and value is not None:
            self[key] = str(int(value))
        elif value is None:
            self[key] = value

    def put_snowflake_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[snowflakes.SnowflakeishOr[snowflakes.Unique]]],
        /,
    ) -> None:
        """Put an array of snowflakes with the given key into this builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Each snowflake should be castable to an `int`.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]]
            The JSON snowflakes to put. This may alternatively be undefined.
            In the latter case, nothing is performed.
        """  # noqa: E501 - Line too long
        if values is not undefined.UNDEFINED:
            self[key] = [str(int(value)) for value in values]
#  Headers

Type hint for HTTP headers.

#  JSONArray

Type hint for a JSON-decoded array representation as a sequence.

#  class JSONDecodeError(builtins.ValueError):
View Source
class JSONDecodeError(ValueError):
    """Subclass of ValueError with the following additional properties:

    msg: The unformatted error message
    doc: The JSON document being parsed
    pos: The start index of doc where parsing failed
    lineno: The line corresponding to pos
    colno: The column corresponding to pos

    """
    # Note that this exception is used from _json
    def __init__(self, msg, doc, pos):
        lineno = doc.count('\n', 0, pos) + 1
        colno = pos - doc.rfind('\n', 0, pos)
        errmsg = '%s: line %d column %d (char %d)' % (msg, lineno, colno, pos)
        ValueError.__init__(self, errmsg)
        self.msg = msg
        self.doc = doc
        self.pos = pos
        self.lineno = lineno
        self.colno = colno

    def __reduce__(self):
        return self.__class__, (self.msg, self.doc, self.pos)

Subclass of ValueError with the following additional properties:

msg: The unformatted error message doc: The JSON document being parsed pos: The start index of doc where parsing failed lineno: The line corresponding to pos colno: The column corresponding to pos

Variables and properties
#  args
Methods
#  def __init__(self, msg, doc, pos):
View Source
    def __init__(self, msg, doc, pos):
        lineno = doc.count('\n', 0, pos) + 1
        colno = pos - doc.rfind('\n', 0, pos)
        errmsg = '%s: line %d column %d (char %d)' % (msg, lineno, colno, pos)
        ValueError.__init__(self, errmsg)
        self.msg = msg
        self.doc = doc
        self.pos = pos
        self.lineno = lineno
        self.colno = colno
#  def with_traceback(unknown):

Exception.with_traceback(tb) -- set self.__traceback__ to tb and return self.

#  JSONObject

Type hint for a JSON-decoded object representation as a mapping.

#  
@typing.final
class JSONObjectBuilder(typing.Dict[str, typing.Union[str, int, float, bool, NoneType, typing.List[typing.Any], typing.Dict[str, typing.Any]]]):
View Source
@typing.final
class JSONObjectBuilder(typing.Dict[str, JSONish]):
    """Helper class used to quickly build JSON objects from various values.

    If provided with any values that are `hikari.undefined.UNDEFINED`,
    then these values will be ignored.

    This speeds up generation of JSON payloads for low level HTTP and websocket
    API interaction.

    .. warning::
        Because this subclasses `dict`, you should not use the
        index operator to set items on this object. Doing so will skip any
        form of validation on the type. Use the `put*` methods instead.
    """

    __slots__: typing.Sequence[str] = ()

    def __init__(self) -> None:
        # Only allow use of empty constructor here.
        super().__init__()

    @typing.overload
    def put(self, key: str, value: undefined.UndefinedNoneOr[JSONish], /) -> None:
        ...

    @typing.overload
    def put(
        self,
        key: str,
        value: undefined.UndefinedNoneOr[T_co],
        /,
        *,
        conversion: typing.Callable[[T_co], JSONish],
    ) -> None:
        ...

    def put(
        self,
        key: str,
        value: undefined.UndefinedNoneOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON value.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The JSON type to put. This may be a non-JSON type if a conversion
            is also specified. This may alternatively be undefined. In the latter
            case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONish]]
            The optional conversion to apply.
        """
        if value is undefined.UNDEFINED:
            return

        if conversion is None or value is None:
            self[key] = value
        else:
            self[key] = conversion(value)

    @typing.overload
    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[JSONish]],
        /,
    ) -> None:
        ...

    @typing.overload
    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[T_co]],
        /,
        *,
        conversion: typing.Callable[[T_co], JSONish],
    ) -> None:
        ...

    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[typing.Any]],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON array.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        If provided, a conversion will be applied to each item.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[T_co]]
            The JSON types to put. This may be an iterable of non-JSON types if
            a conversion is also specified. This may alternatively be undefined.
            In the latter case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONType]]
            The optional conversion to apply.
        """
        if values is not undefined.UNDEFINED:
            if conversion is not None:
                self[key] = [conversion(value) for value in values]
            else:
                self[key] = list(values)

    def put_snowflake(
        self, key: str, value: undefined.UndefinedNoneOr[snowflakes.SnowflakeishOr[snowflakes.Unique]], /
    ) -> None:
        """Put a key with a snowflake value into the builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedNoneOr[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]
            The JSON type to put. This may alternatively be undefined, in this
            case, nothing is performed. This may also be `None`, in this
            case the value isn't cast.
        """
        if value is not undefined.UNDEFINED and value is not None:
            self[key] = str(int(value))
        elif value is None:
            self[key] = value

    def put_snowflake_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[snowflakes.SnowflakeishOr[snowflakes.Unique]]],
        /,
    ) -> None:
        """Put an array of snowflakes with the given key into this builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Each snowflake should be castable to an `int`.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]]
            The JSON snowflakes to put. This may alternatively be undefined.
            In the latter case, nothing is performed.
        """  # noqa: E501 - Line too long
        if values is not undefined.UNDEFINED:
            self[key] = [str(int(value)) for value in values]

Helper class used to quickly build JSON objects from various values.

If provided with any values that are hikari.undefined.UNDEFINED, then these values will be ignored.

This speeds up generation of JSON payloads for low level HTTP and websocket API interaction.

Warning: Because this subclasses dict, you should not use the index operator to set items on this object. Doing so will skip any form of validation on the type. Use the put* methods instead.

Methods
#  def __init__(self):
View Source
    def __init__(self) -> None:
        # Only allow use of empty constructor here.
        super().__init__()
#  def clear(unknown):

D.clear() -> None. Remove all items from D.

#  def copy(unknown):

D.copy() -> a shallow copy of D

#  def fromkeys(type, iterable, value=None, /):

Create a new dictionary with keys from iterable and values set to value.

#  def get(self, key, default=None, /):

Return the value for key if key is in the dictionary, else default.

#  def items(unknown):

D.items() -> a set-like object providing a view on D's items

#  def keys(unknown):

D.keys() -> a set-like object providing a view on D's keys

#  def pop(unknown):

D.pop(k[,d]) -> v, remove specified key and return the corresponding value.

If key is not found, default is returned if given, otherwise KeyError is raised

#  def popitem(self, /):

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

#  def put(
   self,
   key: str,
   value: Union[Any, hikari.undefined.UndefinedType, NoneType],
   /,
   *,
   conversion: Optional[Callable[[Any], Union[str, int, float, bool, NoneType, List[Any], Dict[str, Any]]]] = None
) -> None:
View Source
    def put(
        self,
        key: str,
        value: undefined.UndefinedNoneOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON value.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The JSON type to put. This may be a non-JSON type if a conversion
            is also specified. This may alternatively be undefined. In the latter
            case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONish]]
            The optional conversion to apply.
        """
        if value is undefined.UNDEFINED:
            return

        if conversion is None or value is None:
            self[key] = value
        else:
            self[key] = conversion(value)

Put a JSON value.

If the value is hikari.undefined.UNDEFINED it will not be stored.

Parameters
  • key (str): The key to give the element.
  • value (hikari.undefined.UndefinedOr[typing.Any]): The JSON type to put. This may be a non-JSON type if a conversion is also specified. This may alternatively be undefined. In the latter case, nothing is performed.
Other Parameters
  • conversion (typing.Optional[typing.Callable[[typing.Any], JSONish]]): The optional conversion to apply.
#  def put_array(
   self,
   key: str,
   values: Union[Iterable[Any], hikari.undefined.UndefinedType],
   /,
   *,
   conversion: Optional[Callable[[Any], Union[str, int, float, bool, NoneType, List[Any], Dict[str, Any]]]] = None
) -> None:
View Source
    def put_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[typing.Any]],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], JSONish]] = None,
    ) -> None:
        """Put a JSON array.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        If provided, a conversion will be applied to each item.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[T_co]]
            The JSON types to put. This may be an iterable of non-JSON types if
            a conversion is also specified. This may alternatively be undefined.
            In the latter case, nothing is performed.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], JSONType]]
            The optional conversion to apply.
        """
        if values is not undefined.UNDEFINED:
            if conversion is not None:
                self[key] = [conversion(value) for value in values]
            else:
                self[key] = list(values)

Put a JSON array.

If the value is hikari.undefined.UNDEFINED it will not be stored.

If provided, a conversion will be applied to each item.

Parameters
  • key (str): The key to give the element.
  • values (hikari.undefined.UndefinedOr[typing.Iterable[T_co]]): The JSON types to put. This may be an iterable of non-JSON types if a conversion is also specified. This may alternatively be undefined. In the latter case, nothing is performed.
Other Parameters
  • conversion (typing.Optional[typing.Callable[[typing.Any], JSONType]]): The optional conversion to apply.
#  def put_snowflake(
   self,
   key: str,
   value: Union[hikari.snowflakes.Unique, hikari.snowflakes.Snowflake, int, hikari.undefined.UndefinedType, NoneType],
   /
) -> None:
View Source
    def put_snowflake(
        self, key: str, value: undefined.UndefinedNoneOr[snowflakes.SnowflakeishOr[snowflakes.Unique]], /
    ) -> None:
        """Put a key with a snowflake value into the builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Parameters
        ----------
        key : str
            The key to give the element.
        value : hikari.undefined.UndefinedNoneOr[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]
            The JSON type to put. This may alternatively be undefined, in this
            case, nothing is performed. This may also be `None`, in this
            case the value isn't cast.
        """
        if value is not undefined.UNDEFINED and value is not None:
            self[key] = str(int(value))
        elif value is None:
            self[key] = value

Put a key with a snowflake value into the builder.

If the value is hikari.undefined.UNDEFINED it will not be stored.

Parameters
#  def put_snowflake_array(
   self,
   key: str,
   values: Union[Iterable[Union[hikari.snowflakes.Unique, hikari.snowflakes.Snowflake, int]], hikari.undefined.UndefinedType],
   /
) -> None:
View Source
    def put_snowflake_array(
        self,
        key: str,
        values: undefined.UndefinedOr[typing.Iterable[snowflakes.SnowflakeishOr[snowflakes.Unique]]],
        /,
    ) -> None:
        """Put an array of snowflakes with the given key into this builder.

        If the value is `hikari.undefined.UNDEFINED` it will not be stored.

        Each snowflake should be castable to an `int`.

        Parameters
        ----------
        key : str
            The key to give the element.
        values : hikari.undefined.UndefinedOr[typing.Iterable[hikari.snowflakes.SnowflakeishOr[hikari.snowflakes.Unique]]]
            The JSON snowflakes to put. This may alternatively be undefined.
            In the latter case, nothing is performed.
        """  # noqa: E501 - Line too long
        if values is not undefined.UNDEFINED:
            self[key] = [str(int(value)) for value in values]

Put an array of snowflakes with the given key into this builder.

If the value is hikari.undefined.UNDEFINED it will not be stored.

Each snowflake should be castable to an int.

Parameters
#  def setdefault(self, key, default=None, /):

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

#  def update(unknown):

D.update([E, ]**F) -> None. Update D from dict/iterable E and F. If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

#  def values(unknown):

D.values() -> an object providing a view on D's values

#  JSONish

Type hint for any valid JSON-decoded type.

#  Query

Type hint for HTTP query string.

#  
@typing.final
class StringMapBuilder(multidict._multidict.MultiDict[str]):
View Source
@typing.final
class StringMapBuilder(multidict.MultiDict[str]):
    """Helper class used to quickly build query strings or header maps.

    This will consume any items that are not `hikari.undefined.UNDEFINED`.
    If a value _is_ unspecified, it will be ignored when inserting it. This reduces
    the amount of boilerplate needed for generating the headers and query strings for
    low-level HTTP API interaction, amongst other things.

    .. warning::
        Because this subclasses `dict`, you should not use the
        index operator to set items on this object. Doing so will skip any
        form of validation on the type. Use the `put*` methods instead.
    """

    __slots__: typing.Sequence[str] = ()

    def __init__(self, arg: _StringMapBuilderArg = (), **kwargs: str) -> None:
        # We have to allow arguments to be passed to the init here otherwise the inherited copy behaviour from
        # multidict.MultiDict fails.
        super().__init__(arg, **kwargs)

    @typing.overload
    def put(
        self,
        key: str,
        value: Stringish,
        /,
    ) -> None:
        ...

    @typing.overload
    def put(
        self,
        key: str,
        value: undefined.UndefinedOr[T_co],
        /,
        *,
        conversion: typing.Callable[[T_co], Stringish],
    ) -> None:
        ...

    def put(
        self,
        key: str,
        value: undefined.UndefinedOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], typing.Any]] = None,
    ) -> None:
        """Add a key and value to the string map.

        .. note::
            The value will always be cast to a `str` before inserting it.
            `True` will be translated to `"true"`, `False` will be
            translated to `"false"`, and `None` will be translated to
            `"null"`.

        Parameters
        ----------
        key : str
            The string key.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The value to set.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], typing.Any]]
            An optional conversion to perform.
        """
        if value is not undefined.UNDEFINED:
            if conversion is not None:
                value = conversion(value)

            if value is True:
                value = "true"
            elif value is False:
                value = "false"
            elif value is None:
                value = "null"
            elif isinstance(value, snowflakes.Unique):
                value = str(value.id)
            else:
                value = str(value)

            # __setitem__ just overwrites the previous value.
            self.add(key, value)

Helper class used to quickly build query strings or header maps.

This will consume any items that are not hikari.undefined.UNDEFINED. If a value _is_ unspecified, it will be ignored when inserting it. This reduces the amount of boilerplate needed for generating the headers and query strings for low-level HTTP API interaction, amongst other things.

Warning: Because this subclasses dict, you should not use the index operator to set items on this object. Doing so will skip any form of validation on the type. Use the put* methods instead.

Methods
#  def __init__(
   self,
   arg: Union[Mapping[str, str], Dict[str, str], multidict._abc.MultiMapping[str], Iterable[Tuple[str, str]]] = (),
   **kwargs: str
):
View Source
    def __init__(self, arg: _StringMapBuilderArg = (), **kwargs: str) -> None:
        # We have to allow arguments to be passed to the init here otherwise the inherited copy behaviour from
        # multidict.MultiDict fails.
        super().__init__(arg, **kwargs)
#  def add(unknown):

Add the key and value, not overwriting any previous value.

#  def clear(unknown):

Remove all items from MultiDict

#  def copy(unknown):

Return a copy of itself.

#  def extend(unknown):

Extend current MultiDict with more values. This method must be used instead of update.

#  def get(unknown):

Get first value matching the key.

The method is alias for .getone().

#  def getall(unknown):

Return a list of all values matching the key.

#  def getone(unknown):

Get first value matching the key.

#  def items(unknown):

Return a new view of the dictionary's items *(key, value) pairs).

#  def keys(unknown):

Return a new view of the dictionary's keys.

#  def pop(unknown):

Remove the last occurrence of key and return the corresponding value.

If key is not found, default is returned if given, otherwise KeyError is raised.

#  def popall(unknown):

Remove all occurrences of key and return the list of corresponding values.

If key is not found, default is returned if given, otherwise KeyError is raised.

#  def popitem(unknown):

Remove and return an arbitrary (key, value) pair.

#  def popone(unknown):

Remove the last occurrence of key and return the corresponding value.

If key is not found, default is returned if given, otherwise KeyError is raised.

#  def put(
   self,
   key: str,
   value: Union[Any, hikari.undefined.UndefinedType],
   /,
   *,
   conversion: Optional[Callable[[Any], Any]] = None
) -> None:
View Source
    def put(
        self,
        key: str,
        value: undefined.UndefinedOr[typing.Any],
        /,
        *,
        conversion: typing.Optional[typing.Callable[[typing.Any], typing.Any]] = None,
    ) -> None:
        """Add a key and value to the string map.

        .. note::
            The value will always be cast to a `str` before inserting it.
            `True` will be translated to `"true"`, `False` will be
            translated to `"false"`, and `None` will be translated to
            `"null"`.

        Parameters
        ----------
        key : str
            The string key.
        value : hikari.undefined.UndefinedOr[typing.Any]
            The value to set.

        Other Parameters
        ----------------
        conversion : typing.Optional[typing.Callable[[typing.Any], typing.Any]]
            An optional conversion to perform.
        """
        if value is not undefined.UNDEFINED:
            if conversion is not None:
                value = conversion(value)

            if value is True:
                value = "true"
            elif value is False:
                value = "false"
            elif value is None:
                value = "null"
            elif isinstance(value, snowflakes.Unique):
                value = str(value.id)
            else:
                value = str(value)

            # __setitem__ just overwrites the previous value.
            self.add(key, value)

Add a key and value to the string map.

Note: The value will always be cast to a str before inserting it. True will be translated to "true", False will be translated to "false", and None will be translated to "null".

Parameters
Other Parameters
  • conversion (typing.Optional[typing.Callable[[typing.Any], typing.Any]]): An optional conversion to perform.
#  def setdefault(unknown):

Return value for key, set value to default if key is not present.

#  def update(unknown):

Update the dictionary from other, overwriting existing keys.

#  def values(unknown):

Return a new view of the dictionary's values.

#  
@typing.final
class URLEncodedFormBuilder:
View Source
@typing.final
class URLEncodedFormBuilder:
    """Helper class to generate `aiohttp.FormData`."""

    __slots__: typing.Sequence[str] = ("_executor", "_fields", "_resources")

    def __init__(self, executor: typing.Optional[concurrent.futures.Executor] = None) -> None:
        self._executor = executor
        self._fields: typing.List[typing.Tuple[str, str, typing.Optional[str]]] = []
        self._resources: typing.List[typing.Tuple[str, files.Resource[files.AsyncReader]]] = []

    def add_field(self, name: str, data: str, *, content_type: typing.Optional[str] = None) -> None:
        self._fields.append((name, data, content_type))

    def add_resource(self, name: str, resource: files.Resource[files.AsyncReader]) -> None:
        self._resources.append((name, resource))

    async def build(self, stack: contextlib.AsyncExitStack) -> aiohttp.FormData:
        form = aiohttp.FormData()

        for field in self._fields:
            form.add_field(field[0], field[1], content_type=field[2])

        for name, resource in self._resources:
            stream = await stack.enter_async_context(resource.stream(executor=self._executor))
            mimetype = stream.mimetype or _APPLICATION_OCTET_STREAM
            form.add_field(name, stream, filename=stream.filename, content_type=mimetype)

        return form

Helper class to generate aiohttp.FormData.

Methods
#  def __init__(self, executor: Optional[concurrent.futures._base.Executor] = None):
View Source
    def __init__(self, executor: typing.Optional[concurrent.futures.Executor] = None) -> None:
        self._executor = executor
        self._fields: typing.List[typing.Tuple[str, str, typing.Optional[str]]] = []
        self._resources: typing.List[typing.Tuple[str, files.Resource[files.AsyncReader]]] = []
#  def add_field(
   self,
   name: str,
   data: str,
   *,
   content_type: Optional[str] = None
) -> None:
View Source
    def add_field(self, name: str, data: str, *, content_type: typing.Optional[str] = None) -> None:
        self._fields.append((name, data, content_type))
#  def add_resource(
   self,
   name: str,
   resource: hikari.files.Resource[hikari.files.AsyncReader]
) -> None:
View Source
    def add_resource(self, name: str, resource: files.Resource[files.AsyncReader]) -> None:
        self._resources.append((name, resource))
#  async def build(self, stack: contextlib.AsyncExitStack) -> aiohttp.formdata.FormData:
View Source
    async def build(self, stack: contextlib.AsyncExitStack) -> aiohttp.FormData:
        form = aiohttp.FormData()

        for field in self._fields:
            form.add_field(field[0], field[1], content_type=field[2])

        for name, resource in self._resources:
            stream = await stack.enter_async_context(resource.stream(executor=self._executor))
            mimetype = stream.mimetype or _APPLICATION_OCTET_STREAM
            form.add_field(name, stream, filename=stream.filename, content_type=mimetype)

        return form
#  def dump_json(
   obj,
   *,
   skipkeys=False,
   ensure_ascii=True,
   check_circular=True,
   allow_nan=True,
   cls=None,
   indent=None,
   separators=None,
   default=None,
   sort_keys=False,
   **kw
):
View Source
def dumps(obj, *, skipkeys=False, ensure_ascii=True, check_circular=True,
        allow_nan=True, cls=None, indent=None, separators=None,
        default=None, sort_keys=False, **kw):
    """Serialize ``obj`` to a JSON formatted ``str``.

    If ``skipkeys`` is true then ``dict`` keys that are not basic types
    (``str``, ``int``, ``float``, ``bool``, ``None``) will be skipped
    instead of raising a ``TypeError``.

    If ``ensure_ascii`` is false, then the return value can contain non-ASCII
    characters if they appear in strings contained in ``obj``. Otherwise, all
    such characters are escaped in JSON strings.

    If ``check_circular`` is false, then the circular reference check
    for container types will be skipped and a circular reference will
    result in an ``RecursionError`` (or worse).

    If ``allow_nan`` is false, then it will be a ``ValueError`` to
    serialize out of range ``float`` values (``nan``, ``inf``, ``-inf``) in
    strict compliance of the JSON specification, instead of using the
    JavaScript equivalents (``NaN``, ``Infinity``, ``-Infinity``).

    If ``indent`` is a non-negative integer, then JSON array elements and
    object members will be pretty-printed with that indent level. An indent
    level of 0 will only insert newlines. ``None`` is the most compact
    representation.

    If specified, ``separators`` should be an ``(item_separator, key_separator)``
    tuple.  The default is ``(', ', ': ')`` if *indent* is ``None`` and
    ``(',', ': ')`` otherwise.  To get the most compact JSON representation,
    you should specify ``(',', ':')`` to eliminate whitespace.

    ``default(obj)`` is a function that should return a serializable version
    of obj or raise TypeError. The default simply raises TypeError.

    If *sort_keys* is true (default: ``False``), then the output of
    dictionaries will be sorted by key.

    To use a custom ``JSONEncoder`` subclass (e.g. one that overrides the
    ``.default()`` method to serialize additional types), specify it with
    the ``cls`` kwarg; otherwise ``JSONEncoder`` is used.

    """
    # cached encoder
    if (not skipkeys and ensure_ascii and
        check_circular and allow_nan and
        cls is None and indent is None and separators is None and
        default is None and not sort_keys and not kw):
        return _default_encoder.encode(obj)
    if cls is None:
        cls = JSONEncoder
    return cls(
        skipkeys=skipkeys, ensure_ascii=ensure_ascii,
        check_circular=check_circular, allow_nan=allow_nan, indent=indent,
        separators=separators, default=default, sort_keys=sort_keys,
        **kw).encode(obj)

Serialize obj to a JSON formatted str.

If skipkeys is true then dict keys that are not basic types (str, int, float, bool, None) will be skipped instead of raising a TypeError.

If ensure_ascii is false, then the return value can contain non-ASCII characters if they appear in strings contained in obj. Otherwise, all such characters are escaped in JSON strings.

If check_circular is false, then the circular reference check for container types will be skipped and a circular reference will result in an RecursionError (or worse).

If allow_nan is false, then it will be a ValueError to serialize out of range float values (nan, inf, -inf) in strict compliance of the JSON specification, instead of using the JavaScript equivalents (NaN, Infinity, -Infinity).

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (', ', ': ') if indent is None and (',', ': ') otherwise. To get the most compact JSON representation, you should specify (',', ':') to eliminate whitespace.

default(obj) is a function that should return a serializable version of obj or raise TypeError. The default simply raises TypeError.

If sort_keys is true (default: False), then the output of dictionaries will be sorted by key.

To use a custom JSONEncoder subclass (e.g. one that overrides the .default() method to serialize additional types), specify it with the cls kwarg; otherwise JSONEncoder is used.

#  def load_json(
   s,
   *,
   cls=None,
   object_hook=None,
   parse_float=None,
   parse_int=None,
   parse_constant=None,
   object_pairs_hook=None,
   **kw
):
View Source
def loads(s, *, cls=None, object_hook=None, parse_float=None,
        parse_int=None, parse_constant=None, object_pairs_hook=None, **kw):
    """Deserialize ``s`` (a ``str``, ``bytes`` or ``bytearray`` instance
    containing a JSON document) to a Python object.

    ``object_hook`` is an optional function that will be called with the
    result of any object literal decode (a ``dict``). The return value of
    ``object_hook`` will be used instead of the ``dict``. This feature
    can be used to implement custom decoders (e.g. JSON-RPC class hinting).

    ``object_pairs_hook`` is an optional function that will be called with the
    result of any object literal decoded with an ordered list of pairs.  The
    return value of ``object_pairs_hook`` will be used instead of the ``dict``.
    This feature can be used to implement custom decoders.  If ``object_hook``
    is also defined, the ``object_pairs_hook`` takes priority.

    ``parse_float``, if specified, will be called with the string
    of every JSON float to be decoded. By default this is equivalent to
    float(num_str). This can be used to use another datatype or parser
    for JSON floats (e.g. decimal.Decimal).

    ``parse_int``, if specified, will be called with the string
    of every JSON int to be decoded. By default this is equivalent to
    int(num_str). This can be used to use another datatype or parser
    for JSON integers (e.g. float).

    ``parse_constant``, if specified, will be called with one of the
    following strings: -Infinity, Infinity, NaN.
    This can be used to raise an exception if invalid JSON numbers
    are encountered.

    To use a custom ``JSONDecoder`` subclass, specify it with the ``cls``
    kwarg; otherwise ``JSONDecoder`` is used.
    """
    if isinstance(s, str):
        if s.startswith('\ufeff'):
            raise JSONDecodeError("Unexpected UTF-8 BOM (decode using utf-8-sig)",
                                  s, 0)
    else:
        if not isinstance(s, (bytes, bytearray)):
            raise TypeError(f'the JSON object must be str, bytes or bytearray, '
                            f'not {s.__class__.__name__}')
        s = s.decode(detect_encoding(s), 'surrogatepass')

    if (cls is None and object_hook is None and
            parse_int is None and parse_float is None and
            parse_constant is None and object_pairs_hook is None and not kw):
        return _default_decoder.decode(s)
    if cls is None:
        cls = JSONDecoder
    if object_hook is not None:
        kw['object_hook'] = object_hook
    if object_pairs_hook is not None:
        kw['object_pairs_hook'] = object_pairs_hook
    if parse_float is not None:
        kw['parse_float'] = parse_float
    if parse_int is not None:
        kw['parse_int'] = parse_int
    if parse_constant is not None:
        kw['parse_constant'] = parse_constant
    return cls(**kw).decode(s)

Deserialize s (a str, bytes or bytearray instance containing a JSON document) to a Python object.

object_hook is an optional function that will be called with the result of any object literal decode (a dict). The return value of object_hook will be used instead of the dict. This feature can be used to implement custom decoders (e.g. JSON-RPC class hinting).

object_pairs_hook is an optional function that will be called with the result of any object literal decoded with an ordered list of pairs. The return value of object_pairs_hook will be used instead of the dict. This feature can be used to implement custom decoders. If object_hook is also defined, the object_pairs_hook takes priority.

parse_float, if specified, will be called with the string of every JSON float to be decoded. By default this is equivalent to float(num_str). This can be used to use another datatype or parser for JSON floats (e.g. decimal.Decimal).

parse_int, if specified, will be called with the string of every JSON int to be decoded. By default this is equivalent to int(num_str). This can be used to use another datatype or parser for JSON integers (e.g. float).

parse_constant, if specified, will be called with one of the following strings: -Infinity, Infinity, NaN. This can be used to raise an exception if invalid JSON numbers are encountered.

To use a custom JSONDecoder subclass, specify it with the cls kwarg; otherwise JSONDecoder is used.