Skip to content

hikari.files#

Utilities and classes for interacting with files and web resources.

LazyByteIteratorish module-attribute #

Type hint representing an iterator/iterable of bytes.

This may be one of:

Pathish module-attribute #

Pathish = Union['os.PathLike[str]', str]

Type hint representing a literal file or path.

This may be one of:

Rawish module-attribute #

Type hint representing valid raw data types.

This may be one of:

Resourceish module-attribute #

Resourceish = Union['Resource[typing.Any]', Pathish, Rawish]

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

This may be one of:

AsyncReader #

Bases: AsyncIterable[bytes], 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 class-attribute instance-attribute #

filename: str = field(repr=True)

The filename of the resource.

mimetype class-attribute instance-attribute #

mimetype: Optional[str] = field(repr=True)

The mimetype of the resource. May be None if not known.

data_uri async #

data_uri() -> str

Fetch the data URI.

This reads the entire resource.

read async #

read() -> bytes

Read the rest of the resource and return it in a bytes object.

AsyncReaderContextManager #

Bases: ABC, Generic[ReaderImplT]

Context manager that returns a reader.

Bytes #

Bytes(data: Union[Rawish, LazyByteIteratorish], filename: str, /, mimetype: Optional[str] = None, *, spoiler: bool = False)

Bases: Resource[IteratorReader]

Representation of in-memory data to upload.

PARAMETER DESCRIPTION
data

The raw data.

TYPE: Union[Rawish, LazyByteIteratorish]

filename

The filename to use.

TYPE: str

mimetype

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.

TYPE: Optional[str] DEFAULT: None

spoiler

Whether to mark the file as a spoiler in Discord.

TYPE: bool DEFAULT: False

data instance-attribute #

The raw data/provider of raw data to upload.

filename property #

filename: str

Filename of the resource.

is_spoiler instance-attribute #

is_spoiler: bool = spoiler

Whether the file will be marked as a spoiler.

mimetype instance-attribute #

mimetype: Optional[str] = mimetype

The provided mimetype, if provided. Otherwise None.

url property #

url: str

URL of the resource.

from_data_uri staticmethod #

from_data_uri(data_uri: str, filename: Optional[str] = None) -> Bytes

Parse a given data URI.

PARAMETER DESCRIPTION
data_uri

The data URI to parse.

TYPE: str

filename

Filename to use. If this is not provided, then this is generated instead.

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Bytes

The parsed data URI as a hikari.files.Bytes object.

RAISES DESCRIPTION
ValueError

If the parsed argument is not a data URI.

save async #

save(path: Pathish, *, executor: Optional[Executor] = None, force: bool = False) -> None

Save this resource to disk.

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

PARAMETER DESCRIPTION
path

The path to save this resource to. If this is a string, the path will be relative to the current working directory.

TYPE: Pathish

executor

The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

TYPE: Optional[Executor] DEFAULT: None

force

Whether to overwrite an existing file.

TYPE: bool DEFAULT: False

stream #

stream(*, executor: Optional[Executor] = None, head_only: bool = False) -> AsyncReaderContextManager[IteratorReader]

Start streaming the content in chunks.

PARAMETER DESCRIPTION
executor

Not used. Provided only to match the underlying interface.

TYPE: Optional[Executor] DEFAULT: None

head_only

Not used. Provided only to match the underlying interface.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
AsyncReaderContextManager[IteratorReader]

An async context manager that when entered, produces the data stream.

File #

File(path: Pathish, /, filename: Optional[str] = None, *, spoiler: bool = False)

Bases: Resource[ThreadedFileReader]

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

PARAMETER DESCRIPTION
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.

TYPE: Pathish

filename

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

TYPE: Optional[str] DEFAULT: None

spoiler

Whether to mark the file as a spoiler in Discord.

TYPE: bool DEFAULT: False

filename property #

filename: str

Filename of the resource.

is_spoiler instance-attribute #

is_spoiler: bool = spoiler

Whether the file will be marked as a spoiler.

