File Systems¶

When sending files with File or serving them using create_static_files_router(), in addition to doing so from the local file system Litestar also supports custom file systems, including support for fsspec, which enables an integration with a wide range of remote file systems.

At its core, a file system is any class that implements the abstract BaseFileSystem class. By default, Litestar ships with a single implementation: BaseLocalFileSystem.

Sending files from S3¶

import fsspec

from litestar import Litestar, get
from litestar.response import File

s3_fs = fsspec.filesystem("s3", asynchronous=True)

@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name, file_system=s3_fs)

app = Litestar([file_handler])

Supported file systems¶

In addition to file systems implemented on top of BaseFileSystem, all fsspec file systems, synchronous and asynchronous, are supported. They will be wrapped in FsspecSyncWrapper or FsspecAsyncWrapper respectively to conform to a common interface.

fsspec¶

Everywhere BaseFileSystem is accepted, fsspec file systems are usually accepted as well.

It can still make sense to set them up in the Registry instead, or wrap them manually with maybe_wrap_fsspec_file_system() once and use that wrapper, so they don’t have to be wrapped again for every usage:

Don’t

import fsspec

from litestar import Litestar, get
from litestar.response import File

s3_fs = fsspec.filesystem("s3", asynchronous=True)

@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name, file_system=s3_fs)

app = Litestar([file_handler])

Do instead

Explicitly wrapping an fsspec file system¶

import fsspec

from litestar import Litestar, get
from litestar.file_system import FsspecAsyncWrapper
from litestar.response import File

wrapped_s3_fs = FsspecAsyncWrapper(fsspec.filesystem("s3", asynchronous=True))


@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name, file_system=wrapped_s3_fs)


app = Litestar([file_handler])

Using the registry to automatically wrap an fsspec file system¶

import fsspec

from litestar import Litestar, get
from litestar.file_system import FileSystemRegistry
from litestar.response import File


@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name, file_system="s3")


app = Litestar(
    [file_handler],
    plugins=[
        FileSystemRegistry({"s3": fsspec.filesystem("s3", asynchronous=True)}),
    ],
)

Important

If an asynchronous fsspec file system is used, it should always be constructed with asynchronous=True without passing a loop, so Litestar can use their native async functions.

Adapting file systems that support symlinks¶

Handling symlinks can be tricky. To ensure Litestar always does the right thing, existing file systems that do support symlinks can be registered as “linkable” without having to implement LinkableFileSystem.

To register a file system as linkable, register_as_linkable() can be used to register a type and a function to resolve symlinks on this file system.

import os

from fsspec.implementations.local import LocalFileSystem

from litestar.file_system import AnyFileSystem, LinkableFileSystem

async def local_resolver(fs: AnyFileSystem, path: str) -> str:
    return os.path.realpath(path)

# Register LocalFileSystem as supporting symlinks
LinkableFileSystem.register_as_linkable(LocalFileSystem, local_resolver)

Tip

By default, fsspec.implementations.local.LocalFileSystem is already registered as a linkable file system

Registry¶

For easier configuration and testing, FileSystemRegistry can be used to register file systems under a name by which they can be referenced later.

Sending files from S3 by using the registry¶

import fsspec

from litestar import Litestar, get
from litestar.file_system import FileSystemRegistry
from litestar.response import File


@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name, file_system="assets")


def create_app(assets_fs: fsspec.AbstractFileSystem):
    return Litestar(
        [file_handler],
        plugins=[FileSystemRegistry({"assets": assets_fs})],
    )

Tip

When adding an fsspec file system, it will automatically be wrapped to be compatible with BaseFileSystem.

FileSystemRegistry is implemented as a plugin which is automatically set up unless an instance of it is passed to the application explicitly.

If needed, the registry can be accessed like any other plugin:

Accessing the registry¶

from litestar import Litestar, Request, get
from litestar.file_system import FileSystemRegistry


@get("/")
async def handler(request: Request) -> bytes:
    registry = request.app.plugins.get(FileSystemRegistry)
    return registry.default.read_bytes("some/file.txt")


app = Litestar([handler])

Setting a default file system¶

For actions that involve sending files, if no file system is passed explicitly, Litestar will use the default file system defined in the registry. This defaults to BaseLocalFileSystem, but may be configured to any supported file system.

Changing the default file system¶

import fsspec

from litestar import Litestar, get
from litestar.file_system import FileSystemRegistry
from litestar.response import File


@get("/{file_name:str}")
async def file_handler(file_name: str) -> File:
    return File(file_name)


app = Litestar(
    [file_handler],
    plugins=[
        FileSystemRegistry(default=fsspec.filesystem("s3", asynchronous=True)),
    ],
)