hikari.files#

Utilities and classes for interacting with files and web resources.

Module Contents#

class hikari.files.AsyncReader[source]#

Bases: AsyncIterable[bytes], abc.ABC

Protocol for reading a resource asynchronously using bit inception.

This supports being used as an async iterable, although the implementation detail is left to each implementation of this class to define.

filename: str[source]#: The filename of the resource.

mimetype: str | None[source]#: The mimetype of the resource. May be None if not known.

async data_uri()[source]#

Fetch the data URI.

This reads the entire resource.

async read()[source]#: Read the rest of the resource and return it in a bytes object.

class hikari.files.AsyncReaderContextManager[source]#

Bases: abc.ABC, Generic[ReaderImplT]

Context manager that returns a reader.

class hikari.files.Bytes(data, filename, /, mimetype=None, *, spoiler=False)[source]#

Bases: Resource[IteratorReader]

Representation of in-memory data to upload.

Parameters:

datatyping.Union[Rawish, LazyByteIteratorish]: The raw data.
filenamestr: The filename to use.
mimetypetyping.Optional[str]: The mimetype, or None if you do not wish to specify this. If not provided, then this will be generated from the file extension of the filename instead.
spoilerbool: Whether to mark the file as a spoiler in Discord. Defaults to False.

property filename: str[source]#: Filename of the resource.

property url: str[source]#: URL of the resource.

data: bytes | LazyByteIteratorish[source]#: The raw data/provider of raw data to upload.

is_spoiler: bool[source]#: Whether the file will be marked as a spoiler.

mimetype: str | None[source]#: The provided mimetype, if provided. Otherwise None.

static from_data_uri(data_uri, filename=None)[source]#

Parse a given data URI.

Parameters:

data_uristr: The data URI to parse.
filenametyping.Optional[str]: Filename to use. If this is not provided, then this is generated instead.

Returns:

Bytes: The parsed data URI as a Bytes object.

Raises:

ValueError: If the parsed argument is not a data URI.

async save(path, *, executor=None, force=False)[source]#

Save this resource to disk.

This writes the resource file in chunks, and so does not load the entire resource into memory.

Parameters:

pathPathish: The path to save this resource to. If this is a string, the path will be relative to the current working directory.
executortyping.Optional[concurrent.futures.Executor]: The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.
forcebool: Whether to overwrite an existing file. Defaults to False.

stream(*, executor=None, head_only=False)[source]#

Start streaming the content in chunks.

Parameters:

executortyping.Optional[concurrent.futures.Executor]: Not used. Provided only to match the underlying interface.
head_onlybool: Not used. Provided only to match the underlying interface.

Returns:

AsyncReaderContextManager[IteratorReader]: An async context manager that when entered, produces the data stream.

class hikari.files.File(path, /, filename=None, *, spoiler=False)[source]#

Bases: Resource[ThreadedFileReader]

A resource that exists on the local machine’s storage to be uploaded.

Parameters:

pathtyping.Union[str, os.PathLike, pathlib.Path]

The path to use.

If passing a pathlib.Path, this must not be a pathlib.PurePath directly, as it will be used to expand tokens such as that denote the home directory, and for relative paths.

This will all be performed as required in an executor to prevent blocking the event loop.

filenametyping.Optional[str]

The filename to use. If this is None, the name of the file is taken from the path instead.

spoilerbool

Whether to mark the file as a spoiler in Discord. Defaults to False.

property filename: str[source]#: Filename of the resource.

property url: str[source]#: URL of the resource.

is_spoiler: bool[source]#: Whether the file will be marked as a spoiler.

path: pathlib.Path[source]#: The path to the file.

async save(path, *, executor=None, force=False)[source]#

Save this resource to disk.

This writes the resource file in chunks, and so does not load the entire resource into memory.

Parameters:

