FileSystem Storage¶

Local filesystem storage backend using aiofiles for async I/O operations. Stores files in a configurable directory with support for URL generation via a base URL.

Note

Requires the aiofiles package. Install with: pip install litestar-storages[filesystem] or pip install aiofiles

Configuration¶

class litestar_storages.backends.filesystem.FileSystemConfig[source]¶

Bases: object

Configuration for filesystem storage.

Variables:

path – Base directory for file storage
base_url – Optional base URL for generating file URLs (e.g., “https://cdn.example.com/uploads”)
create_dirs – Automatically create directories as needed
permissions – File permissions (octal, default 0o644)

Parameters:

path (Path)
base_url (str | None)
create_dirs (bool)
permissions (int)

path: Path¶

base_url: str | None = None¶

create_dirs: bool = True¶

permissions: int = 420¶

__init__(path, base_url=None, create_dirs=True, permissions=420)¶

Parameters:

path (Path)
base_url (str | None)
create_dirs (bool)
permissions (int)

Storage Class¶

class litestar_storages.backends.filesystem.FileSystemStorage[source]¶

Bases: BaseStorage

Local filesystem storage backend.

Uses aiofiles for async file I/O operations. Stores files in a local directory with support for URL generation via base_url configuration.

Example

>>> storage = FileSystemStorage(
...     config=FileSystemConfig(
...         path=Path("/var/uploads"),
...         base_url="https://cdn.example.com/uploads",
...     )
... )
>>> await storage.put("images/photo.jpg", image_data)
>>> url = await storage.url("images/photo.jpg")
# Returns: https://cdn.example.com/uploads/images/photo.jpg

Security:: Path traversal prevention is implemented via _sanitize_key() to prevent access to files outside the configured base path.

Parameters:: config (FileSystemConfig)

Security

Path traversal attacks are prevented via the _sanitize_key() method, which normalizes paths and prevents access outside the configured base directory.

__init__(config)[source]¶

Initialize FileSystemStorage.

Parameters:: config (FileSystemConfig) – Configuration for the storage backend
Raises:: ConfigurationError – If the path is invalid

async put(key, data, *, content_type=None, metadata=None)[source]¶

Store data at the given key.

Parameters:

key (str) – Storage path/key for the file
data (bytes | AsyncIterator[bytes]) – File contents as bytes or async byte stream
content_type (str | None) – MIME type of the content
metadata (dict[str, str] | None) – Additional metadata to store with the file

Return type:

StoredFile

Returns:

StoredFile with metadata about the stored file

Raises:

StoragePermissionError – If unable to write the file

async get(key)[source]¶

Retrieve file contents as an async byte stream.

Parameters:: key (str) – Storage path/key for the file
Yields:: Chunks of file data as bytes
Raises:: StorageFileNotFoundError – If the file does not exist
Return type:: AsyncIterator[bytes]

async get_bytes(key)[source]¶

Retrieve entire file contents as bytes.

Parameters:: key (str) – Storage path/key for the file
Return type:: bytes
Returns:: Complete file contents as bytes
Raises:: StorageFileNotFoundError – If the file does not exist

async delete(key)[source]¶

Delete a file.

Parameters:: key (str) – Storage path/key for the file
Raises:: StorageFileNotFoundError – If the file does not exist
Return type:: None

async exists(key)[source]¶

Check if a file exists.

Parameters:: key (str) – Storage path/key for the file
Return type:: bool
Returns:: True if the file exists, False otherwise

async list(prefix='', *, limit=None)[source]¶

List files with optional prefix filter.

Parameters:

prefix (str) – Filter results to keys starting with this prefix
limit (int | None) – Maximum number of results to return

Yields:

StoredFile metadata for each matching file

Return type:

AsyncGenerator[StoredFile, None]

async url(key, *, expires_in=None)[source]¶

Generate a URL for accessing the file.

Parameters:

key (str) – Storage path/key for the file
expires_in (timedelta | None) – Optional expiration time (ignored for filesystem storage)

Return type:

str

Returns:

URL string for accessing the file

Note

If base_url is configured, returns base_url + key. Otherwise, returns file:// URL with absolute path.

async copy(source, destination)[source]¶

Copy a file within the storage backend.

Parameters:

source (str) – Source key to copy from
destination (str) – Destination key to copy to

Return type:

StoredFile

Returns:

StoredFile metadata for the new copy

Raises:

FileNotFoundError – If the source file does not exist

async move(source, destination)[source]¶

Move/rename a file within the storage backend.

Parameters:

source (str) – Source key to move from
destination (str) – Destination key to move to

Return type:

StoredFile

Returns:

StoredFile metadata for the moved file

Raises:

FileNotFoundError – If the source file does not exist

async info(key)[source]¶

Get metadata about a file without downloading it.

Parameters:: key (str) – Storage path/key for the file
Return type:: StoredFile
Returns:: StoredFile with metadata
Raises:: StorageFileNotFoundError – If the file does not exist

Usage Examples¶

Basic Usage¶

from pathlib import Path
from litestar_storages import FileSystemStorage, FileSystemConfig

storage = FileSystemStorage(
    config=FileSystemConfig(
        path=Path("/var/uploads"),
    )
)

# Store a file
result = await storage.put(
    "images/photo.jpg",
    image_bytes,
    content_type="image/jpeg",
)
# File saved to: /var/uploads/images/photo.jpg

# Get file info
info = await storage.info("images/photo.jpg")
print(f"Size: {info.size}, Modified: {info.last_modified}")

With CDN Base URL¶

When serving files through a CDN or web server:

from pathlib import Path
from litestar_storages import FileSystemStorage, FileSystemConfig

storage = FileSystemStorage(
    config=FileSystemConfig(
        path=Path("/var/www/uploads"),
        base_url="https://cdn.example.com/uploads",
    )
)

await storage.put("images/photo.jpg", image_bytes)

# Generate public URL
url = await storage.url("images/photo.jpg")
# Returns: "https://cdn.example.com/uploads/images/photo.jpg"

Custom Permissions¶

storage = FileSystemStorage(
    config=FileSystemConfig(
        path=Path("/var/uploads"),
        permissions=0o600,  # Owner read/write only
    )
)

Listing Files¶

# List all files
async for file in storage.list():
    print(f"{file.key}: {file.size} bytes")

# List files with prefix
async for file in storage.list("images/2024/"):
    print(file.key)

# List with limit
async for file in storage.list(limit=10):
    print(file.key)

Streaming Large Files¶

# Stream download
async for chunk in storage.get("large-video.mp4"):
    await response.write(chunk)

# Stream upload from async generator
async def read_chunks():
    async with aiofiles.open("/path/to/source", "rb") as f:
        while chunk := await f.read(64 * 1024):
            yield chunk

await storage.put("large-file.bin", read_chunks())

Path Traversal Protection¶

The storage backend sanitizes all keys to prevent directory traversal attacks:

# These dangerous inputs are sanitized:
storage._sanitize_key("../../../etc/passwd")  # Returns: "etc/passwd"
storage._sanitize_key("/absolute/path")       # Returns: "absolute/path"
storage._sanitize_key("..\\..\\windows")      # Returns: "windows"

All operations are restricted to files within the configured base path.