path instance-attribute #

The path to the file.

url property #

url: str

URL of the resource.

save async #

save(path: Pathish, *, executor: Optional[Executor] = None, force: bool = False) -> None

Save this resource to disk.

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

PARAMETER DESCRIPTION
path

The path to save this resource to. If this is a string, the path will be relative to the current working directory.

TYPE: Pathish

executor

The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

TYPE: Optional[Executor] DEFAULT: None

force

Whether to overwrite an existing file.

TYPE: bool DEFAULT: False

stream #

stream(*, executor: Optional[Executor] = None, head_only: bool = False) -> AsyncReaderContextManager[ThreadedFileReader]

Start streaming the resource using a thread pool executor.

PARAMETER DESCRIPTION
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.ThreadPoolExecutor is supported.

TYPE: Optional[Executor] DEFAULT: None

head_only

Not used. Provided only to match the underlying interface.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
AsyncReaderContextManager[ThreadedFileReader]

An async context manager that when entered, produces the data stream.

RAISES DESCRIPTION
IsADirectoryError

If the file's path leads to a directory.

FileNotFoundError

If the file doesn't exist.

IteratorReader #

Bases: AsyncReader

Asynchronous file reader that operates on in-memory data.

data class-attribute instance-attribute #

The data that will be yielded in chunks.

Resource #

Bases: Generic[ReaderImplT], 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.

extension property #

extension: Optional[str]

File extension, if there is one.

filename abstractmethod property #

filename: str

Filename of the resource.

url abstractmethod property #

url: str

URL of the resource.

read async #

read(*, executor: Optional[Executor] = None) -> bytes

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.

PARAMETER DESCRIPTION
executor

The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

TYPE: Optional[Executor] DEFAULT: None

RETURNS DESCRIPTION
bytes

The entire resource.

save async #

save(path: Pathish, *, executor: Optional[Executor] = None, force: bool = False) -> None

Save this resource to disk.

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

PARAMETER DESCRIPTION
path

The path to save this resource to. If this is a string, the path will be relative to the current working directory.

TYPE: Pathish

executor

The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

TYPE: Optional[Executor] DEFAULT: None

force

Whether to overwrite an existing file.

TYPE: bool DEFAULT: False

stream abstractmethod #

stream(*, executor: Optional[Executor] = None, head_only: bool = False) -> AsyncReaderContextManager[ReaderImplT]

Produce a stream of data for the resource.

PARAMETER DESCRIPTION
executor

The executor to run in for blocking operations. If None, then the default executor is used for the current event loop.

TYPE: Optional[Executor] DEFAULT: None

head_only

If True, then only the headers for the HTTP resource this object points to will be fetched without downloading the entire content, which can be significantly faster if you are scanning file types in messages, for example.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
AsyncReaderContextManager[AsyncReader]

An async iterable of bytes to stream.

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

ThreadedFileReader #

Bases: AsyncReader

Asynchronous file reader that reads a resource from local storage.

This implementation works with pools that exist in the same interpreter instance as the caller, namely thread pool executors, where objects do not need to be pickled to be communicated.

URL #

URL(url: str, filename: Optional[str] = None)

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.

PARAMETER DESCRIPTION
url

The URL of the resource.

TYPE: str

filename

The filename for the resource.

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

TYPE: Optional[str] DEFAULT: None

filename property #

filename: str

Filename of the resource.

url property #

url: str

URL of the resource.

WebReader #

Bases: AsyncReader

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

charset class-attribute instance-attribute #

charset: Optional[str] = field()

Optional character set information, if known.

head_only class-attribute instance-attribute #

head_only: bool = field()

If True, then only the HEAD was requested.

In this case, reading data off the object will return an empty byte string

reason class-attribute instance-attribute #

reason: str = field()

The HTTP response status reason.

size class-attribute instance-attribute #

size: Optional[int] = field()

The size of the resource, if known.

status class-attribute instance-attribute #

status: int = field()

The initial HTTP response status.

