file_system

class litestar.file_system.BaseFileSystem

Bases: ABC

abstract async info(path: PathType, **kwargs: Any) → FileInfo

Return a FileInfo for the given path

abstract async read_bytes(path: PathType, start: int = 0, end: int = -1) → bytes

Read bytes from path

Parameters:

• path – File to read from

• start – Offset to start reading from

• end – Offset to stop reading at (inclusive)

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

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

Bases: LinkableFileSystem

Base class for a local file system.

async info(path: PathType, **kwargs: Any) → FileInfo

Return a FileInfo for the given path

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

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]

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

Return path with all symlinks resolved

class litestar.file_system.FileInfo

Bases: TypedDict

File information gathered from a file system.

is_symlink: bool | None

‘True’ if the file is a symbolic link. ‘None’ if the file system does not support symlinks

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

Bases: InitPlugin

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

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

Return the file system registered under name. If no matching file system is found, return None

register(name: str, fs: AnyFileSystem) → None

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

Bases: BaseFileSystem

__init__(fs: FsspecAsyncFileSystem) → None

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

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

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]

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

Bases: BaseFileSystem

__init__(fs: FsspecFileSystem) → None

Wrap a fsspec.spec.AbstractFileSystem to provide a litestar.file_system.BaseFileSystem compatible interface

async info(path: PathType, **kwargs: Any) → FileInfo

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

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]

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

Bases: BaseFileSystem, ABC

A file system that supports symlinks

abstract async resolve_symlinks(path: PathType) → str

Return path with all symlinks resolved

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

Register a file system as ‘linkable’, i.e. supporting symlinks, that does not itself implement LinkableFileSystem.

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

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

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

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