pathPathish: The path to save this resource to. If this is a string, the path will be relative to the current working directory.
executortyping.Optional[concurrent.futures.Executor]: The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.
forcebool: Whether to overwrite an existing file. Defaults to False.

stream(*, executor=None, head_only=False)[source]#

Start streaming the resource using a thread pool executor.

Parameters:

executortyping.Optional[concurrent.futures.Executor]

The thread executor to run the blocking read operations in. If None, the default executor for the running event loop will be used instead.

Only concurrent.futures.TheadPoolExecutor is supported.

head_onlybool

Not used. Provided only to match the underlying interface.

Returns:

AsyncReaderContextManager[ThreadedFileReader]: An async context manager that when entered, produces the data stream.

Raises:

IsADirectoryError: If the file’s path leads to a directory.
FileNotFoundError: If the file doesn’t exist.

class hikari.files.IteratorReader[source]#

Bases: AsyncReader

Asynchronous file reader that operates on in-memory data.

data: bytes | LazyByteIteratorish[source]#: The data that will be yielded in chunks.

class hikari.files.Resource[source]#

Bases: Generic[ReaderImplT], abc.ABC

Base for any uploadable or downloadable representation of information.

These representations can be streamed using bit inception for performance, which may result in significant decrease in memory usage for larger resources.

property extension: str | None[source]#: File extension, if there is one.

abstract property filename: str[source]#: Filename of the resource.

abstract property url: str[source]#: URL of the resource.

async read(*, executor=None)[source]#

Read the entire resource at once into memory.

data = await resource.read(...)
# ^-- This is a shortcut for the following --v
async with resource.stream(...) as reader:
    data = await reader.read()

Warning

If you simply wish to re-upload this resource to Discord via any endpoint in Hikari, you should opt to just pass this resource object directly. This way, Hikari can perform byte inception, which significantly reduces the memory usage for your bot as it grows larger.

Parameters:

executortyping.Optional[concurrent.futures.Executor]: The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

Returns:

bytes: The entire resource.

async save(path, *, executor=None, force=False)[source]#

Save this resource to disk.

This writes the resource file in chunks, and so does not load the entire resource into memory.

Parameters:

pathPathish: The path to save this resource to. If this is a string, the path will be relative to the current working directory.
executortyping.Optional[concurrent.futures.Executor]: The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.
forcebool: Whether to overwrite an existing file. Defaults to False.

abstract stream(*, executor=None, head_only=False)[source]#

Produce a stream of data for the resource.

Parameters:

executortyping.Optional[concurrent.futures.Executor]: The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.
head_onlybool: Defaults to False. If True, then the implementation may only retrieve HEAD information if supported. This currently only has any effect for web requests. This will fetch the headers for the HTTP resource this object points to without downloading the entire content, which can be significantly faster if you are scanning file types in messages, for example.

Returns:

AsyncReaderContextManager[AsyncReader]

An async iterable of bytes to stream.

This will error on enter if the target resource doesn’t exist.

class hikari.files.URL(url, filename=None)[source]#

Bases: WebResource

A URL that represents a web resource.

Note

Some components may choose to not upload this resource directly and instead simply refer to the URL as needed. The main place this will occur is within embeds.

If you need to re-upload the resource, you should download it into a bytes and pass that instead in these cases.

Parameters:

urlstr

The URL of the resource.

filenametyping.Optional[str]

The filename for the resource.

If not specified, it will be obtained from the url.

property filename: str[source]#: Filename of the resource.

property url: str[source]#: URL of the resource.

class hikari.files.WebReader[source]#

Bases: AsyncReader

Asynchronous reader to use to read data from a web resource.

charset: str | None[source]#: Optional character set information, if known.

head_only: bool[source]#

If True, then only the HEAD was requested.

In this case, neither __aiter__ nor read would return anything other than an empty byte string.

reason: str[source]#: The HTTP response status reason.

size: int | None[source]#: The size of the resource, if known.

status: int[source]#: The initial HTTP response status.

