hikari.files

Utilities and classes for interacting with files and web resources.

LazyByteIteratorish module-attribute

LazyByteIteratorish = Union[
    AsyncIterator[bytes],
    AsyncIterable[bytes],
    Iterator[bytes],
    Iterable[bytes],
    AsyncIterator[str],
    AsyncIterable[str],
    Iterator[str],
    Iterable[str],
    AsyncGenerator[bytes, Any],
    Generator[bytes, Any, Any],
    AsyncGenerator[str, Any],
    Generator[str, Any, Any],
    StreamReader,
    StreamReader,
]

Type hint representing an iterator/iterable of bytes.

This may be one of:

• typing.AsyncIterator[bytes]
• typing.AsyncIterable[bytes]
• typing.Iterator[bytes]
• typing.Iterator[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

Pathish module-attribute

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

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.

Rawish module-attribute

Rawish = Union[bytes, bytearray, memoryview, BytesIO, StringIO]

Type hint representing valid raw data types.

This may be one of:

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

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:

• hikari.files.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).

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

data: Union[bytes, LazyByteIteratorish] = data

The raw data/provider of raw data to upload.

extension property

extension: Optional[str]

File extension, if there is one.

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.

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

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: Union[str, PathLike, Path]
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

extension property

extension: Optional[str]

File extension, if there is one.

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

path: Path = ensure_path(path)

The path to the file.

url 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

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

data: Union[bytes, LazyByteIteratorish] = field()

The data that will be yielded in chunks.

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.

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.

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.

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

extension property

extension: Optional[str]

File extension, if there is one.

filename property

filename: str

Filename of the resource.

url 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

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.

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.

filename class-attribute instance-attribute

filename: str = field(repr=True)

The filename of the resource.

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

mimetype class-attribute instance-attribute

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

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

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.

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.

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.

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

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: bytes

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.