file_system¶

class litestar.file_system.BaseFileSystem[source]¶

Bases: ABC

abstractmethod async info(path: PathType, **kwargs: Any) → FileInfo[source]¶: Return a FileInfo for the given path

abstractmethod async read_bytes(path: PathType, start: int = 0, end: int = -1) → bytes[source]¶

Read bytes from path

Parameters:

path¶ – File to read from
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

abstractmethod iter(path: PathType, chunksize: int, start: int = 0, end: int = -1) → AsyncGenerator[bytes, None][source]¶

Stream bytes from path

Parameters:

path¶ – Path to read from
chunksize¶ – Number of bytes to read per chunk
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

class litestar.file_system.BaseLocalFileSystem[source]¶

Bases: LinkableFileSystem

Base class for a local file system.

async info(path: PathType, **kwargs: Any) → FileInfo[source]¶: Return a FileInfo for the given path

async read_bytes(path: PathType, start: int = 0, end: int = -1) → bytes[source]¶

Read bytes from path

Parameters:

path¶ – File to read from
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

async iter(path: PathType, chunksize: int, start: int = 0, end: int = -1) → AsyncGenerator[bytes, None][source]¶

Stream bytes from path

Parameters:

path¶ – Path to read from
chunksize¶ – Number of bytes to read per chunk
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

async resolve_symlinks(path: PathType) → str[source]¶: Return path with all symlinks resolved

class litestar.file_system.FileInfo[source]¶

Bases: TypedDict

File information gathered from a file system.

is_symlink: NotRequired[bool | None]¶: ‘True’ if the file is a symbolic link. ‘None’ if the file system does not support symlinks

mtime: NotRequired[float]¶: Modified time stamp.

name: str¶: The path of the file.

size: int¶: Total size, in bytes.

type: str¶: The type of the file system object.

class litestar.file_system.FileSystemRegistry[source]¶

Bases: InitPlugin

__init__(file_systems: Mapping[str, AnyFileSystem] | None = None, default: AnyFileSystem | None = None) → None[source]¶

File system registry

Parameters:

file_systems¶ – A mapping of names to file systems
default¶ – Default file system to use when no system is specified. If None, default to BaseLocalFileSystem

property default: BaseFileSystem¶: Return the default file system

get(name: str) → BaseFileSystem | None[source]¶: Return the file system registered under name. If no matching file system is found, return None

register(name: str, fs: AnyFileSystem) → None[source]¶: Register a file system fs under name. If a file system was previously registered under this name, it will be overwritten

class litestar.file_system.FsspecAsyncWrapper[source]¶

Bases: BaseFileSystem

__init__(fs: FsspecAsyncFileSystem) → None[source]¶

Wrap a fsspec.asyn.AsyncFileSystem to provide a class:litestar.file_system.BaseFileSystem compatible interface

Parameters:: fs¶ – An fsspec.asyn.AsyncFileSystem instantiated with ` asynchronous=True`

async info(path: PathType, **kwargs: Any) → FileInfo[source]¶

Return a FileInfo for the given path.

Return info verbatim from fsspec.async.AsyncFileSystem._info, but try to patch ‘mtime’ using litestar.file_system.get_fsspec_mtime_equivalent().

async read_bytes(path: PathType, start: int = 0, end: int = -1) → bytes[source]¶

Read bytes from path using fsspec.asyn.AsyncFileSystem._cat_file

Parameters:

path¶ – File to read from
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

async iter(path: PathType, chunksize: int, start: int = 0, end: int = -1) → AsyncGenerator[bytes, None][source]¶

Stream bytes from path.

If no offsets are given and the file system implements fsspec.async.AsyncFileSystem.async_open, use async_open, otherwise read chunks manually using fsspec.async.AsyncFileSystem._cat_file

Parameters:

path¶ – Path to read from
chunksize¶ – Number of bytes to read per chunk
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

class litestar.file_system.FsspecSyncWrapper[source]¶

Bases: BaseFileSystem

__init__(fs: FsspecFileSystem) → None[source]¶: Wrap a fsspec.spec.AbstractFileSystem to provide a litestar.file_system.BaseFileSystem compatible interface

async info(path: PathType, **kwargs: Any) → FileInfo[source]¶

Return a FileInfo for the given path.

Return info verbatim from fsspec.spec.AbstractFileSystem.info(), but try to patch ‘mtime’ using litestar.file_system.get_fsspec_mtime_equivalent().

async read_bytes(path: PathType, start: int = 0, end: int = -1) → bytes[source]¶

Read bytes from path using fsspec.spec.AbstractFileSystem.cat_file()

Parameters:

path¶ – File to read from
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

async iter(path: PathType, chunksize: int, start: int = 0, end: int = -1) → AsyncGenerator[bytes, None][source]¶

Stream bytes from path using fsspec.spec.AbstractFileSystem.open()

Parameters:

path¶ – Path to read from
chunksize¶ – Number of bytes to read per chunk
start¶ – Offset to start reading from
end¶ – Offset to stop reading at (inclusive)

class litestar.file_system.LinkableFileSystem[source]¶

Bases: BaseFileSystem, ABC

A file system that supports symlinks

abstractmethod async resolve_symlinks(path: PathType) → str[source]¶: Return path with all symlinks resolved

classmethod register_as_linkable(fs_type: type[AnyFileSystem], resolver: SymlinkResolver) → None[source]¶

By default, fsspec.implementations.local.LocalFileSystem is registered.

Parameters:

fs_type¶ – The file system type
resolver¶ – A function that takes in a path and returns an absolute path with all symlinks resolved

classmethod get_symlink_resolver(fs: AnyFileSystem) → SymlinkResolver | None[source]¶

For a given file system, get a function to resolve potential symlinks in a path.

For classes implementing LinkableFileSystem, their resolve_symlinks() is returned. Otherwise, the resolver registered via register_as_linkable will be returned.

litestar.file_system.get_fsspec_mtime_equivalent(info: dict[str, Any]) → float | None[source]¶

Return the ‘mtime’ or equivalent for different fsspec implementations, since they are not standardized.

See https://github.com/fsspec/filesystem_spec/issues/526.

litestar.file_system.maybe_wrap_fsspec_file_system(file_system: AnyFileSystem) → BaseFileSystem[source]¶: If file_system is an fsspec file system, wrap it in FsspecSyncWrapper or FsspecAsyncWrapper depending on its type. If fsspec is not installed or file_system is not an fsspec file system, return file_system