Memory Storage¶

In-memory storage backend for testing and development. Data is stored in a Python dictionary and lost when the process exits.

Warning

This backend is not suitable for production use. Data is not persisted and memory usage grows with stored files.

Configuration¶

class litestar_storages.backends.memory.MemoryConfig[source]¶

Bases: object

Configuration for in-memory storage.

Variables:: max_size – Maximum total bytes to store (None for unlimited)
Parameters:: max_size (int | None)

max_size: int | None = None¶

__init__(max_size=None)¶

Parameters:: max_size (int | None)

Storage Class¶

class litestar_storages.backends.memory.MemoryStorage[source]¶

Bases: BaseStorage

In-memory storage backend for testing and development.

This backend stores files in memory using a dictionary. It is not suitable for production use as data is lost on restart and consumes RAM.

Example

>>> storage = MemoryStorage()
>>> await storage.put("test.txt", b"hello world")
StoredFile(key='test.txt', size=11, ...)
>>> assert await storage.exists("test.txt")
>>> data = await storage.get_bytes("test.txt")
>>> assert data == b"hello world"

Note

The URL method returns memory:// URLs that are not accessible externally. This is primarily useful for testing.

Parameters:: config (MemoryConfig | None)

__init__(config=None)[source]¶

Initialize MemoryStorage.

Parameters:: config (MemoryConfig | None) – Configuration for the storage backend (optional)

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:

StorageError – If max_size would be exceeded

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 (single chunk for memory storage)
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 memory storage)

Returns:

//{key}

Return type:

URL string in the format memory

Note

Memory storage URLs are not accessible externally and are primarily useful for testing and development.

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:

StorageFileNotFoundError – 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:

StorageFileNotFoundError – 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 litestar_storages import MemoryStorage

# Create storage with default settings
storage = MemoryStorage()

# Store a file
result = await storage.put(
    "documents/readme.txt",
    b"Hello, World!",
    content_type="text/plain",
)
print(f"Stored {result.size} bytes")

# Retrieve the file
data = await storage.get_bytes("documents/readme.txt")
print(data.decode())  # "Hello, World!"

# Check existence
exists = await storage.exists("documents/readme.txt")
print(f"Exists: {exists}")  # True

# Delete the file
await storage.delete("documents/readme.txt")

With Size Limit¶

from litestar_storages import MemoryStorage, MemoryConfig
from litestar_storages.exceptions import StorageError

# Limit total storage to 1MB
storage = MemoryStorage(
    config=MemoryConfig(max_size=1024 * 1024)
)

try:
    # This will fail if total exceeds 1MB
    await storage.put("large-file.bin", large_data)
except StorageError as e:
    print(f"Storage full: {e}")

Testing Example¶

import pytest
from litestar_storages import MemoryStorage


@pytest.fixture
def storage():
    """Provide a fresh storage instance for each test."""
    return MemoryStorage()


async def test_upload_and_download(storage):
    # Upload
    result = await storage.put("test.txt", b"test content")
    assert result.key == "test.txt"
    assert result.size == 12

    # Download
    data = await storage.get_bytes("test.txt")
    assert data == b"test content"


async def test_file_not_found(storage):
    from litestar_storages.exceptions import StorageFileNotFoundError

    with pytest.raises(StorageFileNotFoundError):
        await storage.get_bytes("nonexistent.txt")

URL Generation¶

Memory storage generates memory:// URLs that are not externally accessible:

url = await storage.url("test.txt")
# Returns: "memory://test.txt"

This is primarily useful for maintaining API consistency in tests.