Storage Engines

All storage backends extend StorageEngine. Engines are instances configured once at construction and shared across requests.

Base Class

class StorageEngine(ABC):
    name: ClassVar[str] = "storage"

    @abstractmethod
    async def save(
        self,
        *,
        file: UploadFile,
        filename: str,          # sanitized, destination-relative (POSIX style)
        destination: str | None,  # resolved directory / key prefix
        config: StoreConfig,      # merged field configuration
        field_name: str = "",
    ) -> FileData: ...

    async def delete(self, key: str) -> None: ...                     # NotSupportedError by default
    async def presign_upload(self, key, *, expires_in=3600,
                             content_type=None) -> PresignedURL: ...  # NotSupportedError by default
    async def presign_download(self, key, *, expires_in=3600) -> PresignedURL: ...

Built-in Engines

Engine	name	Module	Presign	Delete
LocalEngine	"local"	filestore.engines.local	✗	✓
MemoryEngine	"memory"	filestore.engines.memory	✗	✗
S3Engine	"s3"	filestore.engines.s3	✓	✓
GCSEngine	"gcs"	filestore.engines.gcs	✓	✓
AzureBlobEngine	"azure"	filestore.engines.azure	✓	✓

All engines are importable from the top-level package; cloud engines load their SDKs lazily so the base install never imports them.

Constructor arguments are documented in the Configuration Reference; behaviour is covered in the Storage Engines guide.

PresignedURL

Returned by presign_upload() and presign_download(). A Pydantic model:

Field	Type	Description
url	str	The signed URL
method	str	HTTP method the client must use ("PUT" or "GET")
headers	dict[str, str]	Headers that must be sent with the request
fields	dict[str, str]	Extra form fields (reserved for POST-policy uploads)
key	str	The object key the URL refers to
expires_in	int	Validity window in seconds
storage	str	Backend that produced the URL

See the Presigned URLs guide for usage patterns.

Custom Engines

Create your own storage backend by implementing save():

from filestore import FileData, StorageEngine, StoreConfig
from starlette.datastructures import UploadFile


class RedisEngine(StorageEngine):
    name = "redis"

    def __init__(self, redis):
        self.redis = redis

    async def save(
        self,
        *,
        file: UploadFile,
        filename: str,
        destination: str | None,
        config: StoreConfig,
        field_name: str = "",
    ) -> FileData:
        data = await file.read()
        self.validate_size_limits(
            size=len(data), config=config, field_name=field_name, filename=filename
        )
        key = f"uploads:{destination or ''}:{filename}"
        await self.redis.set(key, data)

        return FileData(
            filename=filename,
            content_type=file.content_type,
            size=len(data),
            key=key,
            url=f"redis://{key}",
            storage=self.name,
        )

    async def delete(self, key: str) -> None:
        await self.redis.delete(key)

Use it:

storage = FileStore("file", engine=RedisEngine(redis))

Notes for engine authors:

• Raise ValidationError / StorageError on failure — FileStore converts them into failed FileData results; anything else is caught and logged too.
• FileStore sets field_name, original_filename, and merges callback metadata onto your returned FileData — you only fill in what the backend knows.
• The uploaded file is closed by FileStore after save() returns; don't close it yourself.

Helper Methods

All engines inherit these from StorageEngine:

Method	Description
resolve_size(file)	Reported size, falling back to measuring the stream
detect_stream_size(file_obj)	Measure a seekable stream without consuming it
validate_size_limits(...)	Check a byte count against min/max config, raising ValidationError