stream: aiohttp.StreamReader[source]#: The aiohttp.StreamReader to read the content from.

url: str[source]#: The URL being read from.

async read()[source]#: Read the rest of the resource and return it in a bytes object.

class hikari.files.WebResource[source]#

Bases: Resource[WebReader], abc.ABC

Base class for a resource that resides on the internet.

The logic for identifying this resource is left to each implementation to define.

For a usable concrete implementation, use URL instead.

Some components may choose to not upload this resource directly and instead simply refer to the URL as needed. The main place this will occur is within embeds. If you need to re-upload the resource, you should download it into a bytes and pass that instead in these cases.

stream(*, executor=None, head_only=False)[source]#

Start streaming the content into memory by downloading it.

You can use this to fetch the entire resource, parts of the resource, or just to view any metadata that may be provided.

Parameters:

executortyping.Optional[concurrent.futures.Executor]: Not used. Provided only to match the underlying interface.
head_onlybool: Defaults to False. If True, then the implementation may only retrieve HEAD information if supported. This currently only has any effect for web requests.

Returns:

AsyncReaderContextManager[WebReader]: An async context manager that when entered, produces the data stream.

Raises:

hikari.errors.BadRequestError: If a 400 is returned.
hikari.errors.UnauthorizedError: If a 401 is returned.
hikari.errors.ForbiddenError: If a 403 is returned.
hikari.errors.NotFoundError: If a 404 is returned.
hikari.errors.ClientHTTPResponseError: If any other 4xx is returned.
hikari.errors.InternalServerError: If any other 5xx is returned.
hikari.errors.HTTPResponseError: If any other unexpected response code is returned.

Examples

Downloading an entire resource at once into memory:

async with obj.stream() as stream:
    data = await stream.read()

Checking the metadata:

async with obj.stream() as stream:
    mimetype = stream.mimetype

if mimetype is None:
    ...
elif mimetype not in whitelisted_mimetypes:
    ...
else:
    ...

Fetching the data-uri of a resource:

async with obj.stream() as stream:
    data_uri = await stream.data_uri()

hikari.files.ensure_path(pathish)[source]#: Convert a path-like object to a pathlib.Path instance.

hikari.files.ensure_resource(url_or_resource, /)[source]#

Given a resource or string, convert it to a valid resource as needed.

Parameters:

url_or_resourceResourceish: The item to convert. If a Resource is passed, it is simply returned again. Anything else is converted to a Resource first.

Returns:

Resource: The resource to use.

hikari.files.unwrap_bytes(data)[source]#: Convert a byte-like object to bytes.

hikari.files.LazyByteIteratorish[source]#

Type hint representing an iterator/iterable of bytes.

This may be one of:

typing.AsyncIterator[bytes]
typing.AsyncIterable[bytes]
typing.Iterator[bytes]
typing.Iterable[bytes]
typing.AsyncIterator[str] (assuming UTF-8 encoding).
typing.AsyncIterable[str] (assuming UTF-8 encoding).
typing.Iterator[str] (assuming UTF-8 encoding).
typing.Iterable[str] (assuming UTF-8 encoding).
asyncio.StreamReader
aiohttp.StreamReader

hikari.files.Pathish[source]#

Type hint representing a literal file or path.

This may be one of:

str path.
os.PathLike derivative, such as pathlib.PurePath and pathlib.Path.

hikari.files.Rawish[source]#

Type hint representing valid raw data types.

This may be one of:

bytes
bytearray
memoryview
io.BytesIO
io.StringIO (assuming UTF-8 encoding).

hikari.files.Resourceish[source]#

Type hint representing a file or path to a file/URL/data URI.

This may be one of:

Resource or a derivative.
str path.
os.PathLike derivative, such as pathlib.PurePath and pathlib.Path.
bytes
bytearray
memoryview
io.BytesIO
io.StringIO (assuming UTF-8 encoding).