stream class-attribute instance-attribute #

stream: StreamReader = field(repr=False)

The aiohttp.StreamReader to read the content from.

url class-attribute instance-attribute #

url: str = field(repr=False)

The URL being read from.

read async #

read() -> bytes

Read the rest of the resource and return it in a bytes object.

WebResource #

Bases: Resource[WebReader], 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 hikari.files.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 #

stream(*, executor: Optional[Executor] = None, head_only: bool = False) -> AsyncReaderContextManager[WebReader]

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.

PARAMETER DESCRIPTION
executor

Not used. Provided only to match the underlying interface.

TYPE: Optional[Executor] DEFAULT: None

head_only

If True, then the implementation may only retrieve HEAD information if supported. This currently only has any effect for web requests.

TYPE: bool DEFAULT: False

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()
RETURNS DESCRIPTION
AsyncReaderContextManager[WebReader]

An async context manager that when entered, produces the data stream.

RAISES DESCRIPTION
BadRequestError

If a 400 is returned.

UnauthorizedError

If a 401 is returned.

ForbiddenError

If a 403 is returned.

NotFoundError

If a 404 is returned.

ClientHTTPResponseError

If any other 4xx is returned.

InternalServerError

If any other 5xx is returned.

HTTPResponseError

If any other unexpected response code is returned.

ensure_path #

ensure_path(pathish: Pathish) -> Path

Convert a path-like object to a pathlib.Path instance.

ensure_resource #

ensure_resource(url_or_resource: Resourceish) -> Resource[AsyncReader]

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

PARAMETER DESCRIPTION
url_or_resource

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

TYPE: Resourceish

RETURNS DESCRIPTION
Resource

The resource to use.

generate_filename_from_details #

generate_filename_from_details(*, mimetype: Optional[str] = None, extension: Optional[str] = None, data: Optional[bytes] = None) -> str

Given optional information about a resource, generate a filename.

PARAMETER DESCRIPTION
mimetype

The mimetype of the content, or None if not known.

TYPE: Optional[str] DEFAULT: None

extension

The file extension to use, or None if not known.

TYPE: Optional[str] DEFAULT: None

data

The data to inspect, or None if not known.

TYPE: Optional[bytes] DEFAULT: None

RETURNS DESCRIPTION
str

A generated quasi-unique filename.

guess_file_extension #

guess_file_extension(mimetype: str) -> Optional[str]

Guess the file extension for a given mimetype.

PARAMETER DESCRIPTION
mimetype

The mimetype to guess the extension for.

TYPE: str

Examples:

>>> guess_file_extension("image/png")
".png"
RETURNS DESCRIPTION
Optional[str]

The file extension, prepended with a .. If no match was found, return None.

guess_mimetype_from_data #

guess_mimetype_from_data(data: bytes) -> Optional[str]

Guess the mimetype of some data from the header.

Warning

This function only detects valid image headers that Discord allows the use of. Anything else will go undetected.

PARAMETER DESCRIPTION
data

The byte content to inspect.

TYPE: bytes

RETURNS DESCRIPTION
Optional[str]

The mimetype, if it was found. If the header is unrecognised, then None is returned.

guess_mimetype_from_filename #

guess_mimetype_from_filename(name: str) -> Optional[str]

Guess the mimetype of an object given a filename.

PARAMETER DESCRIPTION
name

The filename to inspect.

TYPE: str

RETURNS DESCRIPTION
Optional[str]

The closest guess to the given filename. May be None if no match was found.

to_data_uri #

to_data_uri(data: bytes, mimetype: Optional[str]) -> str

Convert the data and mimetype to a data URI.

PARAMETER DESCRIPTION
data

The data to encode as base64.

TYPE: bytes

mimetype

The mimetype, or None if we should attempt to guess it.

TYPE: Optional[str]

RETURNS DESCRIPTION
str

A data URI string.

unwrap_bytes #

unwrap_bytes(data: Rawish) -> bytes

Convert a byte-like object to bytes.